natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add accord journal standalone dump tool 

Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20738
This commit is contained in:
Alex Petrov 2025-07-01 07:52:33 +02:00
parent 674b8d5100
commit ccf12ef845
15 changed files with 500 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitmodules vendored
View File

@ -1,4 +1,4 @@
[submodule "modules/accord"]
path = modules/accord
url = https://github.com/apache/cassandra-accord.git
branch = trunk
branch = trunk

3
src/java/org/apache/cassandra/config/AccordSpec.java
View File

@ -196,10 +196,11 @@ public class AccordSpec
private volatile long flushCombinedBlockPeriod = Long.MIN_VALUE;
public Version version = Version.DOWNGRADE_SAFE_VERSION;
public void setFlushPeriod(DurationSpec newFlushPeriod)
public JournalSpec setFlushPeriod(DurationSpec newFlushPeriod)
{
flushPeriod = newFlushPeriod;
flushCombinedBlockPeriod = Long.MIN_VALUE;
return this;
}
public void setPeriodicFlushLagBlock(DurationSpec newPeriodicFlushLagBlock)

2
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -2234,7 +2234,7 @@ public class DatabaseDescriptor
}
catch (FSWriteError e)
{
throw new IllegalStateException(e.getCause().getMessage() + "; unable to start server");
throw new IllegalStateException(e.getCause().getMessage() + "; unable to start server", e);
}
}

2
src/java/org/apache/cassandra/journal/Descriptor.java
View File

@ -109,7 +109,7 @@ public final class Descriptor implements Comparable<Descriptor>
return new Descriptor(directory, timestamp, generation, journalVersion, userVersion);
}
static Descriptor fromFile(File file)
public static Descriptor fromFile(File file)
{
return fromName(file.parent(), file.name());
}

42
src/java/org/apache/cassandra/journal/DumpUtil.java Normal file
View File

@ -0,0 +1,42 @@
/*
* 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.cassandra.journal;
import java.util.function.Consumer;
/**
* Helper file to avoid exposing components outside their package-local visibility scope
*/
public class DumpUtil
{
public static void dumpMetadata(Descriptor descriptor, Consumer<String> out)
{
if (Component.METADATA.existsFor(descriptor))
{
out.accept(Metadata.load(descriptor).toString());
}
else
out.accept("Metadata absent for " + descriptor);
}
public static <K, V> StaticSegment<K, V> open(Descriptor descriptor, KeySupport<K> keySupport)
{
return StaticSegment.open(descriptor, keySupport);
}
}

9
src/java/org/apache/cassandra/journal/Metadata.java
View File

@ -154,4 +154,13 @@ final class Metadata
metadata.persist(descriptor);
return metadata;
}
@Override
public String toString()
{
return "Metadata{" +
"fsyncLimit=" + fsyncLimit +
", recordsCount=" + recordsCount +
'}';
}
}

2
src/java/org/apache/cassandra/journal/StaticSegment.java
View File

