diff --git a/CHANGES.txt b/CHANGES.txt index 9a575f7f02..ed6af40d08 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,6 +3,7 @@ * Update cassandra-stress to support TLS 1.3 by default by auto-negotiation (CASSANDRA-21007) * Ensure schema created before 2.1 without tableId in folder name can be loaded in SnapshotLoader (CASSANDRA-21173) Merged from 4.1: + * Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290) Merged from 4.0: Backported from 6.0: * Improved observability in AutoRepair to report both expected vs. actual repair bytes and expected vs. actual keyspaces (CASSANDRA-20581) diff --git a/src/java/org/apache/cassandra/service/DataResurrectionCheck.java b/src/java/org/apache/cassandra/service/DataResurrectionCheck.java index 4cf3278110..7a4389f748 100644 --- a/src/java/org/apache/cassandra/service/DataResurrectionCheck.java +++ b/src/java/org/apache/cassandra/service/DataResurrectionCheck.java @@ -19,6 +19,7 @@ package org.apache.cassandra.service; import java.io.IOException; +import java.nio.file.Files; import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; @@ -45,6 +46,7 @@ import org.apache.cassandra.io.util.File; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.utils.Clock; +import org.apache.cassandra.utils.Hex; import org.apache.cassandra.utils.JsonUtils; import org.apache.cassandra.utils.Pair; @@ -53,7 +55,6 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.stream.Collectors.joining; import static java.util.stream.Collectors.toList; import static java.util.stream.Collectors.toSet; -import static org.apache.cassandra.exceptions.StartupException.ERR_WRONG_DISK_STATE; import static org.apache.cassandra.exceptions.StartupException.ERR_WRONG_MACHINE_STATE; import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis; @@ -86,12 +87,24 @@ public class DataResurrectionCheck implements StartupCheck public void serializeToJsonFile(File outputFile) throws IOException { - JsonUtils.serializeToJsonFile(this, outputFile); + JsonUtils.serializeToJsonFileAtomic(this, outputFile); } public static Heartbeat deserializeFromJsonFile(File file) throws IOException { - return JsonUtils.deserializeFromJsonFile(Heartbeat.class, file); + byte[] bytes = Files.readAllBytes(file.toPath()); + try + { + return JsonUtils.deserializeFromJsonBytes(Heartbeat.class, bytes); + } + catch (IOException ex) + { + int maxLogBytes = Math.min(bytes.length, 1024); + String hexContent = bytes.length > 0 ? Hex.bytesToHex(bytes, 0, maxLogBytes) : "(empty)"; + LOGGER.error("Failed to deserialize heartbeat file {} (length: {} bytes, first {} bytes hex: {})", + file, bytes.length, maxLogBytes, hexContent, ex); + throw ex; + } } @Override @@ -173,7 +186,10 @@ public class DataResurrectionCheck implements StartupCheck } catch (IOException ex) { - throw new StartupException(ERR_WRONG_DISK_STATE, "Failed to deserialize heartbeat file " + heartbeatFile); + LOGGER.warn("Failed to deserialize heartbeat file {}. Falling back to file last modified time.", + heartbeatFile, ex); + Instant lastModified = Instant.ofEpochMilli(heartbeatFile.lastModified()); + heartbeat = new Heartbeat(lastModified); } if (heartbeat.lastHeartbeat == null) @@ -297,7 +313,7 @@ public class DataResurrectionCheck implements StartupCheck List getTablesGcPeriods(String userKeyspace) { Optional keyspaceMetadata = SchemaKeyspace.fetchNonSystemKeyspaces().get(userKeyspace); - if (!keyspaceMetadata.isPresent()) + if (keyspaceMetadata.isEmpty()) return Collections.emptyList(); KeyspaceMetadata ksmd = keyspaceMetadata.get(); diff --git a/src/java/org/apache/cassandra/utils/JsonUtils.java b/src/java/org/apache/cassandra/utils/JsonUtils.java index 1cdc13c55c..489cfa95d9 100644 --- a/src/java/org/apache/cassandra/utils/JsonUtils.java +++ b/src/java/org/apache/cassandra/utils/JsonUtils.java @@ -180,6 +180,38 @@ public final class JsonUtils } } + public static void serializeToJsonFileAtomic(Object object, File outputFile) throws IOException + { + // Try to write then perform atomic move so that file can't be corrupted + // by process crash in the middle of the writing. + File tempFile = new File(outputFile.path() + ".tmp"); + try + { + // Serialize to bytes first so we can flush and fsync before close. + // Jackson's writeValue(OutputStream, ...) auto-closes the stream, + // which would prevent us from calling sync() afterwards. + byte[] data = JSON_OBJECT_PRETTY_WRITER.writeValueAsBytes(object); + try (FileOutputStreamPlus out = tempFile.newOutputStream(OVERWRITE)) + { + out.write(data); + // Force data to disk before rename to ensure durability. + // Without this, a crash after rename but before OS flushes to disk + // can leave the file with zero-filled or corrupted blocks. + out.sync(); + } + tempFile.move(outputFile); + // Fsync the parent directory to ensure the rename is durable. + // Without this, a crash after rename can revert to the old directory entry. + // See: https://transactional.blog/how-to-learn/disk-io + SyncUtil.trySyncDir(outputFile.parent()); + } + catch (IOException ex) + { + tempFile.deleteIfExists(); + throw ex; + } + } + public static T deserializeFromJsonFile(Class tClass, File file) throws IOException { try (FileInputStreamPlus in = file.newInputStream()) @@ -188,6 +220,11 @@ public final class JsonUtils } } + public static T deserializeFromJsonBytes(Class tClass, byte[] bytes) throws IOException + { + return JSON_OBJECT_MAPPER.readValue(bytes, tClass); + } + /** * Handles unquoting and case-insensitivity in map keys. */ diff --git a/test/unit/org/apache/cassandra/service/StartupChecksTest.java b/test/unit/org/apache/cassandra/service/StartupChecksTest.java index 6fb063987a..48e673fd3d 100644 --- a/test/unit/org/apache/cassandra/service/StartupChecksTest.java +++ b/test/unit/org/apache/cassandra/service/StartupChecksTest.java @@ -18,21 +18,23 @@ package org.apache.cassandra.service; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.nio.file.FileStore; import java.nio.file.FileSystem; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.attribute.FileTime; import java.nio.file.spi.FileSystemProvider; import java.time.Instant; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.concurrent.Callable; import java.util.stream.Collectors; import org.junit.After; -import org.junit.AfterClass; import org.junit.Assert; import org.junit.Assume; import org.junit.Before; @@ -74,14 +76,13 @@ public class StartupChecksTest { StartupChecks startupChecks; Path sstableDir; - static File heartbeatFile; + File heartbeatFile; StartupChecksOptions options = new StartupChecksOptions(); @BeforeClass public static void setupServer() { - heartbeatFile = createTempFile("cassandra-heartbeat-", ""); SchemaLoader.prepareServer(); } @@ -97,6 +98,8 @@ public class StartupChecksTest sstableDir = Paths.get(dataDir.absolutePath(), "Keyspace1", "Standard1"); Files.createDirectories(sstableDir); + heartbeatFile = createTempFile("cassandra-heartbeat-" + UUID.randomUUID(), ""); + options.enable(check_data_resurrection); options.getConfig(check_data_resurrection) .put(HEARTBEAT_FILE_CONFIG_PROPERTY, heartbeatFile.absolutePath()); @@ -108,11 +111,6 @@ public class StartupChecksTest public void tearDown() throws IOException { new File(sstableDir).deleteRecursive(); - } - - @AfterClass - public static void tearDownClass() - { heartbeatFile.delete(); } @@ -229,6 +227,53 @@ public class StartupChecksTest verifyFailure(startupChecks, "Invalid tables: abc.def"); } + @Test + public void testDataResurrectionCheckLastModifiedFallback() throws Exception + { + DataResurrectionCheck check = new DataResurrectionCheck() { + @Override + List getKeyspaces() + { + return singletonList("test_ks"); + } + + @Override + List getTablesGcPeriods(String userKeyspace) + { + return singletonList(new TableGCPeriod("test_table", 10)); + } + }; + + int originalHintWindow = DatabaseDescriptor.getMaxHintWindow(); + try + { + DatabaseDescriptor.setMaxHintWindow(5 * 1000); + + // Empty file + Files.write(heartbeatFile.toPath(), "".getBytes(StandardCharsets.UTF_8)); + Instant recentTimestamp = Instant.ofEpochMilli(Clock.Global.currentTimeMillis()); + Files.setLastModifiedTime(heartbeatFile.toPath(), FileTime.from(recentTimestamp)); + + startupChecks.withTest(check); + verifySuccess(startupChecks); + } + finally + { + DatabaseDescriptor.setMaxHintWindow(originalHintWindow); + } + } + + private void verifySuccess(StartupChecks tests) { + try + { + tests.verify(options); + } + catch (StartupException e) + { + fail("Failed startup check with error: " + e.getMessage()); + } + } + @Test public void testKernelBug1057843Check() throws Exception {