diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDiskFullFolderRecoveryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDiskFullFolderRecoveryIT.java new file mode 100644 index 0000000000000..437df878f951e --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDiskFullFolderRecoveryIT.java @@ -0,0 +1,158 @@ +/* + * 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.it; + +import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; + +import org.awaitility.Awaitility; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.File; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +import static org.apache.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class}) +public class IoTDBDiskFullFolderRecoveryIT { + private static final String DATABASE = "root.disk_full_folder_recovery"; + private static final String FULL_THRESHOLD = "0.9999999999999999"; + private static final String NORMAL_THRESHOLD = "0.05"; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testNewTsFileAfterFullDirectoryReloadAndRecovery() throws Exception { + int dataNodeId = EnvFactory.getEnv().getConfigNodeWrapperList().size(); + File dataDirectory = new File(EnvFactory.getEnv().getDataNodeWrapper(0).getDataPath()); + Assert.assertTrue(dataDirectory.mkdirs() || dataDirectory.isDirectory()); + Assert.assertTrue(dataDirectory.getTotalSpace() > 0); + Assert.assertTrue( + (double) dataDirectory.getFreeSpace() / dataDirectory.getTotalSpace() + < Double.parseDouble(FULL_THRESHOLD)); + + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE " + DATABASE); + statement.execute("CREATE TIMESERIES " + DATABASE + ".d1.s1 WITH DATATYPE=INT64"); + try { + // The first reload installs the threshold; the second rebuilds FolderManagers under it. + setDataNodeConfiguration( + statement, dataNodeId, "disk_space_warning_threshold", FULL_THRESHOLD); + setDataNodeConfiguration(statement, dataNodeId, "enable_partial_insert", "true"); + awaitDataNodeStatus(statement, dataNodeId, "ReadOnly", "DiskFull"); + + // A successful reload of the same configuration must not clear the full-disk status. + setDataNodeConfiguration(statement, dataNodeId, "enable_partial_insert", "true"); + Assert.assertEquals("ReadOnly", getDataNodeStatus(statement, dataNodeId)); + } finally { + setDataNodeConfiguration( + statement, dataNodeId, "disk_space_warning_threshold", NORMAL_THRESHOLD); + } + + awaitDataNodeStatus(statement, dataNodeId, "Running", null); + statement.execute("INSERT INTO " + DATABASE + ".d1(time,s1) VALUES(1,1)"); + statement.execute("FLUSH"); + + try (ResultSet resultSet = statement.executeQuery("SELECT s1 FROM " + DATABASE + ".d1")) { + Assert.assertTrue(resultSet.next()); + Assert.assertEquals(1, resultSet.getLong("Time")); + Assert.assertEquals(1, resultSet.getLong(DATABASE + ".d1.s1")); + Assert.assertFalse(resultSet.next()); + } + Assert.assertTrue(hasSequenceTsFile()); + } + } + + private static void setDataNodeConfiguration( + Statement statement, int nodeId, String key, String value) throws Exception { + statement.execute("SET CONFIGURATION \"" + key + "\"=\"" + value + "\" ON " + nodeId); + } + + private static void awaitDataNodeStatus( + Statement statement, int nodeId, String expectedStatus, String expectedReason) { + Awaitility.await() + .atMost(30, TimeUnit.SECONDS) + .until( + () -> { + try (ResultSet resultSet = statement.executeQuery("SHOW DATANODES")) { + while (resultSet.next()) { + if (resultSet.getInt("NodeID") == nodeId) { + if (!expectedStatus.equals(resultSet.getString("Status"))) { + return false; + } + return expectedReason == null + || expectedReason.equals(resultSet.getString("StatusReason")); + } + } + } + return false; + }); + } + + private static String getDataNodeStatus(Statement statement, int nodeId) throws Exception { + try (ResultSet resultSet = statement.executeQuery("SHOW DATANODES")) { + while (resultSet.next()) { + if (resultSet.getInt("NodeID") == nodeId) { + return resultSet.getString("Status"); + } + } + } + Assert.fail("DataNode not found: " + nodeId); + return null; + } + + private static boolean hasSequenceTsFile() throws Exception { + Path sequenceDirectory = + Paths.get( + EnvFactory.getEnv().getDataNodeWrapper(0).getDataPath(), + IoTDBConstant.SEQUENCE_FOLDER_NAME, + DATABASE); + if (!Files.exists(sequenceDirectory)) { + return false; + } + try (Stream paths = Files.walk(sequenceDirectory)) { + return paths.anyMatch(path -> path.toString().endsWith(TSFILE_SUFFIX)); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index 08db154326ba2..04fe3381b7541 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -2517,7 +2517,6 @@ public synchronized void loadHotModifiedProps() throws QueryProcessException { return; } - boolean wasReadOnlyForDiskFull = commonConfig.isReadOnlyForDiskFull(); TrimProperties commonProperties = new TrimProperties(); try (InputStream inputStream = url.openStream()) { LOGGER.info(DataNodeMiscMessages.START_RELOAD_CONFIG_FILE, url); @@ -2535,9 +2534,6 @@ public synchronized void loadHotModifiedProps() throws QueryProcessException { ConfigurationFileUtils.releaseDefault(); } reloadMetricProperties(commonProperties); - if (wasReadOnlyForDiskFull) { - commonConfig.setRunningIfDiskFull(); - } } public void reloadMetricProperties(TrimProperties properties) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 505ac60a26ccb..6c128c027a441 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -815,8 +815,6 @@ public TSStatus setConfiguration(TSetConfigurationReq req) { return tsStatus; } - CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); - boolean wasReadOnlyForDiskFull = commonConfig.isReadOnlyForDiskFull(); try { ConfigurationFileUtils.updateConfiguration( new File(configFileUrl.getFile()), @@ -834,10 +832,6 @@ public TSStatus setConfiguration(TSetConfigurationReq req) { } return RpcUtils.getStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR, e.getMessage()); } - // A SET can apply in memory and still fail when persisting the configuration file. - if (wasReadOnlyForDiskFull) { - commonConfig.setRunningIfDiskFull(); - } return tsStatus; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java index 59bc5877979e4..17047e732470a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.disk.FolderManager; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException; +import org.apache.iotdb.commons.utils.JVMCommonUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.StorageEngineMessages; @@ -58,20 +59,12 @@ public class TierManager { private static final Logger logger = LoggerFactory.getLogger(TierManager.class); private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); - private DirectoryStrategyType directoryStrategyType = DirectoryStrategyType.SEQUENCE_STRATEGY; - /** - * seq folder manager of each storage tier, managing both data directories and multi-dir strategy - */ - private volatile List seqTiers = new ArrayList<>(); + private record TierDirectory(int tierLevel, String folderName) {} - /** - * unSeq folder manager of each storage tier, managing both data directories and multi-dir - * strategy - */ - private volatile List unSeqTiers = new ArrayList<>(); - - private volatile List objectTiers = new ArrayList<>(); + private volatile Map folderManagers = Map.of(); + // A failed constructor can report DiskFull before the replacement managers are published. + private volatile boolean rebuildingFolders; /** seq file folder's rawFsPath path -> tier level */ private volatile Map seqDir2TierLevel = new HashMap<>(); @@ -81,31 +74,28 @@ public class TierManager { private List objectDirs; - private List copyToTargetDirs; - - // private FolderManager copyToFolderManager; - - private MemoizedCheckedSupplier - copyToFolderManager; - /** total space of each tier, Long.MAX_VALUE when one tier contains remote storage */ - private long[] tierDiskTotalSpace; + private volatile long[] tierDiskTotalSpace = new long[0]; private TierManager() { initFolders(); } public synchronized void initFolders() { - initFolders(seqTiers, unSeqTiers, objectTiers, seqDir2TierLevel, unSeqDir2TierLevel); + CommonDescriptor.getInstance().getConfig().invalidateDiskFullRecoverySnapshot(); + rebuildingFolders = true; + try { + rebuildFolders(); + } finally { + rebuildingFolders = false; + } } - private void initFolders( - List seqTiers, - List unSeqTiers, - List objectTiers, - Map seqDir2TierLevel, - Map unSeqDir2TierLevel) { - directoryStrategyType = + private void rebuildFolders() { + Map newManagers = new HashMap<>(); + Map newSeqDir2TierLevel = new HashMap<>(); + Map newUnSeqDir2TierLevel = new HashMap<>(); + DirectoryStrategyType directoryStrategyType = DirectoryStrategyType.fromClassName(config.getMultiDirStrategyClassName()); config.updatePath(); @@ -140,13 +130,14 @@ private void initFolders( .getPath()) .collect(Collectors.toList()); mkDataDirs(seqDirs); - try { - seqTiers.add(new FolderManager(seqDirs, directoryStrategyType)); - } catch (DiskSpaceInsufficientException e) { - logger.error(StorageEngineMessages.ALL_DISKS_OF_TIER_FULL, tierLevel, e); - } + addFolderManager( + newManagers, + tierLevel, + IoTDBConstant.SEQUENCE_FOLDER_NAME, + seqDirs, + directoryStrategyType); for (String dir : seqDirs) { - seqDir2TierLevel.put(dir, tierLevel); + newSeqDir2TierLevel.put(dir, tierLevel); } List unSeqDirs = @@ -159,17 +150,18 @@ private void initFolders( .getPath()) .collect(Collectors.toList()); mkDataDirs(unSeqDirs); - try { - unSeqTiers.add(new FolderManager(unSeqDirs, directoryStrategyType)); - } catch (DiskSpaceInsufficientException e) { - logger.error(StorageEngineMessages.ALL_DISKS_OF_TIER_FULL, tierLevel, e); - } + addFolderManager( + newManagers, + tierLevel, + IoTDBConstant.UNSEQUENCE_FOLDER_NAME, + unSeqDirs, + directoryStrategyType); for (String dir : unSeqDirs) { - unSeqDir2TierLevel.put(dir, tierLevel); + newUnSeqDir2TierLevel.put(dir, tierLevel); } if (tierLevel == 0) { - copyToTargetDirs = + List copyToTargetDirs = Arrays.stream(tierDirs[tierLevel]) .filter(Objects::nonNull) .map( @@ -178,9 +170,12 @@ private void initFolders( .getFile(v, IoTDBConstant.COPY_TO_TARGET_FOLDER_NAME) .getPath()) .collect(Collectors.toList()); - copyToFolderManager = - MemoizedCheckedSupplier.valueOf( - () -> new FolderManager(copyToTargetDirs, directoryStrategyType)); + addFolderManager( + newManagers, + tierLevel, + IoTDBConstant.COPY_TO_TARGET_FOLDER_NAME, + copyToTargetDirs, + directoryStrategyType); } objectDirs = @@ -193,11 +188,12 @@ private void initFolders( .getPath()) .collect(Collectors.toList()); - try { - objectTiers.add(new FolderManager(objectDirs, directoryStrategyType)); - } catch (DiskSpaceInsufficientException e) { - logger.error(StorageEngineMessages.ALL_DISKS_OF_TIER_FULL, tierLevel, e); - } + addFolderManager( + newManagers, + tierLevel, + IoTDBConstant.OBJECT_FOLDER_NAME, + objectDirs, + directoryStrategyType); // try to remove empty objectDirs for (String dir : objectDirs) { File dirFile = FSFactoryProducer.getFSFactory().getFile(dir); @@ -211,23 +207,74 @@ private void initFolders( } tierDiskTotalSpace = getTierDiskSpace(DiskSpaceType.TOTAL); + seqDir2TierLevel = newSeqDir2TierLevel; + unSeqDir2TierLevel = newUnSeqDir2TierLevel; + folderManagers = newManagers; + } + + private void addFolderManager( + Map newManagers, + int tierLevel, + String folderName, + List folders, + DirectoryStrategyType type) { + TierDirectory key = new TierDirectory(tierLevel, folderName); + TierFolderManager entry = new TierFolderManager(key, folders, type); + newManagers.put(key, entry); + if (!IoTDBConstant.COPY_TO_TARGET_FOLDER_NAME.equals(folderName)) { + try { + entry.supplier.get(); + } catch (DiskSpaceInsufficientException e) { + logger.error(StorageEngineMessages.ALL_DISKS_OF_TIER_FULL, tierLevel, e); + } + } + } + + private final class TierFolderManager { + private final List folders; + private final MemoizedCheckedSupplier supplier; + private volatile FolderManager manager; + + private TierFolderManager(TierDirectory key, List folders, DirectoryStrategyType type) { + this.folders = folders; + supplier = + MemoizedCheckedSupplier.valueOf( + () -> { + manager = new FolderManager(folders, type, () -> reportDiskFull(key, this)); + return manager; + }); + } + + private boolean hasSpaceForRecovery() { + FolderManager current = manager; + return current == null + ? folders.stream().anyMatch(JVMCommonUtils::hasSpaceWithoutRetry) + : current.hasSpaceForRecovery(); + } + } + + private void reportDiskFull(TierDirectory key, TierFolderManager source) { + if (!rebuildingFolders && folderManagers.get(key) != source) { + return; + } + CommonDescriptor.getInstance() + .getConfig() + .setReadOnlyForDiskFull(key, () -> hasSpaceForRecovery(key)); + } + + private boolean hasSpaceForRecovery(TierDirectory key) { + Map active = folderManagers; + if (rebuildingFolders) { + return false; + } + TierFolderManager current = active.get(key); + boolean recovered = current == null || current.hasSpaceForRecovery(); + return recovered && active == folderManagers && !rebuildingFolders; } public synchronized void resetFolders() { long startTime = System.currentTimeMillis(); - List newSeqTiers = new ArrayList<>(); - List newUnSeqTiers = new ArrayList<>(); - List newObjectTiers = new ArrayList<>(); - Map newSeqDir2TierLevel = new HashMap<>(); - Map newUnSeqDir2TierLevel = new HashMap<>(); - - initFolders( - newSeqTiers, newUnSeqTiers, newObjectTiers, newSeqDir2TierLevel, newUnSeqDir2TierLevel); - seqTiers = newSeqTiers; - unSeqTiers = newUnSeqTiers; - objectTiers = newObjectTiers; - seqDir2TierLevel = newSeqDir2TierLevel; - unSeqDir2TierLevel = newUnSeqDir2TierLevel; + initFolders(); long endTime = System.currentTimeMillis(); logger.info(StorageEngineMessages.FOLDERS_RESET_SUCCESSFULLY, (endTime - startTime)); } @@ -251,21 +298,31 @@ private void mkDataDirs(List folders) { public String getNextFolderForTsFile(int tierLevel, boolean sequence) throws DiskSpaceInsufficientException { - return sequence - ? seqTiers.get(tierLevel).getNextFolder() - : unSeqTiers.get(tierLevel).getNextFolder(); + return getFolderManager(tierLevel, sequence).getNextFolder(); } public String getNextFolderForCopyToTargetFile() throws DiskSpaceInsufficientException { - return copyToFolderManager.get().getNextFolder(); + return getFolderManager(0, IoTDBConstant.COPY_TO_TARGET_FOLDER_NAME).getNextFolder(); } public String getNextFolderForObjectFile() throws DiskSpaceInsufficientException { - return objectTiers.get(0).getNextFolder(); + return getFolderManager(0, IoTDBConstant.OBJECT_FOLDER_NAME).getNextFolder(); + } + + public FolderManager getFolderManager(int tierLevel, boolean sequence) + throws DiskSpaceInsufficientException { + String folderName = + sequence ? IoTDBConstant.SEQUENCE_FOLDER_NAME : IoTDBConstant.UNSEQUENCE_FOLDER_NAME; + return getFolderManager(tierLevel, folderName); } - public FolderManager getFolderManager(int tierLevel, boolean sequence) { - return sequence ? seqTiers.get(tierLevel) : unSeqTiers.get(tierLevel); + private FolderManager getFolderManager(int tierLevel, String folderName) + throws DiskSpaceInsufficientException { + TierFolderManager entry = folderManagers.get(new TierDirectory(tierLevel, folderName)); + if (entry == null) { + throw new IndexOutOfBoundsException(tierLevel); + } + return entry.supplier.get(); } public List getAllFilesFolders() { @@ -356,7 +413,7 @@ public List getAllMatchedObjectDirs(String regionIdStr, String... path) { } public int getTiersNum() { - return seqTiers.size(); + return tierDiskTotalSpace.length; } public int getFileTierLevel(File file) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DiskFullHotReloadTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DiskFullHotReloadTest.java index 38e1a44d5512f..620f7e1ec7e93 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DiskFullHotReloadTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/DiskFullHotReloadTest.java @@ -39,7 +39,6 @@ import java.util.Map; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; public class DiskFullHotReloadTest { @@ -74,20 +73,20 @@ public void tearDown() { } @Test - public void loadSameConfigurationClearsOldDiskFullChecks() throws Exception { + public void loadSameConfigurationKeepsUnrecoveredDiskFullChecks() throws Exception { config.setReadOnlyForDiskFull("old directory", () -> false); descriptor.loadHotModifiedProps(); - assertRecoveredAndCleared(); + assertStillProtected(); } @Test - public void setUnrelatedPropertyClearsOldDiskFullChecks() { + public void setUnrelatedPropertyKeepsUnrecoveredDiskFullChecks() { config.setReadOnlyForDiskFull("old directory", () -> false); Map update = Map.of("enable_partial_insert", Boolean.toString(!originalPartialInsert)); assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), setConfiguration(update)); assertEquals(!originalPartialInsert, descriptor.getConfig().isEnablePartialInsert()); - assertRecoveredAndCleared(); + assertStillProtected(); } @Test @@ -131,14 +130,10 @@ private int setConfiguration(Map items) { .getCode(); } - private void assertRecoveredAndCleared() { - assertEquals(NodeStatus.Running, config.getNodeStatus()); - assertNull(config.getStatusReason()); - // A later aggregate-only failure must have no callback left from the previous failure. - config.setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.DISK_FULL); + private void assertStillProtected() { + assertReadOnly(NodeStatus.DISK_FULL); config.tryRecoverFromDiskFull(); - assertEquals(NodeStatus.Running, config.getNodeStatus()); - assertNull(config.getStatusReason()); + assertReadOnly(NodeStatus.DISK_FULL); } private void assertReadOnly(String reason) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerDiskFullReloadTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerDiskFullReloadTest.java new file mode 100644 index 0000000000000..9fd46374039d8 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerDiskFullReloadTest.java @@ -0,0 +1,120 @@ +/* + * 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.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.conf.IoTDBConstant; +import org.apache.iotdb.commons.utils.JVMCommonUtils; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PowerMockIgnore; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; + +import java.io.File; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.Assert.assertEquals; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.when; + +@RunWith(PowerMockRunner.class) +@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) +@PrepareForTest(JVMCommonUtils.class) +public class TierManagerDiskFullReloadTest { + + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + private String[][] originalTierDataDirs; + private NodeStatus originalStatus; + private String originalReason; + private TierManager tierManager; + private File oldDir; + private File newDir; + + @Before + public void setUp() { + originalTierDataDirs = config.getTierDataDirs(); + originalStatus = commonConfig.getNodeStatus(); + originalReason = commonConfig.getStatusReason(); + oldDir = new File(temporaryFolder.getRoot(), "old"); + newDir = new File(temporaryFolder.getRoot(), "new"); + config.setTierDataDirs(new String[][] {{oldDir.getAbsolutePath()}}); + tierManager = TierManager.getInstance(); + commonConfig.setNodeStatus(NodeStatus.Running); + PowerMockito.mockStatic(JVMCommonUtils.class); + } + + @After + public void tearDown() { + when(JVMCommonUtils.hasSpace(anyString())).thenReturn(true); + config.setTierDataDirs(originalTierDataDirs); + tierManager.resetFolders(); + commonConfig.setNodeStatus(NodeStatus.Running); + commonConfig.setNodeStatusWithReason(originalStatus, originalReason); + } + + @Test + public void testNewDataDirectoryDoesNotClearAnotherFullSource() throws Exception { + tierManager.resetFolders(); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + + tierManager.resetFolders(); + commonConfig.tryRecoverFromDiskFull(); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + + AtomicBoolean walAvailable = new AtomicBoolean(); + commonConfig.setReadOnlyForDiskFull("wal", walAvailable::get); + config.setTierDataDirs(new String[][] {{oldDir.getAbsolutePath(), newDir.getAbsolutePath()}}); + for (String folderName : + new String[] { + IoTDBConstant.SEQUENCE_FOLDER_NAME, + IoTDBConstant.UNSEQUENCE_FOLDER_NAME, + IoTDBConstant.OBJECT_FOLDER_NAME + }) { + String folder = new File(newDir, folderName).getCanonicalPath(); + when(JVMCommonUtils.hasSpace(folder)).thenReturn(true); + when(JVMCommonUtils.hasSpaceWithoutRetry(folder)).thenReturn(true); + } + tierManager.resetFolders(); + + commonConfig.tryRecoverFromDiskFull(); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, commonConfig.getStatusReason()); + + walAvailable.set(true); + commonConfig.tryRecoverFromDiskFull(); + assertEquals(NodeStatus.Running, commonConfig.getNodeStatus()); + assertEquals( + new File(newDir, IoTDBConstant.SEQUENCE_FOLDER_NAME).getCanonicalPath(), + tierManager.getNextFolderForTsFile(0, true)); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java index 046e41dce4875..8e5b5575d37dc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java @@ -19,7 +19,12 @@ 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.conf.IoTDBConstant; +import org.apache.iotdb.commons.exception.DiskSpaceInsufficientException; +import org.apache.iotdb.commons.utils.JVMCommonUtils; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; @@ -42,7 +47,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class TierManagerTest { @@ -149,6 +156,44 @@ public void testConcurrentResetFoldersAndGetNextFolder() throws Exception { } } + @Test + public void testInitiallyFullTiersRecoverWithoutLosingTierLevels() throws Exception { + CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + NodeStatus originalStatus = commonConfig.getNodeStatus(); + String originalReason = commonConfig.getStatusReason(); + double originalThreshold = commonConfig.getDiskSpaceWarningThreshold(); + String tier0 = new File(temporaryFolder.getRoot(), "tier0").getAbsolutePath(); + String tier1 = new File(temporaryFolder.getRoot(), "tier1").getAbsolutePath(); + + try { + config.setTierDataDirs(new String[][] {{tier0}, {tier1}}); + commonConfig.setNodeStatus(NodeStatus.Running); + JVMCommonUtils.setDiskSpaceWarningThreshold(1.0); + tierManager.resetFolders(); + + assertEquals(2, tierManager.getTiersNum()); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertThrows( + DiskSpaceInsufficientException.class, () -> tierManager.getFolderManager(0, true)); + assertThrows( + DiskSpaceInsufficientException.class, () -> tierManager.getFolderManager(1, false)); + + JVMCommonUtils.setDiskSpaceWarningThreshold(0.0); + commonConfig.tryRecoverFromDiskFull(); + assertEquals(NodeStatus.Running, commonConfig.getNodeStatus()); + assertEquals( + new File(tier0, IoTDBConstant.SEQUENCE_FOLDER_NAME).getCanonicalPath(), + tierManager.getNextFolderForTsFile(0, true)); + assertEquals( + new File(tier1, IoTDBConstant.UNSEQUENCE_FOLDER_NAME).getCanonicalPath(), + tierManager.getNextFolderForTsFile(1, false)); + } finally { + JVMCommonUtils.setDiskSpaceWarningThreshold(originalThreshold); + commonConfig.setNodeStatus(NodeStatus.Running); + commonConfig.setNodeStatusWithReason(originalStatus, originalReason); + } + } + private List assertFoldersAvailable(List folders) { assertTrue(folders.size() >= DATA_DIR_NUM); return folders; 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 f3ba8c27b1060..d2a92a958f46e 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 @@ -189,6 +189,12 @@ public class CommonConfig { // a live FolderManager uses its identity so its health restrictions are also checked. private final Map diskFullRecoveryChecks = new HashMap<>(); + /** + * Invalidates a recovery check that was evaluated before another disk failure or directory + * reload. + */ + private long diskFullRecoveryVersion; + private final int TTimePartitionSlotTransmitLimit = 1000; /** Disk Monitor. */ @@ -853,29 +859,32 @@ public synchronized void setReadOnlyForDiskFull(Object source, BooleanSupplier r setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.DISK_FULL); } - public synchronized boolean isReadOnlyForDiskFull() { - return status == NodeStatus.ReadOnly && NodeStatus.DISK_FULL.equals(statusReason); + public synchronized void invalidateDiskFullRecoverySnapshot() { + diskFullRecoveryVersion++; } - /** Recheck the reason under the status lock before clearing disk-full recovery checks. */ - public synchronized void setRunningIfDiskFull() { - if (isReadOnlyForDiskFull()) { - setNodeStatus(NodeStatus.Running); - } + public synchronized boolean isReadOnlyForDiskFull() { + return status == NodeStatus.ReadOnly && NodeStatus.DISK_FULL.equals(statusReason); } /** Called after the aggregate disk-space check passes. */ public void tryRecoverFromDiskFull() { final List recoveryChecks; + final long recoveryVersion; synchronized (this) { if (!isReadOnlyForDiskFull()) { return; } recoveryChecks = new ArrayList<>(diskFullRecoveryChecks.values()); + recoveryVersion = diskFullRecoveryVersion; } // Do not hold the status monitor during filesystem I/O. if (recoveryChecks.stream().allMatch(BooleanSupplier::getAsBoolean)) { - setRunningIfDiskFull(); + synchronized (this) { + if (recoveryVersion == diskFullRecoveryVersion && isReadOnlyForDiskFull()) { + setNodeStatus(NodeStatus.Running); + } + } } } @@ -885,6 +894,9 @@ public void tryRecoverFromDiskFull() { * reason. */ public synchronized void setNodeStatusWithReason(NodeStatus newStatus, String newReason) { + if (newStatus == NodeStatus.ReadOnly && NodeStatus.DISK_FULL.equals(newReason)) { + diskFullRecoveryVersion++; + } if (newStatus != NodeStatus.ReadOnly) { diskFullRecoveryChecks.clear(); newReason = 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 1cb969043e9d9..4dee48cc47cb6 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 @@ -19,7 +19,6 @@ package org.apache.iotdb.commons.disk; -import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategy; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; import org.apache.iotdb.commons.disk.strategy.MaxDiskUsableSpaceFirstStrategy; @@ -75,6 +74,21 @@ public FolderManager(List folders, DirectoryStrategyType type) public FolderManager( List folders, DirectoryStrategyType type, boolean changeSystemStatusToReadOnly) throws DiskSpaceInsufficientException { + this(folders, type, changeSystemStatusToReadOnly, null); + } + + /** For an owner that replaces this manager on configuration reload. */ + public FolderManager(List folders, DirectoryStrategyType type, Runnable diskFullHandler) + throws DiskSpaceInsufficientException { + this(folders, type, true, diskFullHandler); + } + + private FolderManager( + List folders, + DirectoryStrategyType type, + boolean changeSystemStatusToReadOnly, + Runnable diskFullHandler) + throws DiskSpaceInsufficientException { this.folders = folders; this.changeSystemStatusToReadOnly = changeSystemStatusToReadOnly; folders.forEach(dir -> foldersStates.put(dir, FolderState.HEALTHY)); @@ -95,6 +109,7 @@ public FolderManager( throw new RuntimeException(); } this.selectStrategy.setChangeSystemStatusToReadOnly(changeSystemStatusToReadOnly); + this.selectStrategy.setDiskFullHandler(diskFullHandler); this.selectStrategy.setFolders(folders); this.selectStrategy.setFoldersStates(foldersStates); } @@ -128,6 +143,10 @@ private boolean hasFolderWithAvailableDiskSpace(Predicate hasSpace) { && hasSpace.test(folder)); } + public boolean hasSpaceForRecovery() { + return hasFolderWithAvailableDiskSpace(JVMCommonUtils::hasSpaceWithoutRetry); + } + private void changeToReadOnlyIfDiskFull(DiskSpaceInsufficientException e) { if (!changeSystemStatusToReadOnly) { return; @@ -137,10 +156,7 @@ private void changeToReadOnlyIfDiskFull(DiskSpaceInsufficientException e) { if (LoggerPeriodicalLogReducer.shouldLog(UtilMessages.ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY)) { logger.error(UtilMessages.ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY, e); } - CommonDescriptor.getInstance() - .getConfig() - .setReadOnlyForDiskFull( - this, () -> hasFolderWithAvailableDiskSpace(JVMCommonUtils::hasSpaceWithoutRetry)); + selectStrategy.reportDiskFull(this, this::hasSpaceForRecovery); } else { logger.warn(UtilMessages.CANNOT_SELECT_FOLDER_BUT_DISK_HAS_SPACE, 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 a592766ec7884..4e1072d07d462 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 @@ -32,6 +32,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.function.BooleanSupplier; import static org.apache.iotdb.commons.disk.FolderManager.FolderState.HEALTHY; @@ -48,6 +49,20 @@ public abstract class DirectoryStrategy { private boolean changeSystemStatusToReadOnly = true; + private Runnable diskFullHandler; + + public void setDiskFullHandler(Runnable diskFullHandler) { + this.diskFullHandler = diskFullHandler; + } + + public void reportDiskFull(Object source, BooleanSupplier recoveryCheck) { + if (diskFullHandler != null) { + diskFullHandler.run(); + } else { + CommonDescriptor.getInstance().getConfig().setReadOnlyForDiskFull(source, recoveryCheck); + } + } + public void setChangeSystemStatusToReadOnly(boolean changeSystemStatusToReadOnly) { this.changeSystemStatusToReadOnly = changeSystemStatusToReadOnly; } @@ -70,11 +85,8 @@ public void setFolders(List folders) throws DiskSpaceInsufficientExcepti if (changeSystemStatusToReadOnly) { LOGGER.error(UtilMessages.DISK_SPACE_INSUFFICIENT_READ_ONLY); final List fullFolders = List.copyOf(folders); - CommonDescriptor.getInstance() - .getConfig() - .setReadOnlyForDiskFull( - fullFolders, - () -> fullFolders.stream().anyMatch(JVMCommonUtils::hasSpaceWithoutRetry)); + reportDiskFull( + fullFolders, () -> fullFolders.stream().anyMatch(JVMCommonUtils::hasSpaceWithoutRetry)); } else { if (LoggerPeriodicalLogReducer.shouldLog( UtilMessages.MESSAGE_DISK_SPACE_INSUFFICIENT_DF6205B0)) { 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 index 494db7d94ed89..95fc05351b7fd 100644 --- 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 @@ -254,6 +254,50 @@ public void testDiskRecoveryCannotClearConcurrentHigherPriorityReason() { } } + @Test + public void testDiskRecoveryCannotClearNewFailureDuringRecoveryCheck() { + CommonConfig config = new CommonConfig(); + AtomicBoolean reportedNewFailure = new AtomicBoolean(); + AtomicBoolean newSourceAvailable = new AtomicBoolean(); + config.setReadOnlyForDiskFull( + "first", + () -> { + if (reportedNewFailure.compareAndSet(false, true)) { + config.setReadOnlyForDiskFull("second", newSourceAvailable::get); + } + return true; + }); + + config.tryRecoverFromDiskFull(); + Assert.assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + Assert.assertEquals(NodeStatus.DISK_FULL, config.getStatusReason()); + + newSourceAvailable.set(true); + config.tryRecoverFromDiskFull(); + Assert.assertEquals(NodeStatus.Running, config.getNodeStatus()); + } + + @Test + public void testDiskRecoveryCannotClearDirectoryReloadDuringRecoveryCheck() { + CommonConfig config = new CommonConfig(); + AtomicBoolean reloaded = new AtomicBoolean(); + config.setReadOnlyForDiskFull( + "tier", + () -> { + if (reloaded.compareAndSet(false, true)) { + config.invalidateDiskFullRecoverySnapshot(); + } + return true; + }); + + config.tryRecoverFromDiskFull(); + Assert.assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + Assert.assertEquals(NodeStatus.DISK_FULL, config.getStatusReason()); + + config.tryRecoverFromDiskFull(); + Assert.assertEquals(NodeStatus.Running, config.getNodeStatus()); + } + @Test public void testManualRunningClearsPreviousDiskRecoveryChecks() { CommonConfig config = new CommonConfig();