diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java index ec67e5f451bd..0e9216a5c982 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java @@ -372,6 +372,19 @@ public CommonConfig setWalBufferSize(int walBufferSize) { return this; } + @Override + public CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked) { + setProperty("check_period_when_insert_blocked", String.valueOf(checkPeriodWhenInsertBlocked)); + return this; + } + + @Override + public CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked) { + setProperty( + "max_waiting_time_when_insert_blocked", String.valueOf(maxWaitingTimeWhenInsertBlocked)); + return this; + } + @Override public CommonConfig setDegreeOfParallelism(int degreeOfParallelism) { setProperty("degree_of_query_parallelism", String.valueOf(degreeOfParallelism)); diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java index 36d06aefe889..cff20f9cff87 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java @@ -373,6 +373,20 @@ public CommonConfig setWalBufferSize(int walBufferSize) { return this; } + @Override + public CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked) { + cnConfig.setCheckPeriodWhenInsertBlocked(checkPeriodWhenInsertBlocked); + dnConfig.setCheckPeriodWhenInsertBlocked(checkPeriodWhenInsertBlocked); + return this; + } + + @Override + public CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked) { + cnConfig.setMaxWaitingTimeWhenInsertBlocked(maxWaitingTimeWhenInsertBlocked); + dnConfig.setMaxWaitingTimeWhenInsertBlocked(maxWaitingTimeWhenInsertBlocked); + return this; + } + @Override public CommonConfig setDegreeOfParallelism(int degreeOfParallelism) { cnConfig.setDegreeOfParallelism(degreeOfParallelism); diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java index 752dcd009db0..f40d422892a2 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java @@ -263,6 +263,16 @@ public CommonConfig setWalBufferSize(int walBufferSize) { return this; } + @Override + public CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked) { + return this; + } + + @Override + public CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked) { + return this; + } + @Override public CommonConfig setDegreeOfParallelism(int degreeOfParallelism) { return this; diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java index 0ad3c23af16f..b52429d2155b 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java @@ -119,6 +119,10 @@ CommonConfig setEnableAutoLeaderBalanceForRatisConsensus( CommonConfig setWalBufferSize(int walBufferSize); + CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked); + + CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked); + CommonConfig setDegreeOfParallelism(int degreeOfParallelism); CommonConfig setDataRatisTriggerSnapshotThreshold(long threshold); diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java new file mode 100644 index 000000000000..5158cd1632e1 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java @@ -0,0 +1,335 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.it.load; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; +import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; +import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.cluster.RegionRoleType; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo; +import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; +import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +@RunWith(IoTDBTestRunner.class) +@Category({ClusterIT.class}) +public class IoTDBRegionGroupLeaderBalanceWithWALBlockIT { + + private static final String TEST_SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS = + ConsensusFactory.RATIS_CONSENSUS; + private static final String TEST_DATA_REGION_CONSENSUS_PROTOCOL_CLASS = + ConsensusFactory.IOT_CONSENSUS; + private static final int TEST_REPLICATION_FACTOR = 3; + private static final int TEST_DATA_NODE_NUM = 3; + private static final int DATABASE_NUM = 3; + private static final int RETRY_NUM = 60; + private static final int TEST_SERIES_PARTITION_SLOT = 0; + private static final long TEST_TIME_PARTITION_SLOT = 0; + private static final int WAL_FILE_SIZE_THRESHOLD_IN_BYTE = 1024; + private static final int WAL_PAYLOAD_REPEAT_COUNT = 128; + + private static final String DATABASE = "root.wal_block_db"; + private static final String WAL_THROTTLE_THRESHOLD_IN_BYTE = "wal_throttle_threshold_in_byte"; + private static final String WAL_FILE_SIZE_THRESHOLD_IN_BYTE_CONFIG = + "wal_file_size_threshold_in_byte"; + private static final String WAL_BLOCKED_STATUS = + NodeStatus.ReadOnly.getStatus() + "(" + NodeStatus.WAL_BLOCKED + ")"; + + @Before + public void setUp() { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setEnableAutoLeaderBalanceForRatisConsensus(true) + .setEnableAutoLeaderBalanceForIoTConsensus(true) + .setSchemaRegionConsensusProtocolClass(TEST_SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS) + .setDataRegionConsensusProtocolClass(TEST_DATA_REGION_CONSENSUS_PROTOCOL_CLASS) + .setSchemaReplicationFactor(TEST_REPLICATION_FACTOR) + .setDataReplicationFactor(TEST_REPLICATION_FACTOR) + .setSeriesSlotNum(1) + .setCheckPeriodWhenInsertBlocked(50) + .setMaxWaitingTimeWhenInsertBlocked(2000); + EnvFactory.getEnv().initClusterEnvironment(1, TEST_DATA_NODE_NUM); + } + + @After + public void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testRegionLeaderBalanceWhenWalLongTermBlocked() throws Exception { + try (SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { + createDataRegionGroups(client); + waitUntil( + "all DataNodes have balanced DataRegion leaders", + () -> isLeaderDistributionBalanced(client)); + + TRegionInfo targetLeader = findAnyDataRegionLeader(client); + long walDiskUsage = generateWalTraffic(client, targetLeader); + triggerLongTermWalBlockingOnDataNode(client, targetLeader.getDataNodeId(), walDiskUsage); + + waitUntil( + "target leader DataNode becomes ReadOnly because of long-term WAL blocking", + () -> + WAL_BLOCKED_STATUS.equals( + getNodeStatusWithReason(client, targetLeader.getDataNodeId()))); + waitUntil( + "Region leaders are moved away from ReadOnly DataNodes", + () -> hasNoLeaderOnReadOnlyDataNode(client, targetLeader.getDataNodeId())); + } + } + + private void createDataRegionGroups(SyncConfigNodeIServiceClient client) throws Exception { + for (int i = 0; i < DATABASE_NUM; i++) { + TSStatus status = client.setDatabase(new TDatabaseSchema(DATABASE + i)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + + Map seriesSlotMap = new HashMap<>(); + seriesSlotMap.put( + new TSeriesPartitionSlot(TEST_SERIES_PARTITION_SLOT), + new TTimeSlotList() + .setTimePartitionSlots( + Collections.singletonList(new TTimePartitionSlot(TEST_TIME_PARTITION_SLOT)))); + Map> databaseSlotsMap = new HashMap<>(); + databaseSlotsMap.put(DATABASE + i, seriesSlotMap); + + TDataPartitionTableResp dataPartitionTableResp = + client.getOrCreateDataPartitionTable(new TDataPartitionReq(databaseSlotsMap)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + dataPartitionTableResp.getStatus().getCode()); + } + } + + private boolean isLeaderDistributionBalanced(SyncConfigNodeIServiceClient client) + throws Exception { + Map leaderCounter = new HashMap<>(); + for (TRegionInfo regionInfo : getUserDataRegionInfoList(client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + leaderCounter.merge(regionInfo.getDataNodeId(), 1, Integer::sum); + } + } + if (leaderCounter.size() != TEST_DATA_NODE_NUM) { + return false; + } + for (Integer leaderCount : leaderCounter.values()) { + if (leaderCount != DATABASE_NUM / TEST_DATA_NODE_NUM) { + return false; + } + } + return true; + } + + private TRegionInfo findAnyDataRegionLeader(SyncConfigNodeIServiceClient client) + throws Exception { + for (TRegionInfo regionInfo : getUserDataRegionInfoList(client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + return regionInfo; + } + } + throw new AssertionError("DataRegion leader not found"); + } + + private void triggerLongTermWalBlockingOnDataNode( + SyncConfigNodeIServiceClient client, int dataNodeId, long walDiskUsage) throws Exception { + Assert.assertTrue("No WAL traffic was generated on target DataNode", walDiskUsage > 0); + + Map configItems = new HashMap<>(); + configItems.put(WAL_THROTTLE_THRESHOLD_IN_BYTE, Long.toString(walDiskUsage)); + TSStatus status = client.setConfiguration(new TSetConfigurationReq(configItems, dataNodeId)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + } + + private long generateWalTraffic(SyncConfigNodeIServiceClient client, TRegionInfo targetLeader) + throws Exception { + int dataNodeId = targetLeader.getDataNodeId(); + long originalWalDiskUsage = countWalDiskUsage(dataNodeId); + Map configItems = new HashMap<>(); + configItems.put( + WAL_FILE_SIZE_THRESHOLD_IN_BYTE_CONFIG, Integer.toString(WAL_FILE_SIZE_THRESHOLD_IN_BYTE)); + TSStatus status = client.setConfiguration(new TSetConfigurationReq(configItems, dataNodeId)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + + DataNodeWrapper dataNodeWrapper = + EnvFactory.getEnv() + .dataNodeIdToWrapper(dataNodeId) + .orElseThrow(() -> new AssertionError("DataNode not found: " + dataNodeId)); + + try (Connection connection = + EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); + Statement statement = connection.createStatement()) { + String payload = String.join("", Collections.nCopies(WAL_PAYLOAD_REPEAT_COUNT, "wal_block")); + String device = + targetLeader.getDatabase() + ".d" + targetLeader.getConsensusGroupId().getId(); + statement.execute("CREATE TIMESERIES " + device + ".s WITH DATATYPE=TEXT, ENCODING=PLAIN"); + for (int i = 0; i < 16; i++) { + statement.execute( + "INSERT INTO " + + device + + "(time,s) VALUES(" + + (TEST_TIME_PARTITION_SLOT + i) + + ", '" + + payload + + "')"); + } + } + + final long[] currentWalDiskUsage = new long[1]; + waitUntil( + "target DataNode generates WAL files", + () -> { + currentWalDiskUsage[0] = countWalDiskUsage(dataNodeId); + return currentWalDiskUsage[0] > originalWalDiskUsage; + }); + return currentWalDiskUsage[0]; + } + + private long countWalDiskUsage(int dataNodeId) throws IOException { + DataNodeWrapper dataNodeWrapper = + EnvFactory.getEnv() + .dataNodeIdToWrapper(dataNodeId) + .orElseThrow(() -> new AssertionError("DataNode not found: " + dataNodeId)); + Path walDir = new File(dataNodeWrapper.getWalDir()).toPath(); + if (!Files.exists(walDir)) { + return 0; + } + try (Stream paths = Files.walk(walDir)) { + Map> walFilesByDir = + paths + .filter(Files::isRegularFile) + .filter( + path -> + WALFileUtils.walFilenameFilter( + path.getParent().toFile(), path.getFileName().toString())) + .collect(Collectors.groupingBy(Path::getParent)); + long walDiskUsage = 0; + for (List walFiles : walFilesByDir.values()) { + if (walFiles.size() <= 1) { + continue; + } + Path currentWalFile = + Collections.max( + walFiles, + Comparator.comparingLong( + path -> WALFileUtils.parseVersionId(path.getFileName().toString()))); + for (Path walFile : walFiles) { + if (!walFile.equals(currentWalFile)) { + walDiskUsage += walFile.toFile().length(); + } + } + } + return walDiskUsage; + } + } + + private String getNodeStatusWithReason(SyncConfigNodeIServiceClient client, int dataNodeId) + throws Exception { + TShowClusterResp showClusterResp = client.showCluster(); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), showClusterResp.getStatus().getCode()); + return showClusterResp.getNodeStatus().get(dataNodeId); + } + + private boolean hasNoLeaderOnReadOnlyDataNode( + SyncConfigNodeIServiceClient client, int readOnlyDataNodeId) throws Exception { + for (TRegionInfo regionInfo : getUserDataRegionInfoList(client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + if (regionInfo.getDataNodeId() == readOnlyDataNodeId + || NodeStatus.ReadOnly.getStatus().equals(regionInfo.getStatus())) { + return false; + } + } + } + return true; + } + + private List getUserDataRegionInfoList(SyncConfigNodeIServiceClient client) + throws Exception { + TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode()); + + List result = new ArrayList<>(); + for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) { + if (TConsensusGroupType.DataRegion.equals(regionInfo.getConsensusGroupId().getType()) + && !regionInfo.getDatabase().startsWith(SystemConstant.SYSTEM_DATABASE) + && !regionInfo.getDatabase().startsWith(SystemConstant.AUDIT_DATABASE)) { + result.add(regionInfo); + } + } + return result; + } + + private void waitUntil(String condition, WaitCondition waitCondition) throws Exception { + for (int retry = 0; retry < RETRY_NUM; retry++) { + if (waitCondition.evaluate()) { + return; + } + TimeUnit.SECONDS.sleep(1); + } + Assert.fail("Failed to wait until " + condition); + } + + @FunctionalInterface + private interface WaitCondition { + boolean evaluate() throws Exception; + } +} diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 10ea4763af9a..a0541ca7c465 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -239,6 +239,8 @@ public final class ManagerMessages { "Heartbeat service is started successfully."; public static final String HEARTBEAT_SERVICE_IS_STOPPED_SUCCESSFULLY = "Heartbeat service is stopped successfully."; + public static final String RECONFIGURE_PEER_PRIORITIES_FAILED = + "Failed to reconfigure ConfigNode peer priorities to {}."; public static final String INCORRECT_VERSION_OF = "Incorrect version of "; public static final String INIT_CONSENSUSMANAGER_SUCCESSFULLY_WHEN_RESTARTED = "Init ConsensusManager successfully when restarted"; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 499e922f7a7d..c7353f9c0f4d 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -237,6 +237,8 @@ public final class ManagerMessages { "心跳服务已成功启动。"; public static final String HEARTBEAT_SERVICE_IS_STOPPED_SUCCESSFULLY = "心跳服务已成功停止。"; + public static final String RECONFIGURE_PEER_PRIORITIES_FAILED = + "重新配置 ConfigNode 节点优先级 {} 失败。"; public static final String INCORRECT_VERSION_OF = "版本不正确："; public static final String INIT_CONSENSUSMANAGER_SUCCESSFULLY_WHEN_RESTARTED = "重启时成功初始化 ConsensusManager"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java index f49525ea7da7..944463e697af 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java @@ -41,6 +41,7 @@ import java.io.File; import java.lang.reflect.Field; import java.util.Arrays; +import java.util.List; public class ConfigNodeConfig { @@ -523,6 +524,15 @@ public void setConsensusDir(String consensusDir) { this.consensusDir = consensusDir; } + /** + * Directories whose loss would render this ConfigNode unable to serve. Used by the periodic + * disk-health check on both leader (in HeartbeatService loop) and followers (in the + * heartbeat-receive path). + */ + public List getCriticalDirs() { + return Arrays.asList(systemDir, consensusDir); + } + public String getConfigNodeConsensusProtocolClass() { return configNodeConsensusProtocolClass; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java index 3416246811ed..dcd12145dad3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java @@ -547,6 +547,16 @@ public String getNodeStatusWithReason(int nodeId) { .orElseGet(() -> NodeStatus.Unknown.getStatus() + "(NoHeartbeat)"); } + /** + * @return The raw {@code statusReason} string for {@code nodeId}, or {@code null} when no cache + * entry exists yet or no reason has been reported. + */ + public String getNodeStatusReason(int nodeId) { + return Optional.ofNullable(nodeCacheMap.get(nodeId)) + .map(BaseNodeCache::getStatusReason) + .orElse(null); + } + /** * Get all Node's current status with reason. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java index 4eeb96344ccd..434d5bbf3fe2 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java @@ -66,4 +66,11 @@ public String getNodeStatusWithReason() { ? statistics.getStatus().getStatus() : statistics.getStatus().getStatus() + "(" + statistics.getStatusReason() + ")"; } + + /** + * @return The raw reason string (may be {@code null}) accompanying the current NodeStatus. + */ + public String getStatusReason() { + return ((NodeStatistics) currentStatistics.get()).getStatusReason(); + } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java index 4d675e1e8a44..7228fca648cf 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.manager.load.cache.node; import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.manager.load.cache.AbstractHeartbeatSample; @@ -49,28 +50,39 @@ public ConfigNodeHeartbeatCache(int configNodeId, NodeStatistics statistics) { @Override public synchronized void updateCurrentStatistics(boolean forceUpdate) { - // Skip itself and the Removing status can not be updated - if (nodeId == CURRENT_NODE_ID || NodeStatus.Removing.equals(getNodeStatus())) { + // Removing status can not be updated + if (NodeStatus.Removing.equals(getNodeStatus())) { return; } - NodeHeartbeatSample lastSample; - // Update Node status - NodeStatus status; long currentNanoTime = System.nanoTime(); - final List heartbeatHistory; - synchronized (slidingWindow) { - lastSample = (NodeHeartbeatSample) getLastSample(); - heartbeatHistory = Collections.unmodifiableList(slidingWindow); + NodeStatus status; + String statusReason; + + if (nodeId == CURRENT_NODE_ID) { + // Self entry: heartbeat loop never sends to itself, so mirror the status that + // this ConfigNode's local disk-check / startup wrote into CommonConfig. + status = CommonDescriptor.getInstance().getConfig().getNodeStatus(); + statusReason = CommonDescriptor.getInstance().getConfig().getStatusReason(); + } else { + NodeHeartbeatSample lastSample; + final List heartbeatHistory; + synchronized (slidingWindow) { + lastSample = (NodeHeartbeatSample) getLastSample(); + heartbeatHistory = Collections.unmodifiableList(slidingWindow); - if (lastSample == null) { - /* First heartbeat not received from this ConfigNode, status is UNKNOWN */ - status = NodeStatus.Unknown; - } else if (!failureDetector.isAvailable(nodeId, heartbeatHistory)) { - /* Failure detector decides that this ConfigNode is UNKNOWN */ - status = NodeStatus.Unknown; - } else { - status = lastSample.getStatus(); + if (lastSample == null) { + /* First heartbeat not received from this ConfigNode, status is UNKNOWN */ + status = NodeStatus.Unknown; + statusReason = null; + } else if (!failureDetector.isAvailable(nodeId, heartbeatHistory)) { + /* Failure detector decides that this ConfigNode is UNKNOWN */ + status = NodeStatus.Unknown; + statusReason = null; + } else { + status = lastSample.getStatus(); + statusReason = lastSample.getStatusReason(); + } } } @@ -79,6 +91,6 @@ public synchronized void updateCurrentStatistics(boolean forceUpdate) { // TODO: Construct load score module long loadScore = NodeStatus.isNormalStatus(status) ? 0 : Long.MAX_VALUE; - currentStatistics.set(new NodeStatistics(currentNanoTime, status, null, loadScore)); + currentStatistics.set(new NodeStatistics(currentNanoTime, status, statusReason, loadScore)); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java index 8217593f5d67..5ead20d291f4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java @@ -74,8 +74,13 @@ public NodeHeartbeatSample(TAIHeartbeatResp heartbeatResp) { /** Constructor for ConfigNode sample. */ public NodeHeartbeatSample(TConfigNodeHeartbeatResp heartbeatResp) { super(heartbeatResp.getTimestamp()); - this.status = NodeStatus.Running; - this.statusReason = null; + // Old ConfigNodes don't populate status/statusReason — fall back to Running/null + // so a rolling upgrade leaves the leader's view of legacy peers unchanged. + this.status = + heartbeatResp.isSetStatus() + ? NodeStatus.parse(heartbeatResp.getStatus()) + : NodeStatus.Running; + this.statusReason = heartbeatResp.isSetStatusReason() ? heartbeatResp.getStatusReason() : null; this.loadSample = null; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java index ef732145f22d..2c07212f10cd 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java @@ -24,9 +24,12 @@ import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation; import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.cluster.DiskChecker; +import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.confignode.client.async.AsyncAINodeHeartbeatClientPool; import org.apache.iotdb.confignode.client.async.AsyncConfigNodeHeartbeatClientPool; @@ -43,6 +46,7 @@ import org.apache.iotdb.confignode.manager.load.cache.node.ConfigNodeHeartbeatCache; import org.apache.iotdb.confignode.manager.node.NodeManager; import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatReq; +import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq; @@ -50,7 +54,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.concurrent.Future; @@ -150,10 +156,57 @@ private void heartbeatLoopBody() { genHeartbeatReq(), getNodeManager().getRegisteredDataNodes()); // Send heartbeat requests to all the registered AINodes pingRegisteredAINodes(genAIHeartbeatReq(), getNodeManager().getRegisteredAINodes()); + // Sample free-space on the same cadence DataNode samples its load. Runs after + // the async heartbeat dispatches so the OS call does not delay fanout. DiskCrash + // is observed passively by the Ratis write-path, not polled here. + if (heartbeatCounter.get() % 10 == 0) { + DiskChecker.checkFreeRatioAndApply( + ConfigNodeDescriptor.getInstance().getConf().getCriticalDirs(), + CommonDescriptor.getInstance().getConfig().getDiskSpaceWarningThreshold()); + reconcileConfigNodePeerPriorities(); + } } }); } + /** + * Push Ratis peer priorities to reflect each ConfigNode's current {@link NodeStatus} (see {@link + * NodeStatus#priorityForStatus}). Only fires on the leader; only acts when at least one peer's + * desired priority differs from the live group configuration. {@code Unknown}/{@code Removing} + * peers and non-disk {@code ReadOnly} reasons are left untouched so transient blips and + * operator-driven states cannot churn the group config. + */ + private void reconcileConfigNodePeerPriorities() { + Map desired = new HashMap<>(); + for (TConfigNodeLocation peer : getNodeManager().getRegisteredConfigNodes()) { + int nodeId = peer.getConfigNodeId(); + NodeStatus status; + String reason; + if (nodeId == ConfigNodeHeartbeatCache.CURRENT_NODE_ID) { + // The leader's own cache entry mirrors CommonConfig; reading from CommonConfig directly + // sidesteps any ordering between the local disk-check and the next cache refresh. + status = CommonDescriptor.getInstance().getConfig().getNodeStatus(); + reason = CommonDescriptor.getInstance().getConfig().getStatusReason(); + } else { + status = loadCache.getNodeStatus(nodeId); + reason = loadCache.getNodeStatusReason(nodeId); + } + NodeStatus.priorityForStatus(status, reason) + .ifPresent(priority -> desired.put(nodeId, priority)); + } + if (desired.isEmpty()) { + return; + } + try { + configManager + .getConsensusManager() + .getConsensusImpl() + .reconfigurePeerPriorities(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID, desired); + } catch (ConsensusException e) { + LOGGER.warn(ManagerMessages.RECONFIGURE_PEER_PRIORITIES_FAILED, desired, e); + } + } + protected TDataNodeHeartbeatReq genHeartbeatReq() { /* Generate heartbeat request */ TDataNodeHeartbeatReq heartbeatReq = new TDataNodeHeartbeatReq(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java index b6fe2c0fc219..ea60ffd33447 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java @@ -40,6 +40,7 @@ import org.apache.iotdb.commons.auth.entity.PrivilegeModelType; import org.apache.iotdb.commons.auth.entity.PrivilegeType; import org.apache.iotdb.commons.auth.entity.PrivilegeUnion; +import org.apache.iotdb.commons.cluster.DiskChecker; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConsensusGroupId; @@ -252,12 +253,17 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; /** ConfigNodeRPCServer exposes the interface that interacts with the DataNode */ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Iface { private static final Logger LOGGER = LoggerFactory.getLogger(ConfigNodeRPCServiceProcessor.class); + private static final int DISK_CHECK_INTERVAL_IN_HEARTBEATS = 10; + + private final AtomicLong heartbeatReceivedCounter = new AtomicLong(0); + protected final CommonConfig commonConfig; protected final ConfigNodeConfig configNodeConfig; protected final ConfigNode configNode; @@ -1114,6 +1120,16 @@ public TRegionRouteMapResp getLatestRegionRouteMap() { public TConfigNodeHeartbeatResp getConfigNodeHeartBeat(TConfigNodeHeartbeatReq heartbeatReq) { TConfigNodeHeartbeatResp resp = new TConfigNodeHeartbeatResp(); resp.setTimestamp(heartbeatReq.getTimestamp()); + // Sample free-space on the same cadence DataNode samples its load. DiskCrash is observed + // passively from the Ratis write-path on this node, not polled here. + if (heartbeatReceivedCounter.getAndIncrement() % DISK_CHECK_INTERVAL_IN_HEARTBEATS == 0) { + DiskChecker.checkFreeRatioAndApply( + configNodeConfig.getCriticalDirs(), commonConfig.getDiskSpaceWarningThreshold()); + } + resp.setStatus(commonConfig.getNodeStatus().getStatus()); + if (commonConfig.getStatusReason() != null) { + resp.setStatusReason(commonConfig.getStatusReason()); + } return resp; } diff --git a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java index f980de691b26..0857f9107840 100644 --- a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java +++ b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java @@ -35,6 +35,9 @@ private RatisMessages() {} "null reply received in writeWithRetry for request "; public static final String LEADER_READ_ONLY_STEP_DOWN_FAILED = "leader {} read only, force step down failed due to, "; + public static final String + LOG_DISK_FAILURE_OBSERVED_IN_RATIS_GROUP_ARG_MARKING_NODE_READONLY_DISKCRASH_31E461D2 = + "Disk failure observed in Ratis group {}; marking node ReadOnly(DiskCrash)."; public static final String TRY_ADD_CONFLICTING_PEER = "{}: try to add a peer {} with conflicting id or address in {}"; public static final String IS_LEADER_REQUEST_FAILED = diff --git a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java index fdf4660bf8f7..73cde2a8ce6c 100644 --- a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java +++ b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java @@ -34,6 +34,9 @@ private RatisMessages() {} "writeWithRetry 中收到空回复，请求为 "; public static final String LEADER_READ_ONLY_STEP_DOWN_FAILED = "leader {} 处于只读模式，强制降级失败，原因 "; + public static final String + LOG_DISK_FAILURE_OBSERVED_IN_RATIS_GROUP_ARG_MARKING_NODE_READONLY_DISKCRASH_31E461D2 = + "在 Ratis 组 {} 中发现磁盘故障，节点标记为 ReadOnly(DiskCrash)。"; public static final String TRY_ADD_CONFLICTING_PEER = "{}：尝试添加 ID 或地址冲突的 peer {} 到 {}"; public static final String IS_LEADER_REQUEST_FAILED = diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java index c462aa3a046e..20b2af0d1ba2 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java @@ -167,6 +167,24 @@ public interface IConsensus { */ void resetPeerList(ConsensusGroupId groupId, List correctPeers) throws ConsensusException; + /** + * Adjust the leader-election priority of one or more peers in {@code groupId}. + * + *

Implementations that have no concept of per-peer priority (e.g. Simple, IoT consensus) + * should leave the default no-op. The Ratis implementation rewrites the group configuration so a + * peer with a lower priority is less likely to win subsequent elections — this is the lever the + * cluster uses to demote a {@code ReadOnly} ConfigNode. + * + * @param groupId the consensus group whose peer priorities should be updated + * @param nodeIdToPriority desired priorities keyed by node id; peers not in the map keep their + * current priority + * @throws ConsensusException if the underlying reconfiguration fails + */ + default void reconfigurePeerPriorities( + ConsensusGroupId groupId, Map nodeIdToPriority) throws ConsensusException { + // no-op default + } + // management API /** diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java index 67b1fdf7da90..c18129f93ac5 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java @@ -70,17 +70,20 @@ public class ApplicationStateMachineProxy extends BaseStateMachine { private final RaftGroupId groupId; private final TConsensusGroupType consensusGroupType; private final BiConsumer leaderChangeListener; + private final BiConsumer diskFailureListener; ApplicationStateMachineProxy(IStateMachine stateMachine, RaftGroupId id) { - this(stateMachine, id, null); + this(stateMachine, id, null, null); } ApplicationStateMachineProxy( IStateMachine stateMachine, RaftGroupId id, - BiConsumer onLeaderChanged) { + BiConsumer onLeaderChanged, + BiConsumer onDiskFailure) { this.applicationStateMachine = stateMachine; this.leaderChangeListener = onLeaderChanged; + this.diskFailureListener = onDiskFailure; this.groupId = id; snapshotStorage = new SnapshotStorage(applicationStateMachine, groupId); consensusGroupType = Utils.getConsensusGroupTypeFromPrefix(groupId.toString()); @@ -161,6 +164,9 @@ && waitBeforeRetry()) { break; } catch (Throwable rte) { logger.error(RatisMessages.STATEMACHINE_RUNTIME_EXCEPTION, rte); + if (Utils.isDiskFailure(rte) && diskFailureListener != null) { + diskFailureListener.accept(groupId, rte); + } ret = new ResponseMessage( new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java index b7369546e7cd..44f481002d57 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.client.IClientPoolFactory; import org.apache.iotdb.commons.client.exception.ClientManagerException; import org.apache.iotdb.commons.client.property.ClientPoolProperty; +import org.apache.iotdb.commons.cluster.DiskChecker; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.commons.service.metric.MetricService; @@ -234,7 +235,8 @@ public RatisConsensus(ConsensusConfig config, IStateMachine.Registry registry) { registry.apply( Utils.fromRaftGroupIdToConsensusGroupId(raftGroupId)), raftGroupId, - this::onLeaderChanged)) + this::onLeaderChanged, + this::onDiskFailure)) .build()); } @@ -329,9 +331,9 @@ public TSStatus write(ConsensusGroupId groupId, IConsensusRequest request) throw new ConsensusGroupNotExistException(groupId); } - // current Peer is group leader and in ReadOnly State - // We only judge dataRegions here, because schema write when readOnly is handled at - // RegionWriteExecutor + // Current peer is group leader and in ReadOnly state. SchemaRegion writes are gated at + // RegionWriteExecutor; here we step down DataRegion and ConfigRegion leaders so that a + // healthy peer can take over and continue serving writes. if (isLeader(groupId) && Utils.rejectWrite(consensusGroupType)) { try { forceStepDownLeader(raftGroup); @@ -369,6 +371,7 @@ && waitUntilLeaderReady(raftGroupId)) { } catch (GroupMismatchException e) { throw new ConsensusGroupNotExistException(groupId); } catch (Exception e) { + maybeReportDiskFailure(raftGroupId, e); throw new RatisRequestFailedException(e); } } @@ -386,6 +389,7 @@ && waitUntilLeaderReady(raftGroupId)) { } catch (GroupMismatchException e) { throw new ConsensusGroupNotExistException(groupId); } catch (Exception e) { + maybeReportDiskFailure(raftGroupId, e); throw new RatisRequestFailedException(e); } @@ -965,6 +969,39 @@ private RatisClient getRaftClient(RaftGroup group) throws ClientManagerException } } + @Override + public void reconfigurePeerPriorities( + ConsensusGroupId groupId, Map nodeIdToPriority) throws ConsensusException { + if (nodeIdToPriority == null || nodeIdToPriority.isEmpty()) { + return; + } + RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId); + RaftGroup group = getGroupInfo(raftGroupId); + if (group == null || !group.getPeers().contains(myself)) { + throw new ConsensusGroupNotExistException(groupId); + } + boolean changed = false; + List newPeers = new ArrayList<>(group.getPeers().size()); + for (RaftPeer p : group.getPeers()) { + Integer desired = nodeIdToPriority.get(Utils.fromRaftPeerIdToNodeId(p.getId())); + if (desired == null || desired == p.getPriority()) { + newPeers.add(p); + continue; + } + newPeers.add( + RaftPeer.newBuilder() + .setId(p.getId()) + .setAddress(p.getAddress()) + .setPriority(desired) + .build()); + changed = true; + } + if (!changed) { + return; + } + sendReconfiguration(RaftGroup.valueOf(raftGroupId, newPeers)); + } + private RatisClient getConfigurationRaftClient(RaftGroup group) throws ClientManagerException { try { return reconfigurationClientManager.borrowClient(group); @@ -991,6 +1028,27 @@ private RaftClientReply sendReconfiguration(RaftGroup newGroupConf) return reply; } + /** + * Called from the state machine apply path and from the write-path catch blocks when a Ratis + * operation surfaces a disk-level error (see {@link Utils#isDiskFailure(Throwable)}). Drives the + * node to {@code ReadOnly(DiskCrash)} via {@link DiskChecker#apply}, which encapsulates the + * priority rules (DiskCrash wins over DiskFull, etc.). + */ + private void onDiskFailure(RaftGroupId groupId, Throwable cause) { + logger.error( + RatisMessages + .LOG_DISK_FAILURE_OBSERVED_IN_RATIS_GROUP_ARG_MARKING_NODE_READONLY_DISKCRASH_31E461D2, + groupId, + cause); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + } + + private void maybeReportDiskFailure(RaftGroupId groupId, Throwable cause) { + if (Utils.isDiskFailure(cause)) { + onDiskFailure(groupId, cause); + } + } + private void onLeaderChanged(RaftGroupMemberId groupMemberId, RaftPeerId leaderId) { Optional.ofNullable(canServeStaleRead) .ifPresent( diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java index 7b052cebf979..1102e6ecb329 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java @@ -238,7 +238,10 @@ public static TConsensusGroupType getConsensusGroupTypeFromPrefix(String prefix) } public static boolean rejectWrite(TConsensusGroupType type) { - return type == TConsensusGroupType.DataRegion && config.isReadOnly(); + // SchemaRegion writes are gated by RegionWriteExecutor at a higher level, so we only need + // to short-circuit DataRegion and ConfigRegion Ratis writes here. + return (type == TConsensusGroupType.DataRegion || type == TConsensusGroupType.ConfigRegion) + && config.isReadOnly(); } /** @@ -248,7 +251,30 @@ public static boolean rejectWrite(TConsensusGroupType type) { * still allow statemachine to apply while rejecting new client write requests. */ public static boolean stallApply(TConsensusGroupType type) { - return type == TConsensusGroupType.DataRegion && config.isReadOnly() && !config.isStopping(); + return (type == TConsensusGroupType.DataRegion || type == TConsensusGroupType.ConfigRegion) + && config.isReadOnly() + && !config.isStopping(); + } + + /** + * Treat a throwable (including any wrapped cause) as a disk-level failure when it carries an + * {@link java.io.IOError} or a {@link java.nio.file.FileSystemException}. Both are concrete + * filesystem-layer signals: {@code IOError} is thrown by the JVM for unrecoverable storage + * errors, and {@code FileSystemException} surfaces I/O syscalls failing on a specific file. Plain + * {@link java.io.IOException} is intentionally excluded — it is also raised by network code and + * would otherwise misclassify transient connectivity failures as disk crashes. + */ + public static boolean isDiskFailure(Throwable t) { + for (Throwable cur = t; cur != null; cur = cur.getCause()) { + if (cur instanceof java.io.IOError || cur instanceof java.nio.file.FileSystemException) { + return true; + } + // Guard against pathological self-referencing cause chains + if (cur.getCause() == cur) { + break; + } + } + return false; } /** return the max wait duration for retry */ diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index cde5b09b4578..33b4941f0762 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -49,6 +49,7 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.auth.entity.PrivilegeType; import org.apache.iotdb.commons.client.request.AsyncRequestContext; +import org.apache.iotdb.commons.cluster.DiskChecker; import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.concurrent.Await; import org.apache.iotdb.commons.concurrent.AwaitTimeoutException; @@ -65,6 +66,7 @@ import org.apache.iotdb.commons.consensus.SchemaRegionId; import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.ProgressIndexType; +import org.apache.iotdb.commons.disk.FolderManager; import org.apache.iotdb.commons.enums.DataPartitionTableGeneratorState; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; @@ -219,6 +221,8 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALWriteBlockStatus; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeThrottleQuotaManager; import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; @@ -2370,6 +2374,7 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th .forEach((key, value) -> regionRawDataSize.put(Integer.parseInt(key), value.getLeft())); resp.setDataRegionRawDataSize(regionRawDataSize); } + updateWALBlockedStatus(); resp.setHeartbeatTimestamp(req.getHeartbeatTimestamp()); resp.setStatus(commonConfig.getNodeStatus().getStatus()); // Advertise that this DataNode supports metadata-lease self-fencing, so the ConfigNode may @@ -2435,6 +2440,11 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th return resp; } + private void updateWALBlockedStatus() { + WALWriteBlockStatus.updateStatus( + commonConfig, WALManager.getInstance().isLongTermWriteBlocked()); + } + @Override public TSStatus updateRegionCache(TRegionRouteReq req) { boolean result = ClusterPartitionFetcher.getInstance().updateRegionCache(req); @@ -2555,12 +2565,20 @@ private void sampleDiskLoad(TLoadSample loadSample) { SYSTEM) .getValue(); + // Derive the disk status: an ABNORMAL data folder observed by any FolderManager wins over + // a full-disk reading, and a low free-space ratio still drives DISK_FULL when nothing is + // crashed. DiskChecker.apply then performs the actual NodeStatus transition (including the + // "ReadOnly(DiskFull|DiskCrash) -> Running" recovery for the DiskFull path). + DiskChecker.DiskStatus diskStatus = DiskChecker.DiskStatus.NORMAL; + if (FolderManager.hasAnyAbnormalFolder()) { + diskStatus = DiskChecker.DiskStatus.DISK_CRASH; + } if (availableDisk != 0 && totalDisk != 0) { double freeDiskRatio = availableDisk / totalDisk; loadSample.setFreeDiskSpace(availableDisk); loadSample.setDiskUsageRate(1d - freeDiskRatio); - // Reset NodeStatus if necessary - if (freeDiskRatio < commonConfig.getDiskSpaceWarningThreshold()) { + if (diskStatus == DiskChecker.DiskStatus.NORMAL + && freeDiskRatio < commonConfig.getDiskSpaceWarningThreshold()) { LOGGER.warn( DataNodeMiscMessages .MISC_LOG_THE_AVAILABLE_DISK_SPACE_IS_THE_TOTAL_DISK_SPACE_IS_AND_4506856F, @@ -2568,14 +2586,13 @@ private void sampleDiskLoad(TLoadSample loadSample) { RamUsageEstimator.humanReadableUnits((long) totalDisk), freeDiskRatio, commonConfig.getDiskSpaceWarningThreshold()); - commonConfig.setNodeStatus(NodeStatus.ReadOnly); - commonConfig.setStatusReason(NodeStatus.DISK_FULL); - } else if (NodeStatus.ReadOnly.equals(commonConfig.getNodeStatus()) - && NodeStatus.DISK_FULL.equals(commonConfig.getStatusReason())) { - commonConfig.setNodeStatus(NodeStatus.Running); - commonConfig.setStatusReason(null); + diskStatus = DiskChecker.DiskStatus.DISK_FULL; } + } else if (diskStatus == DiskChecker.DiskStatus.NORMAL) { + // Metrics not available yet — fall back to no-op so we don't churn the status. + return; } + DiskChecker.apply(diskStatus); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java index 8fafee304b2b..a579871ba616 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java @@ -52,6 +52,7 @@ import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import static org.apache.iotdb.commons.conf.IoTDBConstant.FILE_NAME_SEPARATOR; @@ -69,6 +70,9 @@ public class WALManager implements IService { private final AtomicLong totalDiskUsage = new AtomicLong(); // total number of wal files private final AtomicLong totalFileNum = new AtomicLong(); + private final AtomicLong walThrottleStartTimeInMs = new AtomicLong(-1); + private final AtomicLong walBufferQueueBlockedStartTimeInMs = new AtomicLong(-1); + private final AtomicInteger walBufferQueueBlockedWriterCount = new AtomicInteger(0); private WALManager() { if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS) @@ -237,8 +241,55 @@ public boolean shouldThrottle() { return getTotalDiskUsage() >= getThrottleThreshold(); } + public boolean isLongTermWriteBlocked() { + return isLongTermWalThrottle() || isLongTermWalBufferQueueBlocked(); + } + + private boolean isLongTermWalThrottle() { + if (!shouldThrottle()) { + walThrottleStartTimeInMs.set(-1); + return false; + } + return isLongTermBlocked(walThrottleStartTimeInMs); + } + + private boolean isLongTermWalBufferQueueBlocked() { + if (walBufferQueueBlockedWriterCount.get() <= 0) { + walBufferQueueBlockedStartTimeInMs.set(-1); + return false; + } + return isLongTermBlocked(walBufferQueueBlockedStartTimeInMs); + } + + private boolean isLongTermBlocked(AtomicLong blockStartTimeInMs) { + long currentTimeInMs = System.currentTimeMillis(); + long blockStartTime = blockStartTimeInMs.get(); + if (blockStartTime < 0) { + blockStartTimeInMs.compareAndSet(-1, currentTimeInMs); + blockStartTime = blockStartTimeInMs.get(); + } + return currentTimeInMs - blockStartTime >= config.getMaxWaitingTimeWhenInsertBlocked(); + } + + public void markWalBufferQueueBlocked() { + if (walBufferQueueBlockedWriterCount.getAndIncrement() == 0) { + walBufferQueueBlockedStartTimeInMs.compareAndSet(-1, System.currentTimeMillis()); + } + } + + public void markWalBufferQueueAvailable() { + if (walBufferQueueBlockedWriterCount.decrementAndGet() <= 0) { + walBufferQueueBlockedWriterCount.set(0); + walBufferQueueBlockedStartTimeInMs.set(-1); + } + } + public long getThrottleThreshold() { - return (long) (config.getThrottleThreshold() * 0.8); + long configuredThrottleThreshold = config.getThrottleThreshold(); + if (configuredThrottleThreshold <= 0) { + return Long.MAX_VALUE; + } + return Math.max((long) (configuredThrottleThreshold * 0.8), 1); } public long getTotalDiskUsage() { @@ -323,6 +374,9 @@ public void syncDeleteOutdatedFilesInWALNodes() { public void clear() { totalDiskUsage.set(0); + walThrottleStartTimeInMs.set(-1); + walBufferQueueBlockedStartTimeInMs.set(-1); + walBufferQueueBlockedWriterCount.set(0); walNodesManager.clear(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java new file mode 100644 index 000000000000..16facd9eafd4 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java @@ -0,0 +1,41 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.storageengine.dataregion.wal; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonConfig; + +public final class WALWriteBlockStatus { + + private WALWriteBlockStatus() {} + + public static void updateStatus(CommonConfig commonConfig, boolean longTermWriteBlocked) { + if (longTermWriteBlocked) { + if (NodeStatus.Running.equals(commonConfig.getNodeStatus())) { + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + commonConfig.setStatusReason(NodeStatus.WAL_BLOCKED); + } + } else if (NodeStatus.ReadOnly.equals(commonConfig.getNodeStatus()) + && NodeStatus.WAL_BLOCKED.equals(commonConfig.getStatusReason())) { + commonConfig.setNodeStatus(NodeStatus.Running); + commonConfig.setStatusReason(null); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java index 0974aef999a3..63d16eb6db8a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; @@ -51,24 +52,37 @@ public WALEntry poll(long timeout, TimeUnit unit) throws InterruptedException { public void put(WALEntry e) throws InterruptedException { long elementSize = getElementSize(e); - synchronized (nonFullCondition) { - while (!SystemInfo.getInstance().getWalBufferQueueMemoryBlock().allocate(elementSize)) { - if (elementSize - > SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getTotalMemorySizeInBytes()) { - throw new IoTDBRuntimeException( - String.format( - StorageEngineMessages - .STORAGE_EXCEPTION_THE_ELEMENT_SIZE_OF_WALENTRY_S_IS_LARGER_THAN_THE_TOTAL_E494520D, - elementSize, - SystemInfo.getInstance() - .getWalBufferQueueMemoryBlock() - .getTotalMemorySizeInBytes()), - WAL_ENTRY_TOO_LARGE.getStatusCode()); + boolean blocked = false; + try { + synchronized (nonFullCondition) { + while (!SystemInfo.getInstance().getWalBufferQueueMemoryBlock().allocate(elementSize)) { + if (elementSize + > SystemInfo.getInstance() + .getWalBufferQueueMemoryBlock() + .getTotalMemorySizeInBytes()) { + throw new IoTDBRuntimeException( + String.format( + StorageEngineMessages + .STORAGE_EXCEPTION_THE_ELEMENT_SIZE_OF_WALENTRY_S_IS_LARGER_THAN_THE_TOTAL_E494520D, + elementSize, + SystemInfo.getInstance() + .getWalBufferQueueMemoryBlock() + .getTotalMemorySizeInBytes()), + WAL_ENTRY_TOO_LARGE.getStatusCode()); + } + if (!blocked) { + blocked = true; + WALManager.getInstance().markWalBufferQueueBlocked(); + } + nonFullCondition.wait(); } - nonFullCondition.wait(); + } + queue.put(e); + } finally { + if (blocked) { + WALManager.getInstance().markWalBufferQueueAvailable(); } } - queue.put(e); } public WALEntry take() throws InterruptedException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java index 2017805ac7e2..0b43f1f21f3a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java @@ -44,6 +44,7 @@ import java.io.File; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; @@ -58,11 +59,15 @@ public class WALManagerTest { }; private String[] prevWalDirs; private String prevConsensus; + private long prevThrottleThreshold; + private int prevMaxWaitingTimeWhenInsertBlocked; @Before public void setUp() throws Exception { prevConsensus = config.getDataRegionConsensusProtocolClass(); prevWalDirs = commonConfig.getWalDirs(); + prevThrottleThreshold = config.getThrottleThreshold(); + prevMaxWaitingTimeWhenInsertBlocked = config.getMaxWaitingTimeWhenInsertBlocked(); config.setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); commonConfig.setWalDirs(walDirs); EnvironmentUtils.envSetUp(); @@ -76,6 +81,8 @@ public void tearDown() throws Exception { } config.setDataRegionConsensusProtocolClass(prevConsensus); commonConfig.setWalDirs(prevWalDirs); + config.setThrottleThreshold(prevThrottleThreshold); + config.setMaxWaitingTimeWhenInsertBlocked(prevMaxWaitingTimeWhenInsertBlocked); } @Test @@ -118,6 +125,58 @@ public void testDeleteOutdatedWALFiles() throws IllegalPathException { } } + @Test + public void testLongTermWriteBlockedByWalThrottle() { + WALManager walManager = WALManager.getInstance(); + walManager.addTotalDiskUsage(walManager.getThrottleThreshold()); + + assertFalse(walManager.isLongTermWriteBlocked()); + + config.setMaxWaitingTimeWhenInsertBlocked(0); + assertTrue(walManager.isLongTermWriteBlocked()); + + walManager.clear(); + assertFalse(walManager.isLongTermWriteBlocked()); + } + + @Test + public void testNonPositiveWalThrottleThresholdIsIgnored() { + WALManager walManager = WALManager.getInstance(); + config.setThrottleThreshold(0); + walManager.addTotalDiskUsage(1); + + assertEquals(Long.MAX_VALUE, walManager.getThrottleThreshold()); + assertFalse(walManager.shouldThrottle()); + assertFalse(walManager.isLongTermWriteBlocked()); + + walManager.clear(); + config.setThrottleThreshold(-1); + walManager.addTotalDiskUsage(1); + assertEquals(Long.MAX_VALUE, walManager.getThrottleThreshold()); + assertFalse(walManager.shouldThrottle()); + + walManager.clear(); + config.setThrottleThreshold(1); + assertEquals(1, walManager.getThrottleThreshold()); + walManager.addTotalDiskUsage(1); + assertTrue(walManager.shouldThrottle()); + walManager.clear(); + } + + @Test + public void testLongTermWriteBlockedByWalBufferQueue() { + WALManager walManager = WALManager.getInstance(); + walManager.markWalBufferQueueBlocked(); + + assertFalse(walManager.isLongTermWriteBlocked()); + + config.setMaxWaitingTimeWhenInsertBlocked(0); + assertTrue(walManager.isLongTermWriteBlocked()); + + walManager.markWalBufferQueueAvailable(); + assertFalse(walManager.isLongTermWriteBlocked()); + } + private InsertRowNode getInsertRowNode() throws IllegalPathException { long time = 110L; TSDataType[] dataTypes = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java new file mode 100644 index 000000000000..c0dd2620a22f --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java @@ -0,0 +1,92 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.storageengine.dataregion.wal; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonConfig; + +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class WALWriteBlockStatusTest { + + @Test + public void testRunningNodeTurnsReadOnlyWhenWalBlocked() { + CommonConfig commonConfig = mockCommonConfig(NodeStatus.Running, null); + + WALWriteBlockStatus.updateStatus(commonConfig, true); + + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertEquals(NodeStatus.WAL_BLOCKED, commonConfig.getStatusReason()); + } + + @Test + public void testWalBlockedReadOnlyNodeRecovers() { + CommonConfig commonConfig = mockCommonConfig(NodeStatus.ReadOnly, NodeStatus.WAL_BLOCKED); + + WALWriteBlockStatus.updateStatus(commonConfig, false); + + assertEquals(NodeStatus.Running, commonConfig.getNodeStatus()); + assertNull(commonConfig.getStatusReason()); + } + + @Test + public void testOtherReadOnlyReasonIsNotOverwrittenOrRecovered() { + CommonConfig commonConfig = mockCommonConfig(NodeStatus.ReadOnly, NodeStatus.DISK_FULL); + + WALWriteBlockStatus.updateStatus(commonConfig, true); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, commonConfig.getStatusReason()); + + WALWriteBlockStatus.updateStatus(commonConfig, false); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, commonConfig.getStatusReason()); + } + + private CommonConfig mockCommonConfig(NodeStatus initialStatus, String initialStatusReason) { + AtomicReference status = new AtomicReference<>(initialStatus); + AtomicReference statusReason = new AtomicReference<>(initialStatusReason); + CommonConfig commonConfig = Mockito.mock(CommonConfig.class); + + Mockito.when(commonConfig.getNodeStatus()).thenAnswer(invocation -> status.get()); + Mockito.when(commonConfig.getStatusReason()).thenAnswer(invocation -> statusReason.get()); + Mockito.doAnswer( + invocation -> { + status.set(invocation.getArgument(0)); + statusReason.set(null); + return null; + }) + .when(commonConfig) + .setNodeStatus(Mockito.any(NodeStatus.class)); + Mockito.doAnswer( + invocation -> { + statusReason.set(invocation.getArgument(0)); + return null; + }) + .when(commonConfig) + .setStatusReason(Mockito.any()); + return commonConfig; + } +} diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index ae97d09ddc25..cacd8ac15f25 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -37,6 +37,14 @@ public final class CommonMessages { public static final String NODE_STATUS_NOT_EXIST = "NodeStatus %s doesn't exist."; public static final String UNKNOWN_NODE_STATUS = "Unknown NodeStatus %s."; + // --- disk health --- + public static final String DISK_FULL_SET_READ_ONLY = + "Free disk space ratio is below the configured threshold; set node status to ReadOnly(DiskFull)."; + public static final String DISK_CRASH_SET_READ_ONLY = + "Detected unwritable disk directory; set node status to ReadOnly(DiskCrash)."; + public static final String DISK_RECOVERED_SET_RUNNING = + "Disk health recovered (previous reason: {}); set node status to Running."; + // --- consensus --- public static final String UNRECOGNIZED_CONSENSUS_GROUP_ID = "Unrecognized ConsensusGroupId: %s"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 71c45ccaf3fc..33f610e769fc 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -36,6 +36,14 @@ public final class CommonMessages { public static final String NODE_STATUS_NOT_EXIST = "NodeStatus %s 不存在。"; public static final String UNKNOWN_NODE_STATUS = "未知 NodeStatus %s。"; + // --- disk health --- + public static final String DISK_FULL_SET_READ_ONLY = + "磁盘剩余空间比例低于配置阈值，将节点状态设为 ReadOnly(DiskFull)。"; + public static final String DISK_CRASH_SET_READ_ONLY = + "检测到不可写的磁盘目录，将节点状态设为 ReadOnly(DiskCrash)。"; + public static final String DISK_RECOVERED_SET_RUNNING = + "磁盘健康已恢复（先前原因：{}），将节点状态设为 Running。"; + // --- consensus --- public static final String UNRECOGNIZED_CONSENSUS_GROUP_ID = "无法识别的 ConsensusGroupId：%s"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java new file mode 100644 index 000000000000..682f9ad4a97a --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java @@ -0,0 +1,136 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.cluster; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.i18n.CommonMessages; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.util.List; + +/** + * Shared utility that drives the global {@link NodeStatus} on {@link CommonConfig} based on disk + * health signals. + * + *

Two sources feed it: + * + *

    + *
  • {@link #checkFreeRatioAndApply} polls usable / total space across critical directories and + * sets {@code ReadOnly(DISK_FULL)} when the ratio drops below a threshold. + *
  • {@link #apply}{@code (DISK_CRASH)} is called from passive failure observers — Ratis + * write-path catch blocks on ConfigNode and {@code FolderManager.ABNORMAL} aggregation on + * DataNode — when a real write IO error has already occurred. + *
+ * + *

Only transitions between {@link NodeStatus#Running} and {@link NodeStatus#ReadOnly} with + * reason {@link NodeStatus#DISK_FULL}/{@link NodeStatus#DISK_CRASH} are managed here. Other + * ReadOnly reasons (e.g. manual) are left untouched. + */ +public class DiskChecker { + + private static final Logger LOGGER = LoggerFactory.getLogger(DiskChecker.class); + + public enum DiskStatus { + NORMAL, + DISK_FULL, + DISK_CRASH + } + + private DiskChecker() {} + + /** + * Evaluate the usable/total space ratio of each directory. Any directory whose ratio is below + * {@code freeRatioThreshold} yields {@link DiskStatus#DISK_FULL}; otherwise {@link + * DiskStatus#NORMAL}. This method never returns {@link DiskStatus#DISK_CRASH} — crash detection + * is driven by the Ratis write-path observer on ConfigNode and by {@code FolderManager} on + * DataNode, both of which call {@link #apply} directly. + */ + public static DiskStatus checkFreeRatio(List dirs, double freeRatioThreshold) { + if (dirs == null || dirs.isEmpty()) { + return DiskStatus.NORMAL; + } + for (String dir : dirs) { + if (dir == null || dir.isEmpty()) { + continue; + } + File f = new File(dir); + long total = f.getTotalSpace(); + long usable = f.getUsableSpace(); + if (total > 0 && (double) usable / total < freeRatioThreshold) { + return DiskStatus.DISK_FULL; + } + } + return DiskStatus.NORMAL; + } + + /** Convenience: run {@link #checkFreeRatio} and apply the result to {@link CommonConfig}. */ + public static void checkFreeRatioAndApply(List dirs, double freeRatioThreshold) { + apply(checkFreeRatio(dirs, freeRatioThreshold)); + } + + /** + * Apply a precomputed status to {@link CommonConfig}. Priority is {@code DiskCrash > DiskFull > + * Normal}; recovery to {@code Running} only fires when the active reason was disk-related. + */ + public static void apply(DiskStatus result) { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + NodeStatus currentStatus = config.getNodeStatus(); + String currentReason = config.getStatusReason(); + boolean currentlyFull = + NodeStatus.ReadOnly.equals(currentStatus) && NodeStatus.DISK_FULL.equals(currentReason); + boolean currentlyCrash = + NodeStatus.ReadOnly.equals(currentStatus) && NodeStatus.DISK_CRASH.equals(currentReason); + + switch (result) { + case DISK_CRASH: + if (!currentlyCrash) { + LOGGER.warn(CommonMessages.DISK_CRASH_SET_READ_ONLY); + config.setNodeStatus(NodeStatus.ReadOnly); + config.setStatusReason(NodeStatus.DISK_CRASH); + } + break; + case DISK_FULL: + // DiskCrash has higher priority — do not downgrade an existing crash to full. + if (currentlyCrash) { + return; + } + if (!currentlyFull) { + LOGGER.warn(CommonMessages.DISK_FULL_SET_READ_ONLY); + config.setNodeStatus(NodeStatus.ReadOnly); + config.setStatusReason(NodeStatus.DISK_FULL); + } + break; + case NORMAL: + default: + // DiskCrash is sticky — only a restart clears it. The free-ratio probe can recover + // DiskFull alone because free-space reappearing is the literal inverse of running low. + if (currentlyFull) { + LOGGER.info(CommonMessages.DISK_RECOVERED_SET_RUNNING, currentReason); + config.setNodeStatus(NodeStatus.Running); + config.setStatusReason(null); + } + break; + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java index 518a9faaed2e..dc5b2cafdeba 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java @@ -21,6 +21,8 @@ import org.apache.iotdb.commons.i18n.CommonMessages; +import java.util.OptionalInt; + /** Node status for showing cluster */ public enum NodeStatus { /** Node running properly */ @@ -35,6 +37,8 @@ public enum NodeStatus { /** Only query statements are permitted */ ReadOnly("ReadOnly"); public static final String DISK_FULL = "DiskFull"; + public static final String DISK_CRASH = "DiskCrash"; + public static final String WAL_BLOCKED = "WALBlocked"; private final String status; @@ -60,6 +64,35 @@ public static boolean isNormalStatus(NodeStatus status) { return status != null && status.equals(NodeStatus.Running); } + /** + * Map a node's {@link NodeStatus} (and optional reason) to the Ratis peer priority that should + * govern its candidacy in leader elections. + * + * 

+   *   Running                          →   0   (full candidate)
+   *   ReadOnly + {@link #DISK_FULL}    →  -1   (out-rank healthy peers but ahead of crashed)
+   *   ReadOnly + {@link #DISK_CRASH}   →  -2   (most degraded — last choice)
+   *   anything else                    →  empty (priority must not be changed)
+   *
+ * + * Returning {@link OptionalInt#empty()} for Unknown/Removing/manual ReadOnly keeps transient + * blips and operator-driven states from rewriting peer priorities. + */ + public static OptionalInt priorityForStatus(NodeStatus status, String statusReason) { + if (Running.equals(status)) { + return OptionalInt.of(0); + } + if (ReadOnly.equals(status)) { + if (DISK_CRASH.equals(statusReason)) { + return OptionalInt.of(-2); + } + if (DISK_FULL.equals(statusReason)) { + return OptionalInt.of(-1); + } + } + return OptionalInt.empty(); + } + public static boolean isReadable(NodeStatus status) { switch (status) { case Running: diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java index a7707077f84d..06234bc23a9c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java @@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.lang.ref.WeakReference; import java.nio.file.FileStore; import java.nio.file.Files; import java.nio.file.Path; @@ -43,10 +44,20 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; public class FolderManager { private static final Logger logger = LoggerFactory.getLogger(FolderManager.class); + /** + * Registry of every live {@link FolderManager} instance so the DataNode heartbeat path can ask + * "is any folder anywhere on this node currently ABNORMAL?" without each subsystem having to push + * state into a central reporter. Weak references avoid keeping short-lived managers alive (e.g. + * those created per snapshot/load). + */ + private static final List> ALL_INSTANCES = + new CopyOnWriteArrayList<>(); + /** Represents the operational states of a data folder. */ public enum FolderState { /** Indicates the folder is functioning normally with no issues. */ @@ -69,6 +80,7 @@ public FolderManager(List folders, DirectoryStrategyType type) throws DiskSpaceInsufficientException { this.folders = folders; folders.forEach(dir -> foldersStates.put(dir, FolderState.HEALTHY)); + ALL_INSTANCES.add(new WeakReference<>(this)); switch (type) { case SEQUENCE_STRATEGY: this.selectStrategy = new SequenceStrategy(); @@ -94,7 +106,7 @@ public FolderManager(List folders, DirectoryStrategyType type) } } - public void updateFolderState(String folder, FolderState state) { + public synchronized void updateFolderState(String folder, FolderState state) { foldersStates.replace(folder, state); selectStrategy.updateFolderState(folder, state); } @@ -189,4 +201,29 @@ public String getFirstFolderOfSameDisk(String pathStr) { } return null; } + + /** + * Walks every live FolderManager instance and reports whether any folder is currently {@link + * FolderState#ABNORMAL}. Used by the DataNode heartbeat path to derive a {@code + * NodeStatus.ReadOnly(DiskCrash)} signal from already-observed write failures. + * + *

Stale (GC'd) weak references are pruned as a side effect. + */ + public static boolean hasAnyAbnormalFolder() { + for (WeakReference reference : ALL_INSTANCES) { + FolderManager folderManager = reference.get(); + if (folderManager == null) { + continue; + } + synchronized (folderManager) { + for (FolderState state : folderManager.foldersStates.values()) { + if (state == FolderState.ABNORMAL) { + return true; + } + } + } + } + ALL_INSTANCES.removeIf(ref -> ref.get() == null); + return false; + } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java new file mode 100644 index 000000000000..84284b8806a3 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.commons.cluster; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class DiskCheckerTest { + + @Rule public TemporaryFolder tmp = new TemporaryFolder(); + + private NodeStatus savedStatus; + private String savedReason; + + @Before + public void setUp() { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + savedStatus = config.getNodeStatus(); + savedReason = config.getStatusReason(); + config.setNodeStatus(NodeStatus.Running); + config.setStatusReason(null); + } + + @After + public void tearDown() { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + config.setNodeStatus(savedStatus); + config.setStatusReason(savedReason); + } + + // -- checkFreeRatio ------------------------------------------------------------------------ + + @Test + public void checkFreeRatioReturnsNormalWhenRatioMeetsThreshold() throws Exception { + File dir = tmp.newFolder(); + assertEquals( + DiskChecker.DiskStatus.NORMAL, + DiskChecker.checkFreeRatio(Collections.singletonList(dir.getAbsolutePath()), 0.0)); + } + + @Test + public void checkFreeRatioReturnsDiskFullWhenAnyDirBelowThreshold() throws Exception { + File dir = tmp.newFolder(); + // threshold above 1.0 forces any directory with positive total space to register as full + assertEquals( + DiskChecker.DiskStatus.DISK_FULL, + DiskChecker.checkFreeRatio(Collections.singletonList(dir.getAbsolutePath()), 2.0)); + } + + @Test + public void checkFreeRatioSkipsNullAndEmptyEntries() throws Exception { + File dir = tmp.newFolder(); + assertEquals( + DiskChecker.DiskStatus.NORMAL, + DiskChecker.checkFreeRatio(Arrays.asList(null, "", dir.getAbsolutePath()), 0.0)); + } + + @Test + public void emptyDirListIsNormal() { + assertEquals( + DiskChecker.DiskStatus.NORMAL, DiskChecker.checkFreeRatio(Collections.emptyList(), 1.0)); + assertEquals(DiskChecker.DiskStatus.NORMAL, DiskChecker.checkFreeRatio(null, 1.0)); + } + + // -- apply() state machine ----------------------------------------------------------------- + + @Test + public void applyDiskFullSetsReadOnlyFromRunning() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, config.getStatusReason()); + } + + @Test + public void applyDiskCrashSetsReadOnlyFromRunning() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_CRASH, config.getStatusReason()); + } + + @Test + public void applyDiskCrashUpgradesFromDiskFull() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_CRASH, config.getStatusReason()); + } + + @Test + public void applyDiskFullDoesNotDowngradeDiskCrash() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals( + "DiskCrash must outrank DiskFull", NodeStatus.DISK_CRASH, config.getStatusReason()); + } + + @Test + public void applyNormalRecoversFromDiskFull() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + DiskChecker.apply(DiskChecker.DiskStatus.NORMAL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.Running, config.getNodeStatus()); + assertNull(config.getStatusReason()); + } + + @Test + public void applyNormalDoesNotRecoverFromDiskCrash() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + DiskChecker.apply(DiskChecker.DiskStatus.NORMAL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + // DiskCrash is sticky: a free-ratio probe (the only source of NORMAL) cannot prove writes + // work again, so the node stays ReadOnly(DiskCrash) until restart. + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_CRASH, config.getStatusReason()); + } + + @Test + public void applyLeavesNonDiskReadOnlyReasonUntouched() { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + config.setNodeStatus(NodeStatus.ReadOnly); + config.setStatusReason("ManualMaintenance"); + + DiskChecker.apply(DiskChecker.DiskStatus.NORMAL); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals("ManualMaintenance", config.getStatusReason()); + + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + // DISK_FULL only fires when not already DiskFull/DiskCrash — it does take over here, + // mirroring the existing behavior for the legacy sampleDiskLoad path. + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, config.getStatusReason()); + } + + @Test + public void applyIsIdempotentForRepeatedDiskCrash() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + NodeStatus before = CommonDescriptor.getInstance().getConfig().getNodeStatus(); + String reasonBefore = CommonDescriptor.getInstance().getConfig().getStatusReason(); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + assertEquals(before, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertEquals(reasonBefore, CommonDescriptor.getInstance().getConfig().getStatusReason()); + } + + // -- checkFreeRatioAndApply ---------------------------------------------------------------- + + @Test + public void checkFreeRatioAndApplyDrivesStatusEndToEnd() throws Exception { + File dir = tmp.newFolder(); + // Threshold above 1.0 -> always "DiskFull" + DiskChecker.checkFreeRatioAndApply(Collections.singletonList(dir.getAbsolutePath()), 2.0); + assertEquals(NodeStatus.ReadOnly, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertEquals( + NodeStatus.DISK_FULL, CommonDescriptor.getInstance().getConfig().getStatusReason()); + + // Threshold 0.0 -> always "Normal" -> recovery + DiskChecker.checkFreeRatioAndApply(Collections.singletonList(dir.getAbsolutePath()), 0.0); + assertEquals(NodeStatus.Running, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertNull(CommonDescriptor.getInstance().getConfig().getStatusReason()); + } + + @Test + public void checkFreeRatioAndApplyDoesNotClearDiskCrash() throws Exception { + File dir = tmp.newFolder(); + // Simulate a Ratis-passive DiskCrash signal. + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + // Subsequent healthy free-ratio polling must keep the node in ReadOnly(DiskCrash). + DiskChecker.checkFreeRatioAndApply(Collections.singletonList(dir.getAbsolutePath()), 0.0); + assertEquals(NodeStatus.ReadOnly, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertEquals( + NodeStatus.DISK_CRASH, CommonDescriptor.getInstance().getConfig().getStatusReason()); + } +} diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift index 1111740f759a..a8c3341e5976 100644 --- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift +++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift @@ -543,6 +543,13 @@ struct TConfigNodeHeartbeatResp { 1: required i64 timestamp 2: optional string activateStatus 3: optional common.TLicense license + // Reported ConfigNode status (e.g. Running, ReadOnly). Optional for forward + // compatibility — old ConfigNodes do not populate this field, in which case + // the leader falls back to assuming Running. + 4: optional string status + // Optional human/machine readable reason accompanying ReadOnly status, + // e.g. NodeStatus.DISK_FULL or NodeStatus.DISK_CRASH. + 5: optional string statusReason } struct TAddConsensusGroupReq {