This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 930e9bc0cf7 Fix DataNode status handling for Pipe receiver disk
failures (#18484)
930e9bc0cf7 is described below
commit 930e9bc0cf7ce0ab0984501aa1d973770f164b5f
Author: Caideyipi <[email protected]>
AuthorDate: Tue Aug 25 21:04:48 2026 +0800
Fix DataNode status handling for Pipe receiver disk failures (#18484)
* Fix DataNode disk status recovery across multiple disks
* Keep Pipe receiver disk failures out of node status
* Address review comments on disk status handling
---
.../iotconsensusv2/IoTConsensusV2Receiver.java | 3 +-
.../protocol/thrift/IoTDBDataNodeReceiver.java | 4 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 28 +++---
.../DataNodeInternalRPCServiceImplDiskTest.java | 108 +++++++++++++++++++++
.../rescon/disk/FolderManagerTest.java | 25 +++++
.../apache/iotdb/commons/conf/CommonConfig.java | 6 +-
.../apache/iotdb/commons/disk/FolderManager.java | 14 +++
.../commons/disk/strategy/DirectoryStrategy.java | 18 +++-
.../iotdb/commons/conf/CommonConfigTest.java | 40 ++++++++
9 files changed, 224 insertions(+), 22 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/iotconsensusv2/IoTConsensusV2Receiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/iotconsensusv2/IoTConsensusV2Receiver.java
index 9303e4c01f4..3e465ced429 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/iotconsensusv2/IoTConsensusV2Receiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/iotconsensusv2/IoTConsensusV2Receiver.java
@@ -148,7 +148,8 @@ public class IoTConsensusV2Receiver {
}
try {
- this.folderManager = new FolderManager(receiveDirs,
DirectoryStrategyType.SEQUENCE_STRATEGY);
+ this.folderManager =
+ new FolderManager(receiveDirs,
DirectoryStrategyType.SEQUENCE_STRATEGY, false);
this.iotConsensusV2TsFileWriterPool = new
IoTConsensusV2TsFileWriterPool(consensusPipeName);
} catch (Exception e) {
LOGGER.error(DataNodePipeMessages.FAIL_TO_CREATE_IOTCONSENSUSV2_RECEIVER_FILE_FOLDERS,
e);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index 7f98e8b1193..e7d4053ea64 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -206,7 +206,9 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
try {
folderManager =
new FolderManager(
- Arrays.asList(RECEIVER_FILE_BASE_DIRS),
DirectoryStrategyType.SEQUENCE_STRATEGY);
+ Arrays.asList(RECEIVER_FILE_BASE_DIRS),
+ DirectoryStrategyType.SEQUENCE_STRATEGY,
+ false);
} catch (final DiskSpaceInsufficientException e) {
LOGGER.error(DataNodePipeMessages.FAIL_TO_CREATE_PIPE_RECEIVER_FILE_FOLDERS, e);
}
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 9e847795d64..8685df8f4be 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
@@ -228,6 +228,7 @@ import org.apache.iotdb.db.trigger.executor.TriggerExecutor;
import org.apache.iotdb.db.trigger.executor.TriggerFireResult;
import org.apache.iotdb.db.trigger.service.TriggerManagementService;
import org.apache.iotdb.db.utils.SetThreadName;
+import org.apache.iotdb.metrics.metricsets.system.SystemMetrics;
import org.apache.iotdb.metrics.type.AutoGauge;
import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.SystemMetric;
@@ -450,6 +451,8 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
private final DataNodeContext dataNodeContext;
+ private final SystemMetrics systemMetrics;
+
private final ExecutorService schemaExecutor =
new WrappedThreadPoolExecutor(
0,
@@ -465,7 +468,12 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
private static final String SYSTEM = "system";
public DataNodeInternalRPCServiceImpl(DataNodeContext dataNodeContext) {
+ this(dataNodeContext, SystemMetrics.getInstance());
+ }
+
+ DataNodeInternalRPCServiceImpl(DataNodeContext dataNodeContext,
SystemMetrics systemMetrics) {
super();
+ this.systemMetrics = systemMetrics;
partitionFetcher = ClusterPartitionFetcher.getInstance();
schemaFetcher = ClusterSchemaFetcher.getInstance();
this.dataNodeContext = dataNodeContext;
@@ -2552,23 +2560,9 @@ public class DataNodeInternalRPCServiceImpl implements
IDataNodeRPCService.Iface
return result;
}
- private void sampleDiskLoad(TLoadSample loadSample) {
- double availableDisk =
- MetricService.getInstance()
- .getAutoGauge(
- SystemMetric.SYS_DISK_AVAILABLE_SPACE.toString(),
- MetricLevel.CORE,
- Tag.NAME.toString(),
- SYSTEM)
- .getValue();
- double totalDisk =
- MetricService.getInstance()
- .getAutoGauge(
- SystemMetric.SYS_DISK_TOTAL_SPACE.toString(),
- MetricLevel.CORE,
- Tag.NAME.toString(),
- SYSTEM)
- .getValue();
+ void sampleDiskLoad(TLoadSample loadSample) {
+ double availableDisk = systemMetrics.getSystemDiskAvailableSpace();
+ double totalDisk = systemMetrics.getSystemDiskTotalSpace();
if (availableDisk != 0 && totalDisk != 0) {
double freeDiskRatio = availableDisk / totalDisk;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplDiskTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplDiskTest.java
new file mode 100644
index 00000000000..7dce3cc3c2d
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplDiskTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.protocol.thrift.impl;
+
+import org.apache.iotdb.common.rpc.thrift.TLoadSample;
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.service.DataNode.DataNodeContext;
+import org.apache.iotdb.metrics.metricsets.system.SystemMetrics;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class DataNodeInternalRPCServiceImplDiskTest {
+
+ private final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
+ private final IoTDBConfig dataNodeConfig =
IoTDBDescriptor.getInstance().getConfig();
+ private NodeStatus originalStatus;
+ private String originalStatusReason;
+ private double originalDiskSpaceWarningThreshold;
+ private int originalDataNodeId;
+
+ @Before
+ public void setUp() {
+ originalStatus = commonConfig.getNodeStatus();
+ originalStatusReason = commonConfig.getStatusReason();
+ originalDiskSpaceWarningThreshold =
commonConfig.getDiskSpaceWarningThreshold();
+ originalDataNodeId = dataNodeConfig.getDataNodeId();
+
+ dataNodeConfig.setDataNodeId(0);
+ commonConfig.setNodeStatus(NodeStatus.Running);
+ commonConfig.setStatusReason(null);
+ commonConfig.setDiskSpaceWarningThreshold(0.05);
+ commonConfig.setNodeStatus(NodeStatus.ReadOnly);
+ commonConfig.setStatusReason(NodeStatus.DISK_FULL);
+ }
+
+ @After
+ public void tearDown() {
+ commonConfig.setNodeStatus(originalStatus);
+ commonConfig.setStatusReason(originalStatusReason);
+
commonConfig.setDiskSpaceWarningThreshold(originalDiskSpaceWarningThreshold);
+ dataNodeConfig.setDataNodeId(originalDataNodeId);
+ }
+
+ @Test
+ public void testRunningRecoveryUsesAggregateDiskRatio() {
+ SystemMetrics systemMetrics = mock(SystemMetrics.class);
+ // The aggregate free ratio is 52%, so the node can recover to Running.
+ when(systemMetrics.getSystemDiskAvailableSpace()).thenReturn(104L);
+ when(systemMetrics.getSystemDiskTotalSpace()).thenReturn(200L);
+
+ DataNodeContext dataNodeContext = mock(DataNodeContext.class);
+ DataNodeInternalRPCServiceImpl service =
+ new DataNodeInternalRPCServiceImpl(dataNodeContext, systemMetrics);
+ TLoadSample loadSample = new TLoadSample();
+
+ service.sampleDiskLoad(loadSample);
+
+ Assert.assertEquals(NodeStatus.Running, commonConfig.getNodeStatus());
+ Assert.assertNull(commonConfig.getStatusReason());
+ Assert.assertEquals(104.0, loadSample.getFreeDiskSpace(), 0.0);
+ Assert.assertEquals(0.48, loadSample.getDiskUsageRate(), 1e-10);
+ }
+
+ @Test
+ public void testStorageEngineDiskAggregateStillEntersReadOnly() {
+ SystemMetrics systemMetrics = mock(SystemMetrics.class);
+ when(systemMetrics.getSystemDiskAvailableSpace()).thenReturn(4L);
+ when(systemMetrics.getSystemDiskTotalSpace()).thenReturn(100L);
+
+ commonConfig.setNodeStatus(NodeStatus.Running);
+ commonConfig.setStatusReason(null);
+ DataNodeContext dataNodeContext = mock(DataNodeContext.class);
+ DataNodeInternalRPCServiceImpl service =
+ new DataNodeInternalRPCServiceImpl(dataNodeContext, systemMetrics);
+
+ service.sampleDiskLoad(new TLoadSample());
+
+ Assert.assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus());
+ Assert.assertEquals(NodeStatus.DISK_FULL, commonConfig.getStatusReason());
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManagerTest.java
index 52f13c16e64..44d8b5816c3 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/FolderManagerTest.java
@@ -19,6 +19,9 @@
package org.apache.iotdb.db.storageengine.rescon.disk;
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.conf.CommonConfig;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.disk.FolderManager;
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException;
@@ -37,8 +40,10 @@ import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -186,4 +191,24 @@ public class FolderManagerTest {
fail("Should have thrown DiskSpaceInsufficientException");
}
}
+
+ @Test
+ public void testFolderManagerCanSkipNodeStatusChangeWhenDiskFull() {
+ CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig();
+ NodeStatus originalStatus = commonConfig.getNodeStatus();
+ String originalStatusReason = commonConfig.getStatusReason();
+ commonConfig.setNodeStatus(NodeStatus.Running);
+ commonConfig.setStatusReason(null);
+
+ try {
+ new FolderManager(Collections.emptyList(), strategyType, false);
+ fail("Expected DiskSpaceInsufficientException");
+ } catch (DiskSpaceInsufficientException e) {
+ assertEquals(NodeStatus.Running, commonConfig.getNodeStatus());
+ assertEquals(null, commonConfig.getStatusReason());
+ } finally {
+ commonConfig.setNodeStatus(originalStatus);
+ commonConfig.setStatusReason(originalStatusReason);
+ }
+ }
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 4adacb0d7aa..2b756199413 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -836,7 +836,11 @@ public class CommonConfig {
return status;
}
- public void setNodeStatus(NodeStatus newStatus) {
+ public synchronized void setNodeStatus(NodeStatus newStatus) {
+ if (status == newStatus) {
+ return;
+ }
+
logger.info(ConfigMessages.SET_SYSTEM_MODE, status, newStatus);
this.status = newStatus;
this.statusReason = null;
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 a7707077f84..93ee02a2c15 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
@@ -65,9 +65,18 @@ public class FolderManager {
private final DirectoryStrategy selectStrategy;
+ private final boolean changeSystemStatusToReadOnly;
+
public FolderManager(List<String> folders, DirectoryStrategyType type)
throws DiskSpaceInsufficientException {
+ this(folders, type, true);
+ }
+
+ public FolderManager(
+ List<String> folders, DirectoryStrategyType type, boolean
changeSystemStatusToReadOnly)
+ throws DiskSpaceInsufficientException {
this.folders = folders;
+ this.changeSystemStatusToReadOnly = changeSystemStatusToReadOnly;
folders.forEach(dir -> foldersStates.put(dir, FolderState.HEALTHY));
switch (type) {
case SEQUENCE_STRATEGY:
@@ -85,6 +94,7 @@ public class FolderManager {
default:
throw new RuntimeException();
}
+
this.selectStrategy.setChangeSystemStatusToReadOnly(changeSystemStatusToReadOnly);
try {
this.selectStrategy.setFolders(folders);
this.selectStrategy.setFoldersStates(foldersStates);
@@ -124,6 +134,10 @@ public class FolderManager {
}
private void changeToReadOnlyIfDiskFull(DiskSpaceInsufficientException e) {
+ if (!changeSystemStatusToReadOnly) {
+ return;
+ }
+
if (!hasFolderWithAvailableDiskSpace()) {
if
(LoggerPeriodicalLogReducer.shouldLog(UtilMessages.ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY))
{
logger.error(UtilMessages.ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY, e);
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/strategy/DirectoryStrategy.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/strategy/DirectoryStrategy.java
index e8f53ffe4c2..44c4a93749e 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/strategy/DirectoryStrategy.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/strategy/DirectoryStrategy.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.disk.FolderManager;
import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException;
import org.apache.iotdb.commons.i18n.UtilMessages;
+import org.apache.iotdb.commons.log.LoggerPeriodicalLogReducer;
import org.apache.iotdb.commons.utils.JVMCommonUtils;
import org.slf4j.Logger;
@@ -46,6 +47,12 @@ public abstract class DirectoryStrategy {
/** All the folders of data files, should be init once the subclass is
created. */
List<String> folders = new ArrayList<>();
+ private boolean changeSystemStatusToReadOnly = true;
+
+ public void setChangeSystemStatusToReadOnly(boolean
changeSystemStatusToReadOnly) {
+ this.changeSystemStatusToReadOnly = changeSystemStatusToReadOnly;
+ }
+
/**
* To init folders. Do not recommend to overwrite. This method guarantees
that at least one folder
* has available space.
@@ -61,8 +68,15 @@ public abstract class DirectoryStrategy {
}
}
if (!hasSpace) {
- LOGGER.error(UtilMessages.DISK_SPACE_INSUFFICIENT_READ_ONLY);
-
CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly);
+ if (changeSystemStatusToReadOnly) {
+ LOGGER.error(UtilMessages.DISK_SPACE_INSUFFICIENT_READ_ONLY);
+
CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly);
+ } else {
+ if (LoggerPeriodicalLogReducer.shouldLog(
+ UtilMessages.MESSAGE_DISK_SPACE_INSUFFICIENT_DF6205B0)) {
+ LOGGER.error(UtilMessages.MESSAGE_DISK_SPACE_INSUFFICIENT_DF6205B0);
+ }
+ }
throw new DiskSpaceInsufficientException(folders);
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/conf/CommonConfigTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/conf/CommonConfigTest.java
new file mode 100644
index 00000000000..ac3a001175a
--- /dev/null
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/conf/CommonConfigTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.conf;
+
+import org.apache.iotdb.commons.cluster.NodeStatus;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class CommonConfigTest {
+
+ @Test
+ public void testSameNodeStatusDoesNotClearStatusReason() {
+ CommonConfig config = new CommonConfig();
+ config.setNodeStatus(NodeStatus.ReadOnly);
+ config.setStatusReason(NodeStatus.DISK_FULL);
+
+ config.setNodeStatus(NodeStatus.ReadOnly);
+
+ Assert.assertEquals(NodeStatus.ReadOnly, config.getNodeStatus());
+ Assert.assertEquals(NodeStatus.DISK_FULL, config.getStatusReason());
+ }
+}
