diff --git a/CHANGES.txt b/CHANGES.txt index d64329acbc..4eee5cba9e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -5,6 +5,7 @@ Merged from 5.0: * Backport Automated Repair Inside Cassandra for CEP-37 (CASSANDRA-21138) Merged from 4.1: + * Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290) Merged from 4.0: * Rate limit password changes (CASSANDRA-21202) diff --git a/src/java/org/apache/cassandra/service/DataResurrectionCheck.java b/src/java/org/apache/cassandra/service/DataResurrectionCheck.java index 47c6d7beff..d2374deeac 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; @@ -46,6 +47,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; @@ -54,7 +56,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; @@ -87,12 +88,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 @@ -186,7 +199,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) @@ -310,7 +326,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 3c589c2c5b..a9b5d18f22 100644 --- a/src/java/org/apache/cassandra/utils/JsonUtils.java +++ b/src/java/org/apache/cassandra/utils/JsonUtils.java @@ -194,6 +194,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()) @@ -202,6 +234,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 6141802b57..806e3cd65f 100644 --- a/test/unit/org/apache/cassandra/service/StartupChecksTest.java +++ b/test/unit/org/apache/cassandra/service/StartupChecksTest.java @@ -18,11 +18,13 @@ 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.HashMap; @@ -30,13 +32,13 @@ import java.util.List; import java.util.Map; import java.util.ServiceLoader; import java.util.Set; +import java.util.UUID; import java.util.concurrent.Callable; import java.util.stream.Collectors; import com.vdurmont.semver4j.Semver; import org.junit.After; -import org.junit.AfterClass; import org.junit.Assert; import org.junit.Assume; import org.junit.Before; @@ -95,14 +97,13 @@ public class StartupChecksTest StartupChecks startupChecks; Path sstableDir; - static File heartbeatFile; + File heartbeatFile; StartupChecksConfiguration options = new StartupChecksConfiguration(new StartupChecks().withDefaultTests(), new HashMap<>()); @BeforeClass public static void setupServer() { - heartbeatFile = createTempFile("cassandra-heartbeat-", ""); SchemaLoader.prepareServer(); } @@ -118,6 +119,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()); @@ -129,11 +132,6 @@ public class StartupChecksTest public void tearDown() throws IOException { new File(sstableDir).deleteRecursive(); - } - - @AfterClass - public static void tearDownClass() - { heartbeatFile.delete(); } @@ -250,6 +248,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 {