This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-4.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-4.0 by this push:
new 80578c2df5 Validate Digest.crc32 when zero-copy streaming SSTables
80578c2df5 is described below
commit 80578c2df5a394b275432da83be53d6ddcc7fc82
Author: nivy <[email protected]>
AuthorDate: Fri Sep 18 21:54:09 2026 -0400
Validate Digest.crc32 when zero-copy streaming SSTables
Patch by Nivy Kani; reviewed by Caleb Rackliffe and marcuse for
CASSANDRA-21694
---
CHANGES.txt | 1 +
conf/cassandra.yaml | 8 ++
src/java/org/apache/cassandra/config/Config.java | 6 +
.../cassandra/config/DatabaseDescriptor.java | 11 ++
.../CassandraEntireSSTableStreamReader.java | 38 ++++++
.../apache/cassandra/metrics/StreamingMetrics.java | 9 ++
.../apache/cassandra/streaming/StreamSession.java | 6 +
.../test/hostreplacement/HostReplacementTest.java | 83 +++++++++++++
.../CassandraEntireSSTableStreamWriterTest.java | 137 +++++++++++++++++++++
9 files changed, 299 insertions(+)
diff --git a/CHANGES.txt b/CHANGES.txt
index 95c3861d2f..7fcbc24dce 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0.22
+ * Validate Digest.crc32 when zero-copy streaming SSTables (CASSANDRA-21694)
* Avoid DNS lookup for hostname column in ClientsTable and GossipInfoTable
(CASSANDRA-21539)
* Separate memtable memory usage tracking from limit checking, and fixes a a
flush writeBarrier deadlock (CASSANDRA-21019)
* Fixes for fully-expired SSTable index notification in compaction
(CASSANDRA-21577)
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 14f11587a8..e7bb1c65ad 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -942,6 +942,14 @@ sstable_preemptive_open_interval_in_mb: 50
# disabled if internode encryption is enabled.
# stream_entire_sstables: true
+# When an entire SSTable is received via zero-copy streaming, validate the
Digest.crc32 that was streamed with it
+# against the received Data.db before the SSTable is added to the table. This
detects corruption introduced in
+# transit as well as pre-existing corruption on the sending node's disk,
neither of which is otherwise noticed by
+# this path, since it writes component bytes to disk without parsing them.
+# If there's no Digest.crc32, this check passes.
+# When unset, the default is not validating Digest.crc32.
+# entire_sstable_stream_digest_validation_enabled: false
+
# Throttles all outbound streaming file transfers on this node to the
# given total throughput in Mbps. This is necessary because Cassandra does
# mostly sequential IO when streaming data during bootstrap or repair, which
diff --git a/src/java/org/apache/cassandra/config/Config.java
b/src/java/org/apache/cassandra/config/Config.java
index 24df356a4b..2611379a8f 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -490,6 +490,12 @@ public class Config
public volatile int max_concurrent_automatic_sstable_upgrades = 1;
public boolean stream_entire_sstables = true;
+ /**
+ * When receiving a zero-copy stream, validate Digest.crc32 before the
sstable is made visible.
+ */
+ public volatile boolean entire_sstable_stream_digest_validation_enabled =
false;
+
+
public volatile AuditLogOptions audit_logging_options = new
AuditLogOptions();
public volatile FullQueryLoggerOptions full_query_logging_options = new
FullQueryLoggerOptions();
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 00a4aa912f..8aa01c53d4 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -2860,6 +2860,17 @@ public class DatabaseDescriptor
return conf.stream_entire_sstables;
}
+ public static boolean getEntireSSTableStreamDigestValidationEnabled()
+ {
+ return conf.entire_sstable_stream_digest_validation_enabled;
+ }
+
+ public static void setEntireSSTableStreamDigestValidationEnabled(boolean
value)
+ {
+ conf.entire_sstable_stream_digest_validation_enabled = value;
+ }
+
+
public static String getLocalDataCenter()
{
return localDC;
diff --git
a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
index 0bfe993112..dadc026f3a 100644
---
a/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
+++
b/src/java/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamReader.java
@@ -21,6 +21,8 @@ package org.apache.cassandra.db.streaming;
import java.io.File;
import java.io.IOException;
import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import java.util.function.UnaryOperator;
import com.google.common.base.Throwables;
@@ -28,6 +30,7 @@ import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.io.sstable.Component;
@@ -37,11 +40,13 @@ import org.apache.cassandra.io.sstable.format.SSTableFormat;
import org.apache.cassandra.io.sstable.format.big.BigTableZeroCopyWriter;
import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataIntegrityMetadata;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.streaming.ProgressInfo;
import org.apache.cassandra.streaming.StreamReceiver;
import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.streaming.messages.StreamMessageHeader;
+import org.apache.cassandra.utils.NoSpamLogger;
import static java.lang.String.format;
import static org.apache.cassandra.utils.FBUtilities.prettyPrintMemory;
@@ -135,6 +140,8 @@ public class CassandraEntireSSTableStreamReader implements
IStreamReader
prettyPrintMemory(totalSize));
}
+ validateDataDigest(writer.descriptor, manifest);
+
UnaryOperator<StatsMetadata> transform = stats ->
stats.mutateLevel(header.sstableLevel)
.mutateRepairedMetadata(messageHeader.repairedAt, messageHeader.pendingRepair,
false);
String description = String.format("level %s and repairedAt time
%s and pendingRepair %s",
@@ -152,6 +159,37 @@ public class CassandraEntireSSTableStreamReader implements
IStreamReader
}
}
+ /**
+ * Validates the received {@code Data.db} against {@code Digest.crc32}.
+ * @throws IOException if a digest was streamed and the received data file
does not match it
+ */
+ private void validateDataDigest(Descriptor descriptor, ComponentManifest
manifest) throws IOException
+ {
+ if
(!DatabaseDescriptor.getEntireSSTableStreamDigestValidationEnabled())
+ return;
+
+ List<Component> components = manifest.components();
+ if (!components.contains(Component.DATA) ||
!components.contains(Component.DIGEST))
+ {
+ NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 1,
TimeUnit.MINUTES,
+ "Accepting zero-copy sstable stream from {}
without validating CRC due to " +
+ "missing {} or {} from streamed components: {}",
+ session.peer, Component.DATA, Component.DIGEST,
components);
+ return;
+ }
+
+ try (DataIntegrityMetadata.FileDigestValidator validator = new
DataIntegrityMetadata.FileDigestValidator(descriptor))
+ {
+ validator.validate();
+ }
+ catch (IOException e)
+ {
+ session.countEntireSSTableDigestMismatch();
+ throw new IOException(format("[Stream #%s] Digest mismatch for %s
received from %s, aborting stream",
+ session.planId(), descriptor,
session.peer), e);
+ }
+ }
+
private File getDataDir(ColumnFamilyStore cfs, long totalSize) throws
IOException
{
Directories.DataDirectory localDir =
cfs.getDirectories().getWriteableLocation(totalSize);
diff --git a/src/java/org/apache/cassandra/metrics/StreamingMetrics.java
b/src/java/org/apache/cassandra/metrics/StreamingMetrics.java
index 54df23357e..0fe1f76f88 100644
--- a/src/java/org/apache/cassandra/metrics/StreamingMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/StreamingMetrics.java
@@ -47,6 +47,7 @@ public class StreamingMetrics
public final Counter outgoingBytes;
/* Measures the time taken for processing the incoming stream message
after being deserialized, including the time to flush to disk. */
public final Timer incomingProcessTime;
+ private final Counter entireSSTableDigestMismatches;
public static StreamingMetrics get(InetAddressAndPort ip)
{
@@ -79,5 +80,13 @@ public class StreamingMetrics
incomingBytes =
Metrics.counter(factory.createMetricName("IncomingBytes"));
outgoingBytes=
Metrics.counter(factory.createMetricName("OutgoingBytes"));
incomingProcessTime =
Metrics.timer(factory.createMetricName("IncomingProcessTime"));
+
+ entireSSTableDigestMismatches =
Metrics.counter(factory.createMetricName("EntireSSTableDigestMismatches"));
+ }
+
+ /** An entire-sstable stream was rejected because the received data file
did not match the streamed digest. */
+ public void countEntireSSTableDigestMismatch()
+ {
+ entireSSTableDigestMismatches.inc();
}
}
diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java
b/src/java/org/apache/cassandra/streaming/StreamSession.java
index 786311b1d8..0d2668c927 100644
--- a/src/java/org/apache/cassandra/streaming/StreamSession.java
+++ b/src/java/org/apache/cassandra/streaming/StreamSession.java
@@ -754,6 +754,12 @@ public class StreamSession implements
IEndpointStateChangeSubscriber
});
}
+ public void countEntireSSTableDigestMismatch()
+ {
+ metrics.countEntireSSTableDigestMismatch();
+ }
+
+
/**
* Finish preparing the session. This method is blocking (memtables are
flushed in {@link #addTransferRanges}),
* so the logic should not execute on the main IO thread (read: netty
event loop).
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java
b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java
index 3de0bf51d5..941174f681 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/hostreplacement/HostReplacementTest.java
@@ -19,13 +19,17 @@
package org.apache.cassandra.distributed.test.hostreplacement;
import java.io.IOException;
+import java.io.RandomAccessFile;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.TimeoutException;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.Constants;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
@@ -36,6 +40,8 @@ import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.AssertUtils;
import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.assertj.core.api.Assertions;
import static
org.apache.cassandra.config.CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK;
@@ -205,6 +211,83 @@ public class HostReplacementTest extends TestBaseImpl
}
}
+ private static final int HEALTHY_PK = 10;
+ private static final int CORRUPT_PK = 11;
+
+ /**
+ * With digest validation enabled, a single corrupted sstable must fail
the entire stream session, so neither
+ * the corrupted sstable nor its healthy sibling should land on the
replacing node.
+ */
+ @Test
+ public void
replaceHostAbortsStreamingWhenDigestValidationCatchesCorruption() throws
IOException, TimeoutException
+ {
+ TokenSupplier even = TokenSupplier.evenlyDistributedTokens(2);
+ try (Cluster cluster = Cluster.build(2)
+ .withConfig(c -> c.with(Feature.GOSSIP,
Feature.NETWORK)
+
.set("entire_sstable_stream_digest_validation_enabled", true))
+ .withTokenSupplier(node ->
even.token(node == 3 ? 2 : node))
+ .start())
+ {
+ IInvokableInstance node1 = cluster.get(1);
+ IInvokableInstance nodeToRemove = cluster.get(2);
+
+ setupCluster(cluster);
+ writeHealthyAndCorruptSSTables(cluster, node1);
+
+ stopUnchecked(nodeToRemove);
+
+ IInvokableInstance replacingNode = replaceHostAndStart(cluster,
nodeToRemove);
+
+ replacingNode.logs().watchFor("Digest mismatch");
+
+ // the whole stream session is aborted on a digest mismatch, so
neither sstable should have landed
+ AssertUtils.assertRows(replacingNode.executeInternal("SELECT *
FROM " + KEYSPACE + ".tbl WHERE pk = " + HEALTHY_PK), EMPTY_ROWS);
+ AssertUtils.assertRows(replacingNode.executeInternal("SELECT *
FROM " + KEYSPACE + ".tbl WHERE pk = " + CORRUPT_PK), EMPTY_ROWS);
+ }
+ }
+
+ /**
+ * Flushes two separate sstables on {@code node1} for {@code
KEYSPACE.tbl}: one containing {@link #HEALTHY_PK}
+ * left untouched, and one containing {@link #CORRUPT_PK} that gets its
Data.db corrupted.
+ */
+ static void writeHealthyAndCorruptSSTables(Cluster cluster,
IInvokableInstance node1) throws IOException
+ {
+ node1.nodetoolResult("disableautocompaction").asserts().success();
+
+ cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk)
VALUES (?)",
+ ConsistencyLevel.ALL,
+ HEALTHY_PK);
+ node1.flush(KEYSPACE);
+
+ cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk)
VALUES (?)",
+ ConsistencyLevel.ALL,
+ CORRUPT_PK);
+ node1.flush(KEYSPACE);
+
+ corruptSSTable(node1, "tbl");
+ }
+
+ /** Flips a byte in the middle of the Data.db file of the most recently
flushed sstable for the given table. */
+ static void corruptSSTable(IInvokableInstance instance, String table)
throws IOException
+ {
+ String dataFilePath = instance.callOnInstance(() -> {
+ ColumnFamilyStore cfs =
Keyspace.open(KEYSPACE).getColumnFamilyStore(table);
+ SSTableReader sstable = cfs.getLiveSSTables().stream()
+
.max(SSTableReader.maxTimestampAscending)
+ .get();
+ return sstable.descriptor.filenameFor(Component.DATA);
+ });
+
+ try (RandomAccessFile raf = new RandomAccessFile(dataFilePath, "rw"))
+ {
+ long corruptAt = raf.length() / 2;
+ raf.seek(corruptAt);
+ int b = raf.read();
+ raf.seek(corruptAt);
+ raf.write(b ^ 0xFF);
+ }
+ }
+
static void setupCluster(Cluster cluster)
{
fixDistributedSchemas(cluster);
diff --git
a/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java
b/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java
index 58d26c11ad..43099cfed3 100644
---
a/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java
+++
b/test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java
@@ -17,18 +17,26 @@
*/
package org.apache.cassandra.db.streaming;
+import java.io.FileInputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import java.util.Collection;
import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.Queue;
import java.util.UUID;
+import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
+import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.apache.cassandra.config.DatabaseDescriptor;
+
+import com.google.common.io.ByteStreams;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
@@ -62,7 +70,9 @@ import
org.apache.cassandra.streaming.messages.StreamMessageHeader;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
public class CassandraEntireSSTableStreamWriterTest
@@ -75,11 +85,14 @@ public class CassandraEntireSSTableStreamWriterTest
private static SSTableReader sstable;
private static Descriptor descriptor;
private static ColumnFamilyStore store;
+ private static boolean originalDigestValidationEnabled;
@BeforeClass
public static void defineSchemaAndPrepareSSTable()
{
SchemaLoader.prepareServer();
+ originalDigestValidationEnabled =
DatabaseDescriptor.getEntireSSTableStreamDigestValidationEnabled();
+
SchemaLoader.createKeyspace(KEYSPACE,
KeyspaceParams.simple(1),
SchemaLoader.standardCFMD(KEYSPACE,
CF_STANDARD),
@@ -109,6 +122,12 @@ public class CassandraEntireSSTableStreamWriterTest
descriptor = sstable.descriptor;
}
+ @AfterClass
+ public static void restoreDigestValidationFlag()
+ {
+
DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(originalDigestValidationEnabled);
+ }
+
@Test
public void testBlockWriterOverWire() throws IOException
{
@@ -169,6 +188,124 @@ public class CassandraEntireSSTableStreamWriterTest
}
}
+ @Test
+ public void testCorruptDataComponentFailsDigestValidation() throws
Throwable
+ {
+ ByteBuf serializedFile = Unpooled.buffer(8192);
+ ComponentManifest manifest = writeEntireSSTable(serializedFile);
+ corruptDataComponent(serializedFile, manifest);
+
+ DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(true);
+ assertThatThrownBy(() -> readEntireSSTable(serializedFile, manifest))
+ .hasCauseInstanceOf(IOException.class)
+ .hasStackTraceContaining("Digest mismatch");
+ }
+
+ @Test
+ public void testCorruptDataComponentAcceptedWhenValidationDisabled()
throws Throwable
+ {
+ ByteBuf serializedFile = Unpooled.buffer(8192);
+ ComponentManifest manifest = writeEntireSSTable(serializedFile);
+ corruptDataComponent(serializedFile, manifest);
+
+
DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(false);
+
+ // shouldn't throw an error
+ SSTableMultiWriter writer = readEntireSSTable(serializedFile,
manifest);
+ assertNotNull(writer);
+ writer.abort(null);
+ }
+
+ @Test
+ public void testMissingDigestIsAcceptedWithoutValidation() throws Throwable
+ {
+ DatabaseDescriptor.setEntireSSTableStreamDigestValidationEnabled(true);
+
+ ComponentManifest full;
+ try (ComponentContext context = ComponentContext.create(descriptor))
+ {
+ full = context.manifest();
+ }
+
+ // remove digest component from sstable
+ LinkedHashMap<Component, Long> withoutDigest = new LinkedHashMap<>();
+ for (Component component : full.components())
+ {
+ if (!component.equals(Component.DIGEST))
+ withoutDigest.put(component, full.sizeOf(component));
+ }
+ ComponentManifest manifest = new ComponentManifest(withoutDigest);
+
+ // stream from live files
+ ByteBuf serializedFile = Unpooled.buffer(8192);
+ for (Component component : manifest.components())
+ {
+ try (InputStream in = new
FileInputStream(descriptor.filenameFor(component)))
+ {
+ serializedFile.writeBytes(ByteStreams.toByteArray(in));
+ }
+ }
+
+ // shouldn't throw an error
+ SSTableMultiWriter writer = readEntireSSTable(serializedFile,
manifest);
+ assertNotNull(writer);
+ writer.abort(null);
+ }
+
+ private ComponentManifest writeEntireSSTable(ByteBuf serializedFile)
throws Exception
+ {
+ StreamSession session = setupStreamingSessionForTest();
+ EmbeddedChannel channel = createMockNettyChannel(serializedFile);
+ try (AsyncStreamingOutputPlus out = new
AsyncStreamingOutputPlus(channel);
+ ComponentContext context = ComponentContext.create(descriptor))
+ {
+ new CassandraEntireSSTableStreamWriter(sstable, session,
context).write(out);
+ return context.manifest();
+ }
+ }
+
+ private SSTableMultiWriter readEntireSSTable(ByteBuf serializedFile,
ComponentManifest manifest) throws Throwable
+ {
+ StreamSession session = setupStreamingSessionForTest();
+ InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort();
+ session.prepareReceiving(new StreamSummary(sstable.metadata().id, 1,
5104));
+
+ CassandraStreamHeader header =
+ CassandraStreamHeader.builder()
+ .withSSTableFormat(sstable.descriptor.formatType)
+ .withSSTableVersion(sstable.descriptor.version)
+ .withSSTableLevel(0)
+ .withEstimatedKeys(sstable.estimatedKeys())
+ .withSections(Collections.emptyList())
+
.withSerializationHeader(sstable.header.toComponent())
+ .withComponentManifest(manifest)
+ .isEntireSSTable(true)
+ .withFirstKey(sstable.first)
+ .withTableId(sstable.metadata().id)
+ .build();
+
+ CassandraEntireSSTableStreamReader reader =
+ new CassandraEntireSSTableStreamReader(new
StreamMessageHeader(sstable.metadata().id, peer, session.planId(), false, 0, 0,
0, null),
+ header, session);
+
+ return reader.read(new DataInputBuffer(serializedFile.nioBuffer(),
false));
+ }
+
+ /** Flips every bit of one byte in the middle of the Data.db region of an
already-serialized stream. */
+ private static void corruptDataComponent(ByteBuf serializedFile,
ComponentManifest manifest)
+ {
+ long dataOffset = 0;
+ for (Component component : manifest.components())
+ {
+ if (component.equals(Component.DATA))
+ break;
+ dataOffset += manifest.sizeOf(component);
+ }
+
+ int corruptAt = (int) (dataOffset + manifest.sizeOf(Component.DATA) /
2);
+ serializedFile.setByte(corruptAt, serializedFile.getByte(corruptAt) ^
0xff);
+ }
+
private EmbeddedChannel createMockNettyChannel(ByteBuf serializedFile)
throws Exception
{
WritableByteChannel wbc = new WritableByteChannel()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