@ -289,7 +289,7 @@ public final class StaticSegment<K, V> extends Segment<K, V>
/**
* Iterate over and invoke the supplied callback on every record.
*/
void forEachRecord(RecordConsumer<K> consumer)
public void forEachRecord(RecordConsumer<K> consumer)
{
try (SequentialReader<K> reader = sequentialReader(descriptor, keySupport, fsyncLimit))
{

10
src/java/org/apache/cassandra/net/InboundMessageHandler.java
View File

@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -167,7 +168,14 @@ public class InboundMessageHandler extends AbstractMessageHandler
{
Message<?> m = serializer.deserialize(in, header, version);
if (in.available() > 0) // bytes remaining after deser: deserializer is busted
throw new InvalidSerializedSizeException(header.verb, size, size - in.available());
{
Throwable t = new InvalidSerializedSizeException(header.verb, size, size - in.available());
ByteBuffer clone = bytes.get();
clone.limit(clone.position() + size); // cap to expected message size
logger.error("Could not deserialize the message: {}", ByteBufferUtil.bytesToHex(clone), t);
throw t;
}
message = m;
}
catch (IncompatibleSchemaException e)

8
src/java/org/apache/cassandra/schema/SchemaConstants.java
View File

@ -92,6 +92,14 @@ public final class SchemaConstants
emptyVersion = UUID.nameUUIDFromBytes(Digest.forSchema().digest());
}
/**
* @return whether the table is an Accord Journal tabe
*/
public static boolean isAccordJournal(String keyspaceName, String tableName)
{
return keyspaceName.equals(SchemaConstants.ACCORD_KEYSPACE_NAME) && tableName.equals(AccordKeyspace.JOURNAL);
}
/**
* @return whether or not the keyspace is a really system one (w/ LocalStrategy, unmodifiable, hardcoded)
*/

7
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -415,7 +415,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
return loadDiffs(commandStoreId, txnId, Load.ALL);
}
private <BUILDER extends FlyweightImage> BUILDER readAll(JournalKey key)
public <BUILDER extends FlyweightImage> BUILDER readAll(JournalKey key)
{
BUILDER builder = (BUILDER) key.type.serializer.mergerFor();
// TODO (expected): this can be further improved to avoid allocating lambdas
@ -425,6 +425,11 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
return builder;
}
public void forEachEntry(JournalKey key, AccordJournalTable.Reader reader)
{
journalTable.readAll(key, reader);
}
private <T> RecordPointer appendInternal(JournalKey key, T write)
{
AccordJournalValueSerializers.FlyweightSerializer<T, ?> serializer = (AccordJournalValueSerializers.FlyweightSerializer<T, ?>) key.type.serializer;

15
src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
View File

@ -55,6 +55,13 @@ public class AccordJournalValueSerializers
void reserialize(JournalKey key, IMAGE from, DataOutputPlus out, Version userVersion) throws IOException;
void deserialize(JournalKey key, IMAGE into, DataInputPlus in, Version userVersion) throws IOException;
default IMAGE deserialize(JournalKey key, DataInputPlus in, Version userVersion) throws IOException
{
IMAGE image = mergerFor();
deserialize(key, image, in, userVersion);
return image;
}
}
public static class CommandDiffSerializer
@ -142,6 +149,14 @@ public class AccordJournalValueSerializers
hasRead = true;
return newValue;
}
@Override
public String toString()
{
return "IdentityAccumulator{" +
initial +
'}';
}
}
public static class RedundantBeforeSerializer

53
src/java/org/apache/cassandra/tcm/ClusterMetadataService.java
View File

