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

Create a fuzz test that randomizes topology changes, cluster actions, and CQL operations 

patch by David Capwell; reviewed by Alex Petrov for CASSANDRA-19847
This commit is contained in:
David Capwell 2024-08-30 09:06:39 -07:00
parent a21c0a75aa
commit f1db115e73
5 changed files with 127 additions and 214 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit e2ccee4f51fe4c7c7f3ea8911897135ed7e37114
Subproject commit a171322f417c117733ca5b514d03a5202b1ac202

129
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
View File

@ -22,7 +22,6 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@ -49,7 +48,6 @@ import accord.primitives.TxnId;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.Property.Command;
import accord.utils.Property.Commands;
import accord.utils.Property.UnitCommand;
import accord.utils.RandomSource;
import org.agrona.collections.Int2ObjectHashMap;
@ -73,6 +71,7 @@ import org.apache.cassandra.utils.RTree;
import org.apache.cassandra.utils.RangeTree;
import org.assertj.core.api.Assertions;
import static accord.utils.Property.commands;
import static accord.utils.Property.stateful;
public class RouteIndexTest extends CQLTester.InMemory
@ -109,85 +108,61 @@ public class RouteIndexTest extends CQLTester.InMemory
cfs().disableAutoCompaction(); // let the test control compaction
//TODO (coverage): include with the ability to mark ranges as durable for compaction cleanup
AccordService.unsafeSetNoop(); // disable accord service since compaction touches it. It would be nice to include this for cleanup support....
stateful().withExamples(50).check(new Commands<State, ColumnFamilyStore>()
stateful().withExamples(50).check(commands(() -> State::new, i -> cfs())
.destroySut(sut -> sut.truncateBlocking())
.add(FLUSH)
.add(COMPACT)
.add((rs, state) -> {
int storeId = rs.nextInt(0, state.numStores);
Domain domain = state.domainGen.next(rs);
TxnId txnId = state.nextTxnId(domain);
Route<?> route = createRoute(state, rs, domain, rs.nextInt(1, 20));
return new InsertTxn(storeId, txnId, SaveStatus.PreAccepted, Durability.NotDurable, route);
})
.add((rs, state) -> new RangeSearch(rs.nextInt(0, state.numStores), state.rangeGen.next(rs)))
.addIf(state -> !state.storeToTableToRangesToTxns.isEmpty(), RouteIndexTest::rangeSearch)
.build());
}
private static RangeSearch rangeSearch(RandomSource rs, State state)
{
int storeId = rs.pickUnorderedSet(state.storeToTableToRangesToTxns.keySet());
var tables = state.storeToTableToRangesToTxns.get(storeId);
TableId tableId = rs.pickUnorderedSet(tables.keySet());
var ranges = tables.get(tableId);
TreeSet<TokenRange> distinctRanges = ranges.stream().map(Map.Entry::getKey).collect(Collectors.toCollection(() -> new TreeSet<>(TokenRange::compareTo)));
TokenRange range;
if (distinctRanges.size() == 1)
{
@Override
public Gen<State> genInitialState()
range = Iterables.getFirst(distinctRanges, null);
}
else
{
switch (rs.nextInt(0, 2))
{
return rs -> new State(rs);
}
@Override
public ColumnFamilyStore createSut(State state)
{
return cfs();
}
@Override
public Gen<Command<State, ColumnFamilyStore, ?>> commands(State state)
{
Map<Gen<Command<State, ColumnFamilyStore, ?>>, Integer> possible = new LinkedHashMap<>();
possible.put(ignore -> FLUSH, 1);
possible.put(ignore -> COMPACT, 1);
possible.put(rs -> {
int storeId = rs.nextInt(0, state.numStores);
Domain domain = state.domainGen.next(rs);
TxnId txnId = state.nextTxnId(domain);
Route<?> route = createRoute(state, rs, domain, rs.nextInt(1, 20));
return new InsertTxn(storeId, txnId, SaveStatus.PreAccepted, Durability.NotDurable, route);
}, 10);
possible.put(rs -> new RangeSearch(rs.nextInt(0, state.numStores), state.rangeGen.next(rs)), 1);
if (!state.storeToTableToRangesToTxns.isEmpty())
case 0: // perfect match
range = rs.pickOrderedSet(distinctRanges);
break;
case 1: // mutli-match
{
possible.put(rs -> {
int storeId = rs.pickUnorderedSet(state.storeToTableToRangesToTxns.keySet());
var tables = state.storeToTableToRangesToTxns.get(storeId);
TableId tableId = rs.pickUnorderedSet(tables.keySet());
var ranges = tables.get(tableId);
TreeSet<TokenRange> distinctRanges = ranges.stream().map(Map.Entry::getKey).collect(Collectors.toCollection(() -> new TreeSet<>(TokenRange::compareTo)));
TokenRange range;
if (distinctRanges.size() == 1)
{
range = Iterables.getFirst(distinctRanges, null);
}
else
{
switch (rs.nextInt(0, 2))
{
case 0: // perfect match
range = rs.pickOrderedSet(distinctRanges);
break;
case 1: // mutli-match
{
TokenRange a = rs.pickOrderedSet(distinctRanges);
TokenRange b = rs.pickOrderedSet(distinctRanges);
while (a.equals(b))
b = rs.pickOrderedSet(distinctRanges);
if (b.compareTo(a) < 0)
{
TokenRange tmp = a;
a = b;
b = tmp;
}
range = new TokenRange((AccordRoutingKey) a.start(), (AccordRoutingKey) b.end());
}
break;
default:
throw new AssertionError();
}
}
return new RangeSearch(storeId, range);
}, 5);
TokenRange a = rs.pickOrderedSet(distinctRanges);
TokenRange b = rs.pickOrderedSet(distinctRanges);
while (a.equals(b))
b = rs.pickOrderedSet(distinctRanges);
if (b.compareTo(a) < 0)
{
TokenRange tmp = a;
a = b;
b = tmp;
}
range = new TokenRange((AccordRoutingKey) a.start(), (AccordRoutingKey) b.end());
}
return Gens.oneOf(possible);
break;
default:
throw new AssertionError();
}
@Override
public void destroySut(ColumnFamilyStore sut)
{
cfs().truncateBlocking();
}
});
}
return new RangeSearch(storeId, range);
}
private static ColumnFamilyStore cfs()
@ -343,7 +318,7 @@ public class RouteIndexTest extends CQLTester.InMemory
}
}
private class RangeSearch implements Command<State, ColumnFamilyStore, Set<TxnId>>
private static class RangeSearch implements Command<State, ColumnFamilyStore, Set<TxnId>>
{
private final int storeId;
private final TokenRange range;

5
test/unit/org/apache/cassandra/io/util/ChecksumedDataTest.java
View File

@ -24,6 +24,8 @@ import java.util.function.Supplier;
import java.util.zip.CRC32C;
import java.util.zip.Checksum;
import javax.annotation.Nullable;
import org.junit.Test;
import accord.utils.Gen;
@ -149,8 +151,9 @@ public class ChecksumedDataTest
}
@Override
public void destroySut(List<StatefulChecksumCommand<?>> sut) throws Throwable
public void destroySut(List<StatefulChecksumCommand<?>> sut, @Nullable Throwable t) throws Throwable
{
if (t != null) return;
ChecksumedDataInputPlus in = new ChecksumedDataInputPlus(new DataInputBuffer(out.unsafeGetBufferAndFlip(), false), CHECKSUM_SUPPLIER);
for (StatefulChecksumCommand<?> cmd : sut)
{

107
test/unit/org/apache/cassandra/service/accord/EpochSyncTest.java
View File

@ -28,7 +28,6 @@ import java.util.EnumMap;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
@ -60,10 +59,7 @@ import accord.primitives.Ranges;
import accord.topology.Topology;
import accord.topology.TopologyManager;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.Invariants;
import accord.utils.Property.Command;
import accord.utils.Property.Commands;
import accord.utils.Property.UnitCommand;
import accord.utils.RandomSource;
import accord.utils.async.AsyncChain;
@ -109,6 +105,7 @@ import org.apache.cassandra.utils.ByteArrayUtil;
import org.apache.cassandra.utils.Pair;
import org.assertj.core.api.Assertions;
import static accord.utils.Property.commands;
import static accord.utils.Property.stateful;
public class EpochSyncTest
@ -126,72 +123,48 @@ public class EpochSyncTest
@Test
public void test()
{
stateful().withExamples(50).check(new Commands<Cluster, Void>()
{
@Override
public Gen<Cluster> genInitialState()
{
return Cluster::new;
}
stateful().withExamples(50).check(commands(() -> Cluster::new)
.destroyState(cluster -> {
cluster.processAll();
cluster.validate(true);
})
.addIf(cluster -> cluster.alive().size() <= cluster.maxNodes, EpochSyncTest::addNode)
.addIf(cluster -> cluster.alive().size() > cluster.minNodes, EpochSyncTest::removeNode)
.addIf(cluster -> cluster.hasWork(), EpochSyncTest::processSome)
.add(rs -> new SimpleCommand("Validate", c -> c.validate(false)))
.add((rs, cluster) -> new SimpleCommand("Bump Epoch " + (cluster.current.epoch.getEpoch() + 1), Cluster::bumpEpoch))
.build());
}
@Override
public Void createSut(Cluster Cluster)
{
return null;
}
private static SimpleCommand addNode(RandomSource rs, Cluster cluster)
{
Node.Id id = new Node.Id(++cluster.nodeCounter);
long token = cluster.tokenGen.nextLong(rs);
while (cluster.tokens.contains(token))
token = cluster.tokenGen.nextLong(rs);
long epoch = cluster.current.epoch.getEpoch() + 1;
long finalToken = token;
return new SimpleCommand("Add Node " + id + "; token=" + token + ", epoch=" + epoch,
c -> c.addNode(id, finalToken));
}
@Override
public Gen<Command<Cluster, Void, ?>> commands(Cluster cluster)
{
List<Node.Id> alive = cluster.alive();
Map<Gen<Command<Cluster, Void, ?>>, Integer> possible = new LinkedHashMap<>();
if (alive.size() < cluster.maxNodes)
{
// add node
possible.put(rs -> {
Node.Id id = new Node.Id(++cluster.nodeCounter);
long token = cluster.tokenGen.nextLong(rs);
while (cluster.tokens.contains(token))
token = cluster.tokenGen.nextLong(rs);
long epoch = cluster.current.epoch.getEpoch() + 1;
long finalToken = token;
return new SimpleCommand("Add Node " + id + "; token=" + token + ", epoch=" + epoch,
c -> c.addNode(id, finalToken));
}, 5);
}
if (alive.size() > cluster.minNodes)
{
possible.put(rs -> {
Node.Id pick = rs.pick(alive);
long token = cluster.instances.get(pick).token;
long epoch = cluster.current.epoch.getEpoch() + 1;
return new SimpleCommand("Remove Node " + pick + "; token=" + token + "; epoch=" + epoch, c -> c.removeNode(pick));
}, 3);
}
if (cluster.hasWork())
{
possible.put(rs -> new SimpleCommand("Process Some",
c -> {//noinspection StatementWithEmptyBody
for (int i = 0, attempts = rs.nextInt(1, 100); i < attempts && c.processOne(); i++)
{
}
}), 10);
}
private static SimpleCommand removeNode(RandomSource rs, Cluster cluster)
{
List<Node.Id> alive = cluster.alive();
Node.Id pick = rs.pick(alive);
long token = cluster.instances.get(pick).token;
long epoch = cluster.current.epoch.getEpoch() + 1;
return new SimpleCommand("Remove Node " + pick + "; token=" + token + "; epoch=" + epoch, c -> c.removeNode(pick));
}
possible.put(rs -> new SimpleCommand("Validate",
c -> c.validate(false)), 1);
possible.put(rs -> new SimpleCommand("Bump Epoch " + (cluster.current.epoch.getEpoch() + 1),
Cluster::bumpEpoch), 10);
return Gens.oneOf(possible);
}
@Override
public void destroyState(Cluster cluster)
{
cluster.processAll();
cluster.validate(true);
}
});
private static SimpleCommand processSome(RandomSource rs)
{
return new SimpleCommand("Process Some",
c -> {//noinspection StatementWithEmptyBody
for (int i = 0, attempts = rs.nextInt(1, 100); i < attempts && c.processOne(); i++)
{
}
});
}
private static class SimpleCommand implements UnitCommand<Cluster, Void>

98
test/unit/org/apache/cassandra/utils/StatefulRangeTreeTest.java
View File

@ -21,7 +21,6 @@ package org.apache.cassandra.utils;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;
@ -35,12 +34,12 @@ import accord.primitives.Range;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.Property.Command;
import accord.utils.Property.Commands;
import accord.utils.Property.UnitCommand;
import accord.utils.RandomSource;
import org.apache.cassandra.service.accord.RangeTreeRangeAccessor;
import org.assertj.core.api.Assertions;
import static accord.utils.Property.commands;
import static accord.utils.Property.stateful;
public class StatefulRangeTreeTest
@ -61,7 +60,7 @@ public class StatefulRangeTreeTest
/**
* Stateful test for RTree.
*
* <p>
* This test is very similar to {@link RangeTreeTest#test} but is fully mutable, so can not
* use the immutable search trees (else rebuidling becomes a large cost). Both tests should exist as they use different
* models, which helps build confidence that the RTree does the correct thing; that test also covers start and end
@ -70,57 +69,25 @@ public class StatefulRangeTreeTest
@Test
public void test()
{
stateful().check(new Commands<State, Sut>()
{
@Override
public Gen<State> genInitialState()
{
return rs -> {
Gen<Range> rangeGen = rangeGen(rs);
int numChildren = NUM_CHILDREN_GEN.nextInt(rs);
int sizeTarget = SIZE_TARGET_DISTRIBUTION.next(rs).filter(s -> s > numChildren).nextInt(rs);
int createWeight = rs.nextInt(1, 100);
int updateWeight = rs.nextInt(1, 20);
int deleteWeight = rs.nextInt(1, 20);
int clearWeight = rs.nextInt(0, 2); // either disabled or enabled with weight=1
int readWeight = rs.nextInt(1, 20);
return new State(sizeTarget, numChildren,
TOKEN_DISTRIBUTION.next(rs), rangeGen,
createWeight, updateWeight, deleteWeight, clearWeight, readWeight);
};
}
@Override
public Sut createSut(State state)
{
return new Sut(state.sizeTarget, state.numChildren);
}
@Override
public Gen<Command<State, Sut, ?>> commands(State state)
{
Map<Gen<Command<State, Sut, ?>>, Integer> possible = new LinkedHashMap<>();
possible.put(rs -> new Create(state.newRange(rs), SMALL_INT_GEN.nextInt(rs)), state.createWeight);
possible.put(rs -> new Read(state.newRange(rs)), state.readWeight);
possible.put(rs -> new KeyRead(IntKey.routing(state.tokenGen.nextInt(rs))), state.readWeight);
possible.put(rs -> new RangeRead(state.rangeGen.next(rs)), state.readWeight);
possible.put(ignore -> Iterate.instance, state.readWeight);
possible.put(ignore -> Clear.instance, state.clearWeight);
if (!state.uniqRanges.isEmpty())
{
possible.put(rs -> new Read(rs.pickOrderedSet(state.uniqRanges)), state.readWeight);
possible.put(rs -> {
Range range = rs.pickOrderedSet(state.uniqRanges);
int token = rs.nextInt(((IntKey.Routing) range.start()).key, ((IntKey.Routing) range.end()).key) + 1;
return new KeyRead(IntKey.routing(token));
}, state.readWeight);
possible.put(rs -> new RangeRead(rs.pickOrderedSet(state.uniqRanges)), state.readWeight);
possible.put(rs -> new Update(rs.pickOrderedSet(state.uniqRanges), SMALL_INT_GEN.nextInt(rs)), state.updateWeight);
possible.put(rs -> new Delete(rs.pickOrderedSet(state.uniqRanges)), state.deleteWeight);
}
return Gens.oneOf(possible);
}
});
stateful().check(commands(() -> State::new, state -> new Sut(state.sizeTarget, state.numChildren))
.add((rs, state) -> new Create(state.newRange(rs), SMALL_INT_GEN.nextInt(rs)))
.add((rs, state) -> new Read(state.newRange(rs)))
.add((rs, state) -> new KeyRead(IntKey.routing(state.tokenGen.nextInt(rs))))
.add((rs, state) -> new RangeRead(state.rangeGen.next(rs)))
.add(Iterate.instance)
.add(Clear.instance)
.addAllIf(state -> !state.uniqRanges.isEmpty(),
b -> b.add((rs, state) -> new Read(rs.pickOrderedSet(state.uniqRanges)))
.add((rs, state) -> {
Range range = rs.pickOrderedSet(state.uniqRanges);
int token = rs.nextInt(((IntKey.Routing) range.start()).key, ((IntKey.Routing) range.end()).key) + 1;
return new KeyRead(IntKey.routing(token));
})
.add((rs, state) -> new RangeRead(rs.pickOrderedSet(state.uniqRanges)))
.add((rs, state) -> new Update(rs.pickOrderedSet(state.uniqRanges), SMALL_INT_GEN.nextInt(rs)))
.add((rs, state) -> new Delete(rs.pickOrderedSet(state.uniqRanges)))
)
.build());
}
private static Gen<Range> rangeGen(RandomSource rand)
@ -401,6 +368,7 @@ public class StatefulRangeTreeTest
static class Iterate extends AbstractRead<Map.Entry<Range, Integer>>
{
static final Iterate instance = new Iterate();
public Iterate()
{
super(COMPARATOR);
@ -432,27 +400,21 @@ public class StatefulRangeTreeTest
private final int sizeTarget, numChildren;
private final Gen.IntGen tokenGen;
private final Gen<Range> rangeGen;
private final int createWeight, updateWeight, deleteWeight, clearWeight, readWeight;
private State(int sizeTarget, int numChildren,
Gen.IntGen tokenGen, Gen<Range> rangeGen,
int createWeight, int updateWeight, int deleteWeight, int clearWeight, int readWeight)
private State(RandomSource rs)
{
this.sizeTarget = sizeTarget;
this.numChildren = numChildren;
this.tokenGen = tokenGen;
this.rangeGen = rangeGen;
this.createWeight = createWeight;
this.updateWeight = updateWeight;
this.deleteWeight = deleteWeight;
this.clearWeight = clearWeight;
this.readWeight = readWeight;
this.numChildren = NUM_CHILDREN_GEN.nextInt(rs);
this.sizeTarget = SIZE_TARGET_DISTRIBUTION.next(rs).filter(s -> s > numChildren).nextInt(rs);
this.tokenGen = TOKEN_DISTRIBUTION.next(rs);
this.rangeGen = rangeGen(rs);
}
public Range newRange(RandomSource rs)
{
Range range;
while ((uniqRanges.contains(range = rangeGen.next(rs)))) {}
while ((uniqRanges.contains(range = rangeGen.next(rs))))
{
}
return range;
}