From 0a4777dae41f260b5de243901e7fdef5b9440105 Mon Sep 17 00:00:00 2001 From: David Capwell Date: Fri, 2 May 2025 17:00:20 -0700 Subject: [PATCH] zero copy streaming allocates direct memory that isnt used, but does help to fragment the memory space patch by David Capwell; reviewed by Yifan Cai for CASSANDRA-20577 --- CHANGES.txt | 1 + modules/accord | 2 +- .../cassandra/config/DatabaseDescriptor.java | 9 + .../test/cql3/MultiNodeTableWalkBase.java | 7 + .../MultiNodeTableWalkWithReadRepairTest.java | 3 - .../test/cql3/MultiNodeTokenConflictTest.java | 1 + .../test/cql3/SingleNodeTableWalkTest.java | 14 +- .../test/cql3/StatefulASTBase.java | 174 +++++++++++---- .../fuzz/topology/TopologyMixupTestBase.java | 42 +--- .../apache/cassandra/repair/FuzzTestBase.java | 63 +----- .../cassandra/repair/RepairGenerators.java | 201 ++++++++++++++++++ .../cassandra/utils/ImmutableUniqueList.java | 8 + .../cassandra/utils/LoggingCommand.java | 76 +++++++ 13 files changed, 458 insertions(+), 143 deletions(-) create mode 100644 test/unit/org/apache/cassandra/repair/RepairGenerators.java create mode 100644 test/unit/org/apache/cassandra/utils/LoggingCommand.java diff --git a/CHANGES.txt b/CHANGES.txt index 35cafede97..f214877478 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -186,6 +186,7 @@ * Add the ability to disable bulk loading of SSTables (CASSANDRA-18781) * Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787) Merged from 5.0: + * zero copy streaming allocates direct memory that isn't used, but does help to fragment the memory space (CASSANDRA-20577) * CQLSSTableWriter supports setting the format (BTI or Big) (CASSANDRA-20609) * Don't allocate in ThreadLocalReadAheadBuffer#close() (CASSANDRA-20551) * Ensure RowFilter#isMutableIntersection() properly evaluates numeric ranges on a single column (CASSANDRA-20566) diff --git a/modules/accord b/modules/accord index 3825403cc5..7f95490b13 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 3825403cc50ef7897d5dfb4cbdca5efbc432e8ee +Subproject commit 7f95490b1390b7fc68a4ff4ced7f161bafd8776b diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 82a26602eb..625fe2be10 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -5711,6 +5711,15 @@ public class DatabaseDescriptor return Objects.requireNonNull(selectedSSTableFormat, "Forgot to initialize DatabaseDescriptor?"); } + @VisibleForTesting + public static void setSelectedSSTableFormat(String name) + { + SSTableFormat format = getSSTableFormats().get(name); + if (format == null) + throw new IllegalArgumentException("Unknown sstable format: " + name); + setSelectedSSTableFormat(format); + } + @VisibleForTesting public static void setSelectedSSTableFormat(SSTableFormat format) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkBase.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkBase.java index d6c0183473..126f9ec908 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkBase.java @@ -66,6 +66,7 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest @Override protected void clusterConfig(IInstanceConfig c) { + super.clusterConfig(c); c.set("range_request_timeout", "180s") .set("read_request_timeout", "180s") .set("write_request_timeout", "180s") @@ -100,6 +101,12 @@ public abstract class MultiNodeTableWalkBase extends SingleNodeTableWalkTest return true; } + @Override + protected boolean allowRepair() + { + return hasEnoughMemtableForRepair() || hasEnoughSSTablesForRepair(); + } + @Override protected IInvokableInstance selectInstance(RandomSource rs) { diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithReadRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithReadRepairTest.java index 7727e3a76a..a864766871 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithReadRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTableWalkWithReadRepairTest.java @@ -18,13 +18,10 @@ package org.apache.cassandra.distributed.test.cql3; -import org.junit.Ignore; - import accord.utils.Property; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.service.reads.repair.ReadRepairStrategy; -@Ignore("In order to stay stable RR tests are ignored for now. Once Single node and multi node w/o RR are stable, then this test should be enabled to include RR testing") public class MultiNodeTableWalkWithReadRepairTest extends MultiNodeTableWalkBase { public MultiNodeTableWalkWithReadRepairTest() diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTokenConflictTest.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTokenConflictTest.java index 7a6dbaa859..081a200618 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTokenConflictTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/MultiNodeTokenConflictTest.java @@ -63,6 +63,7 @@ public class MultiNodeTokenConflictTest extends SingleNodeTokenConflictTest @Override protected void clusterConfig(IInstanceConfig c) { + super.clusterConfig(c); c.set("range_request_timeout", "180s") .set("read_request_timeout", "180s") .set("write_request_timeout", "180s") diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java index 762a2b83bc..171ccf140c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java @@ -60,6 +60,7 @@ import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.test.sai.SAIUtil; +import org.apache.cassandra.utils.LoggingCommand; import org.apache.cassandra.harry.model.BytesPartitionState; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.TableMetadata; @@ -322,7 +323,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase Select select = builder.build(); String annotate = cols.stream().map(symbol -> { var indexed = state.indexes.get(symbol); - return symbol.detailedName() + (indexed == null ? "" : " (indexed with " + indexed.indexDDL.indexer.name() + ")"); + return symbol.detailedName() + (indexed == null ? "" : " (indexed with " + indexed.indexDDL.indexer.name() + ')'); }).collect(Collectors.joining(", ")); return state.command(rs, select, annotate); } @@ -367,15 +368,20 @@ public class SingleNodeTableWalkTest extends StatefulASTBase .add(StatefulASTBase::insert) .add(StatefulASTBase::fullTableScan) .addIf(State::hasPartitions, this::selectExisting) - .addAllIf(State::supportTokens, b -> b.add(this::selectToken) - .add(this::selectTokenRange) - .add(StatefulASTBase::selectMinTokenRange)) + .addAllIf(State::supportTokens, + this::selectToken, + this::selectTokenRange, + StatefulASTBase::selectMinTokenRange) .addIf(State::hasEnoughMemtable, StatefulASTBase::flushTable) .addIf(State::hasEnoughSSTables, StatefulASTBase::compactTable) + .addAllIf(BaseState::allowRepair, + StatefulASTBase::incrementalRepair, + StatefulASTBase::previewRepair) .addIf(State::allowNonPartitionQuery, this::nonPartitionQuery) .addIf(State::allowNonPartitionMultiColumnQuery, this::multiColumnQuery) .addIf(State::allowPartitionQuery, this::partitionRestrictedQuery) .destroyState(State::close) + .commandsTransformer(LoggingCommand.factory()) .onSuccess(onSuccess(logger)) .build()); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java index a16b47c1d7..527a7ea658 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java @@ -22,9 +22,9 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Arrays; import java.util.EnumSet; import java.util.List; -import java.util.Objects; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; @@ -34,6 +34,7 @@ import java.util.stream.Stream; import javax.annotation.Nullable; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; import com.google.common.collect.Maps; import org.slf4j.Logger; @@ -69,6 +70,7 @@ import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.AsciiType; import org.apache.cassandra.db.marshal.BytesType; import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.Feature; @@ -80,6 +82,9 @@ import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.exceptions.RequestFailureReason; import org.apache.cassandra.harry.model.ASTSingleTableModel; import org.apache.cassandra.harry.util.StringUtils; +import org.apache.cassandra.repair.RepairGenerators; +import org.apache.cassandra.repair.RepairGenerators.PreviewType; +import org.apache.cassandra.repair.RepairGenerators.RepairType; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.AbstractTypeGenerators; import org.apache.cassandra.utils.CassandraGenerators; @@ -118,6 +123,8 @@ public class StatefulASTBase extends TestBaseImpl .collect(Collectors.toList())); protected static final Gen FETCH_SIZE_DISTRO = Gens.mixedDistribution(new int[] {1, 10, 100, 1000, 5000}); protected static final Gen LIMIT_DISTRO = Gens.mixedDistribution(1, 1001); + protected static final Gen REPAIR_TYPE_EMPTY_MODEL_DISTRO = Gens.mixedDistribution(0, 2); + protected static final Gen REPAIR_TYPE_DISTRO = Gens.mixedDistribution(0, 3); static { @@ -145,7 +152,7 @@ public class StatefulASTBase extends TestBaseImpl protected void clusterConfig(IInstanceConfig config) { - + config.set("repair.retries.max_attempts", Integer.MAX_VALUE); } protected void clusterInitializer(ClassLoader cl, int node) @@ -185,7 +192,7 @@ public class StatefulASTBase extends TestBaseImpl protected static Property.Command flushTable(RandomSource rs, S state) { - return new Property.SimpleCommand<>("nodetool flush " + state.metadata.keyspace + " " + state.metadata.name, s2 -> { + return new Property.SimpleCommand<>("nodetool flush " + state.metadata.keyspace + ' ' + state.metadata.name, s2 -> { s2.cluster.forEach(i -> i.nodetoolResult("flush", s2.metadata.keyspace, s2.metadata.name).asserts().success()); s2.flush(); }); @@ -193,7 +200,7 @@ public class StatefulASTBase extends TestBaseImpl protected static Property.Command compactTable(RandomSource rs, S state) { - return new Property.SimpleCommand<>("nodetool compact " + state.metadata.keyspace + " " + state.metadata.name, s2 -> { + return new Property.SimpleCommand<>("nodetool compact " + state.metadata.keyspace + ' ' + state.metadata.name, s2 -> { state.cluster.forEach(i -> i.nodetoolResult("compact", s2.metadata.keyspace, s2.metadata.name).asserts().success()); s2.compact(); }); @@ -211,6 +218,59 @@ public class StatefulASTBase extends TestBaseImpl state.commandSafeRandomHistory(selectForMutation(state, mutation), "Select for Mutation Validation")); } + protected static Property.Command incrementalRepair(RandomSource rs, S state) + { + return repair(rs, state, state.repairArgsBuilder().withType(i -> RepairType.IR).withPreviewType(i -> PreviewType.NONE), null); + } + + protected static Property.Command previewRepair(RandomSource rs, S state) + { + return repair(rs, state, state.repairArgsBuilder().withType(i -> RepairType.FULL).withPreviewType(i -> PreviewType.REPAIRED), null); + } + + protected static Property.Command repair(RandomSource rs, S state, RepairGenerators.Builder argsBuilder, @Nullable String annotate) + { + IInvokableInstance inst = state.selectInstance(rs); + Gen> argsGen = argsBuilder.build(); + List args = ImmutableList.builder() + .add("repair") + .addAll(argsGen.next(rs)) + .build(); + boolean preview = RepairGenerators.isPreview(args); + // mimic org.apache.cassandra.repair.state.CoordinatorState.getType + String type; + if (preview) + { + // mimic org.apache.cassandra.tools.nodetool.Repair.getPreviewKind + PreviewType previewType = RepairGenerators.previewType(args); + switch (previewType) + { + case REPAIRED: + type = "preview repaired"; + break; + case UNREPAIRED: + type = RepairGenerators.isFull(args) ? "preview full" : "preview unrepaired"; + break; + default: + throw new UnsupportedOperationException(previewType.name()); + } + } + else + { + type = RepairGenerators.isFull(args) ? "full" : "incremental"; + } + + String postfix = "type " + type + ", on " + inst; + if (annotate == null) annotate = postfix; + else annotate += ", " + postfix; + + return new Property.SimpleCommand<>("nodetool " + String.join(" ", args) + " -- " + annotate, s2 -> { + inst.nodetoolResult(args.toArray(String[]::new)).asserts().success(); + if (!preview) + s2.repair(); + }); + } + private static Select selectForMutation(S state, Mutation mutation) { var select = Select.builder(state.metadata).allowFiltering(); @@ -288,16 +348,19 @@ public class StatefulASTBase extends TestBaseImpl protected final Gen lessThanGen; protected final Gen greaterThanGen; protected final Gen rangeInequalityGen; + protected final Gen.IntGen repairTypeEmptyModelGen, repairTypeGen; protected final Gen.IntGen fetchSizeGen; protected final TableMetadata metadata; protected final TableReference tableRef; protected final ASTSingleTableModel model; + private final String sstableFormatName; private final Visitor debug; - private final int enoughMemtables; - private final int enoughSSTables; + private final int enoughMemtables, enoughMemtablesForRepair; + private final int enoughSSTables, enoughSSTablesForRepair; protected int numMutations, mutationsSinceLastFlush; - protected int numFlushes, flushesSinceLastCompaction; + protected int numFlushes, flushesSinceLastCompaction, flushesSinceLastRepair; protected int numCompact; + protected int numRepairs; protected int operations; protected BaseState(RandomSource rs, Cluster cluster, TableMetadata metadata) @@ -322,13 +385,21 @@ public class StatefulASTBase extends TestBaseImpl this.perPartitionLimitGen = LIMIT_DISTRO.next(rs); this.limitGen = LIMIT_DISTRO.next(rs); - this.enoughMemtables = rs.pickInt(3, 10, 50); + this.repairTypeEmptyModelGen = REPAIR_TYPE_EMPTY_MODEL_DISTRO.next(rs); + this.repairTypeGen = REPAIR_TYPE_DISTRO.next(rs); + + this.enoughMemtables = rs.pickInt(1, 3, 10, 50); + this.enoughMemtablesForRepair = rs.pickInt(1, 3, 10, 50); this.enoughSSTables = rs.pickInt(3, 10, 50); + this.enoughSSTablesForRepair = rs.pickInt(1, 3, 10, 50); this.metadata = metadata; this.tableRef = TableReference.from(metadata); this.model = new ASTSingleTableModel(metadata, IGNORED_ISSUES); createTable(metadata); + + String sstableFormatName = this.sstableFormatName = Generators.toGen(CassandraGenerators.sstableFormatNames()).next(rs); + cluster.forEach(i -> i.runOnInstance(() -> DatabaseDescriptor.setSelectedSSTableFormat(sstableFormatName))); } public boolean hasPartitions() @@ -364,6 +435,35 @@ public class StatefulASTBase extends TestBaseImpl return command(rs, select, null); } + protected boolean allowRepair() + { + return false; + } + + protected RepairGenerators.Builder repairArgsBuilder() + { + return new RepairGenerators.Builder(i -> Arrays.asList(metadata.keyspace, metadata.name)) + // paxos cleanup's finish prepare is delayed based off CAS/Write timeout, but these tests make that 3 minutes (so CI is stable) + // which means this step is delayed 3 minutes, making repairs suppppper slow... + // see org.apache.cassandra.service.paxos.cleanup.PaxosCleanup#finishPrepare + .withSkipPaxosGen(i -> true) + .withRanges(rs -> { + switch (model.isEmpty() ? repairTypeEmptyModelGen.next(rs) : repairTypeGen.next(rs)) + { + case 0: return RepairGenerators.LOCAL_RANGE; + case 1: return RepairGenerators.PRIMARY_RANGE; + case 2: + { + Token a = rs.pickOrderedSet(model.partitionKeys()).token; + return List.of("--start-token", Long.toString(a.getLongValue() - 1), + "--end-token", a.toString()); + } + default: throw new UnsupportedOperationException(); + } + }) + ; + } + protected boolean allowLimit(Select select) { //TODO (coverage): allow this in the model! @@ -467,11 +567,23 @@ public class StatefulASTBase extends TestBaseImpl return mutationsSinceLastFlush > enoughMemtables; } + protected boolean hasEnoughMemtableForRepair() + { + // use last flush rather than last repair as this method cares about data in the memtable + // and not amount of mutations since repair + return mutationsSinceLastFlush > enoughMemtablesForRepair; + } + protected boolean hasEnoughSSTables() { return flushesSinceLastCompaction > enoughSSTables; } + protected boolean hasEnoughSSTablesForRepair() + { + return flushesSinceLastRepair > enoughSSTablesForRepair; + } + protected void mutation() { numMutations++; @@ -483,6 +595,7 @@ public class StatefulASTBase extends TestBaseImpl mutationsSinceLastFlush = 0; numFlushes++; flushesSinceLastCompaction++; + flushesSinceLastRepair++; } protected void compact() @@ -491,6 +604,15 @@ public class StatefulASTBase extends TestBaseImpl numCompact++; } + protected void repair() + { + if (mutationsSinceLastFlush > 0) + flush(); + + numRepairs++; + flushesSinceLastRepair = 0; + } + protected Value value(RandomSource rs, ByteBuffer bb, AbstractType type) { return bindOrLiteralGen.next(rs) ? new Bind(bb, type) : new Literal(bb, type); @@ -599,7 +721,8 @@ public class StatefulASTBase extends TestBaseImpl protected void toString(StringBuilder sb) { - sb.append(createKeyspaceCQL(metadata.keyspace)); + sb.append("Config:\nsstable:\n\tselected_format: ").append(sstableFormatName); + sb.append('\n').append(createKeyspaceCQL(metadata.keyspace)); CassandraGenerators.visitUDTs(metadata, udt -> sb.append('\n').append(udt.toCqlString(false, false, true)).append(';')); sb.append('\n').append(metadata.toCqlString(false, false, false)); } @@ -620,39 +743,6 @@ public class StatefulASTBase extends TestBaseImpl toString(sb); return sb.toString(); } - - private static final class ValueWithType - { - final ByteBuffer value; - final AbstractType type; - - private ValueWithType(ByteBuffer value, AbstractType type) - { - this.value = value; - this.type = type; - } - - @Override - public boolean equals(Object o) - { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - ValueWithType value1 = (ValueWithType) o; - return value.equals(value1.value) && type.equals(value1.type); - } - - @Override - public int hashCode() - { - return Objects.hash(value, type); - } - - @Override - public String toString() - { - return type.toCQLString(value); - } - } } protected static abstract class CommonState extends BaseState diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java b/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java index 180741c4f8..a4341dddeb 100644 --- a/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java +++ b/test/distributed/org/apache/cassandra/fuzz/topology/TopologyMixupTestBase.java @@ -84,8 +84,8 @@ import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tools.nodetool.formatter.TableBuilder; -import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.ConfigGenBuilder; +import org.apache.cassandra.utils.LoggingCommand; import org.apache.cassandra.utils.Retry; import static accord.utils.Property.commands; @@ -491,35 +491,6 @@ public abstract class TopologyMixupTestBase, Void, ?> apply(RandomSource rs, State state); } - private static class LoggingCommand extends Property.ForwardingCommand - { - private static final Logger logger = LoggerFactory.getLogger(LoggingCommand.class); - - private LoggingCommand(Command delegate) - { - super(delegate); - } - - @Override - public Result apply(State s) throws Throwable - { - String name = detailed(s); - long startNanos = Clock.Global.nanoTime(); - try - { - logger.info("Starting command: {}", name); - Result o = super.apply(s); - logger.info("Command {} was success after {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos)); - return o; - } - catch (Throwable t) - { - logger.warn("Command {} failed after {}: {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos), t.toString()); // don't want stack trace, just type/msg - throw t; - } - } - } - protected static class State implements AutoCloseable { final TopologyHistory topologyHistory; @@ -656,16 +627,7 @@ public abstract class TopologyMixupTestBase rs2 -> { - Command, Void, ?> c = commandGen.next(rs2); - if (!(c instanceof Property.MultistepCommand)) - return new LoggingCommand<>(c); - Property.MultistepCommand, Void> multistep = (Property.MultistepCommand, Void>) c; - List, Void, ?>> subcommands = new ArrayList<>(); - for (var sub : multistep) - subcommands.add(new LoggingCommand<>(sub)); - return multistep(subcommands); - }); + commandsTransformers.add(LoggingCommand.factory()); preActions.add(() -> { int[] up = topologyHistory.up(); // use the most recent node just in case the cluster isn't in-sync diff --git a/test/unit/org/apache/cassandra/repair/FuzzTestBase.java b/test/unit/org/apache/cassandra/repair/FuzzTestBase.java index 8ffd96795b..72148701de 100644 --- a/test/unit/org/apache/cassandra/repair/FuzzTestBase.java +++ b/test/unit/org/apache/cassandra/repair/FuzzTestBase.java @@ -47,6 +47,7 @@ import java.util.function.Function; import java.util.function.Supplier; import javax.annotation.Nullable; +import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; import com.google.common.collect.Maps; import com.google.common.collect.Sets; @@ -103,6 +104,8 @@ import org.apache.cassandra.net.MessageDelivery; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.SimulatedMessageDelivery; import org.apache.cassandra.net.SimulatedMessageDelivery.SimulatedMessageReceiver; +import org.apache.cassandra.repair.RepairGenerators.PreviewType; +import org.apache.cassandra.repair.RepairGenerators.RepairType; import org.apache.cassandra.repair.messages.RepairMessage; import org.apache.cassandra.repair.messages.RepairOption; import org.apache.cassandra.repair.messages.ValidationResponse; @@ -534,12 +537,6 @@ public abstract class FuzzTestBase extends CQLTester.InMemory } } - private enum RepairType - {FULL, IR} - - private enum PreviewType - {NONE, REPAIRED, UNREPAIRED} - static RepairOption repairOption(RandomSource rs, Cluster.Node coordinator, String ks, List tableNames) { return repairOption(rs, coordinator, ks, Gens.lists(Gens.pick(tableNames)).ofSizeBetween(1, tableNames.size()), Gens.enums().all(RepairType.class), Gens.enums().all(PreviewType.class), Gens.enums().all(RepairParallelism.class)); @@ -557,53 +554,13 @@ public abstract class FuzzTestBase extends CQLTester.InMemory private static RepairOption repairOption(RandomSource rs, Cluster.Node coordinator, String ks, Gen> tablesGen, Gen repairTypeGen, Gen previewTypeGen, Gen repairParallelismGen) { - RepairType type = repairTypeGen.next(rs); - PreviewType previewType = previewTypeGen.next(rs); - List args = new ArrayList<>(); - args.add(ks); - List tables = tablesGen.next(rs); - args.addAll(tables); - args.add("-pr"); - switch (type) - { - case IR: - // default - break; - case FULL: - args.add("--full"); - break; - default: - throw new AssertionError("Unsupported repair type: " + type); - } - switch (previewType) - { - case NONE: - break; - case REPAIRED: - args.add("--validate"); - break; - case UNREPAIRED: - args.add("--preview"); - break; - default: - throw new AssertionError("Unsupported preview type: " + previewType); - } - RepairParallelism parallelism = repairParallelismGen.next(rs); - switch (parallelism) - { - case SEQUENTIAL: - args.add("--sequential"); - break; - case PARALLEL: - // default - break; - case DATACENTER_AWARE: - args.add("--dc-parallel"); - break; - default: - throw new AssertionError("Unknown parallelism: " + parallelism); - } - if (rs.nextBoolean()) args.add("--optimise-streams"); + List args = new RepairGenerators.Builder(tablesGen.map(l -> ImmutableList.builderWithExpectedSize(l.size() + 1).add(ks).addAll(l).build())) + .withType(repairTypeGen) + .withPreviewType(previewTypeGen) + .withParallelism(repairParallelismGen) + .withRanges(i -> RepairGenerators.PRIMARY_RANGE) + .build() + .next(rs); RepairOption options = RepairOption.parse(Repair.parseOptionMap(() -> "test", args), DatabaseDescriptor.getPartitioner()); if (options.getRanges().isEmpty()) { diff --git a/test/unit/org/apache/cassandra/repair/RepairGenerators.java b/test/unit/org/apache/cassandra/repair/RepairGenerators.java new file mode 100644 index 0000000000..2175a37789 --- /dev/null +++ b/test/unit/org/apache/cassandra/repair/RepairGenerators.java @@ -0,0 +1,201 @@ +/* + * 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.repair; + +import java.util.ArrayList; +import java.util.List; + +import accord.utils.Gen; +import accord.utils.Gens; + +public class RepairGenerators +{ + public static final List LOCAL_RANGE = List.of(); + public static final List PRIMARY_RANGE = List.of("-pr"); // repair calls this partition range, but StorageService calls this primary + + public enum RepairType + { + FULL("--full"), + IR(""); + + public final String arg; + + RepairType(String s) + { + this.arg = s; + } + } + + public enum PreviewType + { + NONE(""), + REPAIRED("--validate"), + UNREPAIRED("--preview"); + + public final String arg; + + PreviewType(String s) + { + this.arg = s; + } + } + + public static boolean isPreview(List args) + { + return args.stream().anyMatch(s -> PreviewType.REPAIRED.arg.equals(s) + || PreviewType.UNREPAIRED.arg.equals(s)); + } + + public static PreviewType previewType(List args) + { + for (String s : args) + { + if (PreviewType.REPAIRED.arg.equals(s)) + return PreviewType.REPAIRED; + if (PreviewType.UNREPAIRED.arg.equals(s)) + return PreviewType.UNREPAIRED; + } + return PreviewType.NONE; + } + + public static boolean isFull(List args) + { + return args.stream().anyMatch(s -> RepairType.FULL.arg.equals(s)); + } + + public static boolean isIncremental(List args) + { + return !isFull(args); + } + + + public static class Builder + { + final Gen> tablesGen; + Gen typeGen = Gens.enums().all(RepairType.class); + Gen previewTypeGen = Gens.enums().all(PreviewType.class); + Gen> ranges = Gens.pick(List.of(), PRIMARY_RANGE); + Gen optimizeStreamsGen = Gens.bools().all(); + Gen parallelismGen = Gens.enums().all(RepairParallelism.class); + Gen skipPaxosGen = i -> false; + Gen skipAccordGen = i -> false; + + public Builder(Gen> tablesGen) + { + this.tablesGen = tablesGen; + } + + public Builder withType(Gen typeGen) + { + this.typeGen = typeGen; + return this; + } + + public Builder withPreviewType(Gen previewTypeGen) + { + this.previewTypeGen = previewTypeGen; + return this; + } + + public Builder withRanges(Gen> ranges) + { + this.ranges = ranges; + return this; + } + + public Builder withOptimizeStreams(Gen optimizeStreamsGen) + { + this.optimizeStreamsGen = optimizeStreamsGen; + return this; + } + + public Builder withParallelism(Gen parallelismGen) + { + this.parallelismGen = parallelismGen; + return this; + } + + public Builder withSkipPaxosGen(Gen skipPaxosGen) + { + this.skipPaxosGen = skipPaxosGen; + return this; + } + + public Builder withSkipAccordGen(Gen skipAccordGen) + { + this.skipAccordGen = skipAccordGen; + return this; + } + + public Gen> build() + { + return rs -> { + RepairType type = typeGen.next(rs); + PreviewType previewType = previewTypeGen.next(rs); + List args = new ArrayList<>(); + args.addAll(tablesGen.next(rs)); + args.addAll(ranges.next(rs)); + if (skipPaxosGen.next(rs)) + args.add("--skip-paxos"); + if (skipAccordGen.next(rs)) + args.add("--skip-accord"); + switch (type) + { + case IR: + // default + break; + case FULL: + args.add(type.arg); + break; + default: + throw new AssertionError("Unsupported repair type: " + type); + } + switch (previewType) + { + case NONE: + break; + case REPAIRED: + case UNREPAIRED: + args.add(previewType.arg); + break; + default: + throw new AssertionError("Unsupported preview type: " + previewType); + } + RepairParallelism parallelism = parallelismGen.next(rs); + switch (parallelism) + { + case SEQUENTIAL: + args.add("--sequential"); + break; + case PARALLEL: + // default + break; + case DATACENTER_AWARE: + args.add("--dc-parallel"); + break; + default: + throw new AssertionError("Unknown parallelism: " + parallelism); + } + if (optimizeStreamsGen.next(rs)) + args.add("--optimise-streams"); + return args; + }; + } + } +} diff --git a/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java b/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java index d4b7393dcd..7db8b56c18 100644 --- a/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java +++ b/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java @@ -62,6 +62,14 @@ public class ImmutableUniqueList extends AbstractList implements RandomAcc return (ImmutableUniqueList) EMPTY; } + public static ImmutableUniqueList of(T... values) + { + Builder builder = builder(values.length); + for (T v : values) + builder.add(v); + return builder.build(); + } + public AsSet asSet() { if (asSet != null) return asSet; diff --git a/test/unit/org/apache/cassandra/utils/LoggingCommand.java b/test/unit/org/apache/cassandra/utils/LoggingCommand.java new file mode 100644 index 0000000000..190192819b --- /dev/null +++ b/test/unit/org/apache/cassandra/utils/LoggingCommand.java @@ -0,0 +1,76 @@ +/* + * 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.utils; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.function.BiFunction; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.utils.Gen; +import accord.utils.Property; +import accord.utils.Property.Command; + +import static accord.utils.Property.multistep; + +public class LoggingCommand extends Property.ForwardingCommand +{ + private static final Logger logger = LoggerFactory.getLogger(LoggingCommand.class); + + public LoggingCommand(Command delegate) + { + super(delegate); + } + + public static BiFunction>, Gen>> factory() + { + return (state, commandGen) -> rs -> { + Command c = commandGen.next(rs); + if (!(c instanceof Property.MultistepCommand)) + return new LoggingCommand<>(c); + Property.MultistepCommand multistep = (Property.MultistepCommand) c; + List> subcommands = new ArrayList<>(); + for (var sub : multistep) + subcommands.add(new LoggingCommand<>(sub)); + return multistep(subcommands); + }; + } + + @Override + public Result apply(State s) throws Throwable + { + String name = detailed(s); + long startNanos = Clock.Global.nanoTime(); + try + { + logger.info("Starting command: {}", name); + Result o = super.apply(s); + logger.info("Command {} was success after {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos)); + return o; + } + catch (Throwable t) + { + logger.warn("Command {} failed after {}: {}", name, Duration.ofNanos(Clock.Global.nanoTime() - startNanos), t.toString()); // don't want stack trace, just type/msg + throw t; + } + } +}