@ -34,6 +34,14 @@ import java.util.function.Supplier;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import org.apache.cassandra.schema.*;
import org.apache.cassandra.service.accord.AccordFastPath;
import org.apache.cassandra.service.accord.AccordStaleReplicas;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
import org.apache.cassandra.tcm.membership.Directory;
import org.apache.cassandra.tcm.ownership.DataPlacements;
import org.apache.cassandra.tcm.ownership.TokenMap;
import org.apache.cassandra.tcm.sequences.LockedRanges;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -51,8 +59,6 @@ import org.apache.cassandra.metrics.TCMMetrics;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.schema.DistributedSchema;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.listeners.SchemaListener;
import org.apache.cassandra.tcm.log.Entry;
import org.apache.cassandra.tcm.log.LocalLog;
@ -288,6 +294,49 @@ public class ClusterMetadataService
ClusterMetadataService.setInstance(cms);
}
public static void empty(Keyspaces keyspaces)
{
if (instance != null)
return;
String localDC = DatabaseDescriptor.getLocalDataCenter();
ClusterMetadata empty = new ClusterMetadata(Epoch.EMPTY,
DatabaseDescriptor.getPartitioner(),
new DistributedSchema(keyspaces),
Directory.EMPTY,
new TokenMap(DatabaseDescriptor.getPartitioner()),
DataPlacements.empty(),
AccordFastPath.EMPTY,
LockedRanges.EMPTY,
InProgressSequences.EMPTY,
ConsensusMigrationState.EMPTY,
Collections.emptyMap(),
AccordStaleReplicas.EMPTY);
LocalLog.LogSpec logSpec = LocalLog.logSpec()
.withInitialState(empty)
.loadSSTables(false)
.withDefaultListeners(false)
.sync()
.withStorage(new AtomicLongBackedProcessor.InMemoryStorage());
LocalLog log = logSpec.createLog();
ClusterMetadataService cms = new ClusterMetadataService(new UniformRangePlacement(),
MetadataSnapshots.NO_OP,
log,
new AtomicLongBackedProcessor(log),
new LogState.ReplicationHandler(log),
new LogState.LogNotifyHandler(log),
new CurrentEpochRequestHandler(),
null,
null,
null);
log.readyUnchecked();
log.bootstrap(FBUtilities.getBroadcastAddressAndPort(), localDC);
ClusterMetadataService.setInstance(cms);
}
@SuppressWarnings("resource")
public static void initializeForClients()
{

353
src/java/org/apache/cassandra/tools/StandaloneJournalUtil.java Normal file
View File

@ -0,0 +1,353 @@
/*
* 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.cassandra.tools;
import accord.local.RedundantBefore;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import io.airlift.airline.Cli;
import io.airlift.airline.Command;
import io.airlift.airline.Option;
import org.apache.cassandra.config.AccordSpec;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.DurationSpec;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.journal.Descriptor;
import org.apache.cassandra.journal.DumpUtil;
import org.apache.cassandra.journal.StaticSegment;
import org.apache.cassandra.schema.Keyspaces;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.Tables;
import org.apache.cassandra.service.accord.AccordJournal;
import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.JournalKey;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.tcm.ClusterMetadataService;
import java.io.IOException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.PathMatcher;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import static accord.api.Journal.Load.ALL;
import static com.google.common.base.Throwables.getStackTraceAsString;
import static org.apache.cassandra.config.DatabaseDescriptor.setAccordJournalDirectory;
/**
* Standalone Accord Journal Util. Useful for inspecting journals on remote nodes in cases of failures
* As a convenience, you can build the dtest-jar and scp it to the server, and run the tool:
java --add-exports java.base/jdk.internal.misc=ALL-UNNAMED \
--add-exports java.base/jdk.internal.ref=ALL-UNNAMED \
--add-exports java.base/sun.nio.ch=ALL-UNNAMED \
--add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED \
--add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED \
--add-exports java.rmi/sun.rmi.server=ALL-UNNAMED \
--add-exports java.rmi/sun.rmi.transport.tcp=ALL-UNNAMED \
--add-exports java.sql/java.sql=ALL-UNNAMED \
\
--add-opens java.base/java.lang.module=ALL-UNNAMED \
--add-opens java.base/java.net=ALL-UNNAMED \
--add-opens java.base/jdk.internal.loader=ALL-UNNAMED \
--add-opens java.base/jdk.internal.ref=ALL-UNNAMED \
--add-opens java.base/jdk.internal.reflect=ALL-UNNAMED \
--add-opens java.base/jdk.internal.math=ALL-UNNAMED \
--add-opens java.base/jdk.internal.module=ALL-UNNAMED \
--add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED \
--add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED \
--add-opens jdk.management/com.sun.beans.introspect=ALL-UNNAMED \
--add-opens jdk.management.jfr/jdk.management.jfr=ALL-UNNAMED \
--add-opens java.desktop/com.sun.beans.introspect=ALL-UNNAMED \
-cp ./dtest-5.1.jar \
org.apache.cassandra.tools.StandaloneJournalUtil \
dump_journal --construct --skip-errors --sstables /tmp/journals/journal-c3412dbac2923051b7914f7b0b79a166/ --journal-segments /tmp/journals/accord_journal/
*
**/
public class StandaloneJournalUtil
{
private static final Output output = Output.CONSOLE;
public static void main(String... args)
{
Util.initDatabaseDescriptor();
DatabaseDescriptor.setPartitioner("org.apache.cassandra.dht.Murmur3Partitioner");
AccordKeyspace.TABLES = Tables.of(AccordKeyspace.journalMetadata("journal", false));
ClusterMetadataService.empty(Keyspaces.of(AccordKeyspace.metadata()));
Cli.CliBuilder<Runnable> builder = Cli.builder("util");
builder.withDescription("Dump journal").withCommand(DumpSegments.class).withCommand(DumpJournal.class);
Cli<Runnable> parser = builder.build();
int status = 0;
try
{
Runnable parse = parser.parse(args);
parse.run();
}
catch (Throwable throwable)
{
err(throwable);
status = 2;
}
System.exit(status);
}
protected static void err(Throwable e)
{
output.err.println("error: " + e.getMessage());
output.err.println("-- StackTrace --");
output.err.println(getStackTraceAsString(e));
}
@Command(name = "dump_segments", description = "Dump journal segments")
public static class DumpSegments implements Runnable
{
@Option(name = {"-d", "--dir"}, description = "Directory to find journal segments")
public String dir;
@Option(name = {"-p", "--pattern"}, description = "Kind to filter by")
public String pattern;
@Option(name = {"-k", "--kind"}, description = "Kind to filter by")
public String kind;
@Option(name = {"-t", "--txnid"}, description = "Transaction id to filter by")
public String txnId;
@Option(name = {"-m", "--metadata-only"}, description = "Only dump metadata file contents")
public boolean metadataOnly;
public void run()
{
PathMatcher matcher = FileSystems.getDefault().getPathMatcher("glob:" + pattern);
try
{
Files.list(Path.of(dir)).filter(Files::isRegularFile).filter(p -> matcher.matches(p.getFileName())).sorted().forEach(path -> {
Descriptor descriptor = Descriptor.fromFile(new File(path));
DumpUtil.dumpMetadata(descriptor, output.out::println);
if (metadataOnly)
return;
StaticSegment<JournalKey, Object> segment = DumpUtil.open(descriptor, JournalKey.SUPPORT);
segment.forEachRecord((segment1, position, key, buffer, userVersion) -> {
if (kind != null && key.type != JournalKey.Type.valueOf(kind))
return;
if (txnId != null && !TxnId.fromString(txnId).equals(key.id))
return;
try (DataInputBuffer in = new DataInputBuffer(buffer, false))
{
Object v = key.type.serializer.deserialize(key, in, Version.V1);
output.out.printf("%s: %s%n", key, v);
}
catch (Throwable t)
{
t.printStackTrace(output.err);
}
});
});
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
}
@Command(name = "dump_journal", description = "Dump journal")
public static class DumpJournal implements Runnable
{
@Option(name = {"-s", "--sstables"}, description = "Path to sstables")
public String sstables;
@Option(name = {"-j", "--journal-segments"}, description = "Path to journal segments")
public String journalSegments;
@Option(name = {"-k", "--kind"}, description = "Kind to filter by")
public String kind;
@Option(name = {"-t", "--txnid"}, description = "Transaction id to filter by")
public String txnId;
@Option(name = {"--since"}, description = "Filter transactions since this timestamp (inclusive)")
public String since;
@Option(name = {"--until"}, description = "Filter transactions until this timestamp (inclusive)")
public String until;
@Option(name = {"-e", "--skip-errors"}, description = "Skip errors: 'true' to skip all, or comma-separated exception class names to skip specific types")
public String skipErrors;
@Option(name = {"-c", "--construct"}, description = "Construct entry")
public boolean construct;
public void run()
{
if (sstables == null && journalSegments == null)
throw new IllegalArgumentException("Either --sstables or --journal-segments must be provided");
Timestamp txnId = this.txnId != null ? TxnId.fromString(this.txnId) : null;
Timestamp sinceTimestamp = this.since != null ? TxnId.fromString(this.since) : null;
Timestamp untilTimestamp = this.until != null ? TxnId.fromString(this.until) : null;
boolean skipAllErrors;
Set<String> skipExceptionTypes = new HashSet<>();
if (skipErrors != null && !skipErrors.trim().isEmpty())
{
String trimmed = skipErrors.trim();
if ("true".equalsIgnoreCase(trimmed))
{
skipAllErrors = true;
}
else
{
skipAllErrors = false;
String[] types = trimmed.split(",");
for (String type : types)
{
skipExceptionTypes.add(type.trim());
}
}
}
else
{
skipAllErrors = false;
}
if (journalSegments == null)
{
try
{
journalSegments = Files.createTempDirectory("dump_journal").getFileName().toString();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
setAccordJournalDirectory(journalSegments);
Keyspace.setInitialized();
AccordJournal journal = new AccordJournal(new AccordSpec.JournalSpec().setFlushPeriod(new DurationSpec.IntMillisecondsBound("1500ms")), new File(journalSegments).parent(), Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL));
Keyspace ks = Schema.instance.getKeyspaceInstance("system_accord");
ColumnFamilyStore cfs = ks.getColumnFamilyStore("journal");
if (sstables != null)
cfs.importNewSSTables(Collections.singleton(sstables), false, false, false, false, false, false, true);
Map<Integer, RedundantBefore> cache = new HashMap<>();
journal.start(null);
journal.forEach(key -> processKey(cache, journal, key, txnId, sinceTimestamp, untilTimestamp, skipAllErrors, skipExceptionTypes));
}
private void processKey(Map<Integer, RedundantBefore> redundantBeforeCache, AccordJournal journal, JournalKey key, Timestamp txnId, Timestamp minTimestamp, Timestamp maxTimestamp, boolean skipAllErrors, Set<String> skipExceptionTypes)
{
if (kind != null && key.type != JournalKey.Type.valueOf(kind))
return;
if (txnId != null && !txnId.equals(key.id))
return;
// Apply since filtering (inclusive)
if ((minTimestamp != null && key.id.compareTo(minTimestamp) < 0) ||
(maxTimestamp != null && key.id.compareTo(maxTimestamp) > 0))
return;
try
{
switch (key.type)
{
case COMMAND_DIFF:
{
AtomicInteger counter = new AtomicInteger();
output.out.println(key + " " + key.id.toStandardString());
output.out.println("Individual entries:");
journal.forEachEntry(key, (in, userVersion) -> {
AccordJournal.Builder builder = new AccordJournal.Builder(key.id, ALL);
builder.deserializeNext(in, userVersion);
output.out.println(String.format("\t%s", builder.toString("\n\t\t")));
counter.getAndIncrement();
});
if (construct)
{
AccordJournal.Builder builder = new AccordJournal.Builder(key.id, ALL);
journal.forEachEntry(key, builder::deserializeNext);
output.out.println("Reconstructed\n\t\t" + builder.construct(redundantBeforeCache.computeIfAbsent(key.commandStoreId, k -> journal.loadRedundantBefore(key.commandStoreId))));
}
break;
}
case REDUNDANT_BEFORE:
case DURABLE_BEFORE:
case SAFE_TO_READ:
case BOOTSTRAP_BEGAN_AT:
case RANGES_FOR_EPOCH:
case TOPOLOGY_UPDATE:
{
Object image = journal.readAll(key);
output.out.println(image);
}
}
}
catch (Throwable t)
{
boolean shouldSkip = false;
if (skipAllErrors)
{
shouldSkip = true;
}
else if (!skipExceptionTypes.isEmpty())
{
String exceptionClassName = t.getClass().getSimpleName();
String fullExceptionClassName = t.getClass().getName();
shouldSkip = skipExceptionTypes.contains(exceptionClassName) ||
skipExceptionTypes.contains(fullExceptionClassName);
}
if (!shouldSkip)
throw t;
output.out.println(String.format("Got error reading key %s", key));
t.printStackTrace();
}
}
}
}

2
src/java/org/apache/cassandra/utils/btree/FullBTreeSearchIterator.java
View File

@ -22,8 +22,6 @@ import static org.apache.cassandra.utils.btree.BTree.size;
import java.util.Comparator;
import java.util.NoSuchElementException;
import accord.utils.Invariants;
public class FullBTreeSearchIterator<K, V> extends TreeCursor<K> implements BTreeSearchIterator<K, V>
{
private final boolean forwards;

3
test/unit/org/apache/cassandra/service/accord/WatermarkCollectorTest.java
View File

@ -30,7 +30,6 @@ import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.Invariants;
import org.agrona.collections.Int2ObjectHashMap;
import org.agrona.collections.Long2LongHashMap;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
@ -105,4 +104,4 @@ public class WatermarkCollectorTest
return map;
};
}
}
}