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 9303e4c01f41c..3e465ced4294f 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 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 8257dadfc9992..e32236d2c2510 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 @@ -205,7 +205,9 @@ private enum TreeDatabaseCreationResult { 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 cde5b09b4578f..33ade6234dcd1 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.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; @@ -2537,23 +2545,9 @@ private double getMemory(String gaugeName) { 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 0000000000000..8393b2fd4e4be --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImplDiskTest.java @@ -0,0 +1,109 @@ +/* + * 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 testPipeReceiverDiskDoesNotBlockRunningRecovery() { + SystemMetrics systemMetrics = mock(SystemMetrics.class); + // The storage-engine disks have an aggregate free ratio of 52%. A full Pipe receiver disk is + // handled by the receiver itself and must not prevent the node from recovering 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 52f13c16e6491..8b12f8fb8d7c2 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.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 void testEventuallyThrowsDiskFullAfterRetries() { fail("Should have thrown DiskSpaceInsufficientException"); } } + + @Test + public void testPipeFolderManagerDoesNotChangeNodeStatusWhenDiskFull() { + 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 4adacb0d7aa1c..2b756199413e2 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 NodeStatus getNodeStatus() { 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 a7707077f84dd..93ee02a2c15d2 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 enum FolderState { private final DirectoryStrategy selectStrategy; + private final boolean changeSystemStatusToReadOnly; + public FolderManager(List folders, DirectoryStrategyType type) throws DiskSpaceInsufficientException { + this(folders, type, true); + } + + public FolderManager( + List 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 FolderManager(List folders, DirectoryStrategyType type) default: throw new RuntimeException(); } + this.selectStrategy.setChangeSystemStatusToReadOnly(changeSystemStatusToReadOnly); try { this.selectStrategy.setFolders(folders); this.selectStrategy.setFoldersStates(foldersStates); @@ -124,6 +134,10 @@ private boolean hasFolderWithAvailableDiskSpace() { } 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 e8f53ffe4c20f..654f18d1ab151 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 @@ -46,6 +46,12 @@ public abstract class DirectoryStrategy { /** All the folders of data files, should be init once the subclass is created. */ List 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 +67,12 @@ public void setFolders(List folders) throws DiskSpaceInsufficientExcepti } } 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 { + 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 0000000000000..ac3a001175afa --- /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()); + } +}