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

Accord: Improve Tracing 

- Introduce pattern tracing, that can intercept failed or new coordinations matching various filters
     - Support additional tracing event collection modes (SAMPLE and RING)

    patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20911
This commit is contained in:
Benedict Elliott Smith 2025-09-14 14:28:04 +01:00
parent cf71b6ee96
commit e768418696
11 changed files with 1526 additions and 294 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 657f344eb3b3570966bf8cff7731bef6eeea98f1
Subproject commit e896c9c328d3e8a3b7ddd85381edeaf1b4b8ccb2

563
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

@ -27,24 +27,31 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.function.IntFunction;
import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import com.google.common.collect.ImmutableSet;
import accord.coordinate.AbstractCoordination;
import accord.coordinate.Coordination;
import accord.coordinate.Coordination.CoordinationKind;
import accord.coordinate.Coordinations;
import accord.coordinate.PrepareRecovery;
import accord.coordinate.tracking.AbstractTracker;
import accord.local.cfk.CommandsForKey.TxnInfo;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.RoutingKeys;
import accord.utils.SortedListMap;
import accord.utils.TinyEnumSet;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.db.marshal.TxnIdUtf8Type;
@ -52,7 +59,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.api.TraceEventType;
import accord.coordinate.FetchData;
import accord.coordinate.FetchRoute;
import accord.coordinate.MaybeRecover;
@ -114,9 +120,14 @@ import org.apache.cassandra.service.accord.AccordJournal;
import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.AccordTracing;
import org.apache.cassandra.service.accord.AccordTracing.BucketMode;
import org.apache.cassandra.service.accord.AccordTracing.CoordinationKinds;
import org.apache.cassandra.service.accord.AccordTracing.TracePattern;
import org.apache.cassandra.service.accord.AccordTracing.TxnKindsAndDomains;
import org.apache.cassandra.service.accord.DebugBlockedTxns;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.JournalKey;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.service.accord.api.AccordAgent;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState;
@ -124,7 +135,6 @@ import org.apache.cassandra.service.consensus.migration.TableMigrationState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.LocalizeString;
import static accord.api.TraceEventType.RECOVER;
import static accord.coordinate.Infer.InvalidIf.NotKnownToBeInvalid;
import static accord.local.RedundantStatus.Property.GC_BEFORE;
import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED;
@ -151,13 +161,14 @@ import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;
public class AccordDebugKeyspace extends VirtualKeyspace
{
public static final String COORDINATIONS = "coordinations";
public static final String EXECUTORS = "executors";
public static final String COMMANDS_FOR_KEY = "commands_for_key";
public static final String COMMANDS_FOR_KEY_UNMANAGED = "commands_for_key_unmanaged";
public static final String CONSTANTS = "constants";
public static final String COORDINATIONS = "coordinations";
public static final String DURABILITY_SERVICE = "durability_service";
public static final String DURABLE_BEFORE = "durable_before";
public static final String EXECUTOR_CACHE = "executor_cache";
public static final String EXECUTORS = "executors";
public static final String JOURNAL = "journal";
public static final String MAX_CONFLICTS = "max_conflicts";
public static final String MIGRATION_STATE = "migration_state";
@ -166,6 +177,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace
public static final String REJECT_BEFORE = "reject_before";
public static final String TXN = "txn";
public static final String TXN_BLOCKED_BY = "txn_blocked_by";
public static final String TXN_PATTERN_TRACE = "txn_pattern_trace";
public static final String TXN_PATTERN_TRACES = "txn_pattern_traces";
public static final String TXN_TRACE = "txn_trace";
public static final String TXN_TRACES = "txn_traces";
public static final String TXN_OPS = "txn_ops";
@ -194,11 +207,42 @@ public class AccordDebugKeyspace extends VirtualKeyspace
new TxnTable(),
new TxnTraceTable(),
new TxnTracesTable(),
new TxnOpsTable()
new TxnPatternTraceTable(),
new TxnPatternTracesTable(),
new TxnOpsTable(),
new ConstantsTable()
));
}
// TODO (desired): human readable packed key tracker (but requires loading Txn, so might be preferable to only do conditionally)
public static final class ConstantsTable extends AbstractVirtualTable
{
private final SimpleDataSet dataSet;
private ConstantsTable()
{
super(parse(VIRTUAL_ACCORD_DEBUG, CONSTANTS,
"Accord Debug Keyspace Constants",
"CREATE TABLE %s (\n" +
" kind text,\n" +
" name text,\n" +
" description text,\n" +
" PRIMARY KEY (kind, name)" +
')', UTF8Type.instance));
dataSet = new SimpleDataSet(metadata());
for (CoordinationKind coordinationKind : CoordinationKind.values())
dataSet.row("CoordinationKind", coordinationKind.name());
for (TxnOpsTable.Op op : TxnOpsTable.Op.values())
dataSet.row("Op", op.name()).column("description", op.description);
}
@Override
public DataSet data()
{
return dataSet;
}
}
public static final class ExecutorsTable extends AbstractLazyVirtualTable
{
private ExecutorsTable()
@ -913,7 +957,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
* UPDATE system_accord_debug.txn_trace SET permits = N WHERE txn_id = ? AND event_type = ?
* SELECT * FROM system_accord_debug.txn_traces WHERE txn_id = ? AND event_type = ?
*/
public static final class TxnTraceTable extends AbstractMutableVirtualTable
public static final class TxnTraceTable extends AbstractMutableLazyVirtualTable
{
private TxnTraceTable()
{
@ -921,63 +965,111 @@ public class AccordDebugKeyspace extends VirtualKeyspace
"Accord Transaction Trace Configuration",
"CREATE TABLE %s (\n" +
" txn_id 'TxnIdUtf8Type',\n" +
" event_type text,\n" +
" permits int,\n" +
" PRIMARY KEY (txn_id, event_type)" +
')', TxnIdUtf8Type.instance));
" bucket_mode text,\n" +
" bucket_seen int,\n" +
" bucket_size int,\n" +
" bucket_sub_size int,\n" +
" chance float,\n" +
" current_size int,\n" +
" trace_events text,\n" +
" managed_by_pattern boolean,\n" +
" PRIMARY KEY (txn_id)" +
')', TxnIdUtf8Type.instance), FAIL, UNSORTED);
}
@Override
public DataSet data()
protected void collect(PartitionsCollector collector)
{
AccordTracing tracing = tracing();
SimpleDataSet dataSet = new SimpleDataSet(metadata());
tracing.forEach(id -> true, (txnId, eventType, permits, events) -> {
dataSet.row(txnId.toString(), eventType.toString()).column("permits", permits);
tracing.forEach(id -> true, (txnId, events) -> {
collector.row(txnId.toString())
.eagerCollect(columns -> {
columns.add("bucket_mode", events.bucketMode().name())
.add("bucket_seen", events.bucketSeen())
.add("bucket_size", events.bucketSize())
.add("bucket_sub_size", events.bucketSubSize())
.add("chance", events.chance())
.add("current_size", events.size())
.add("trace_events", events.traceEvents(), TO_STRING)
.add("managed_by_pattern", events.hasOwner())
;
});
});
return dataSet;
}
private AccordTracing tracing()
{
return ((AccordAgent)AccordService.instance().agent()).tracing();
}
@Override
protected void applyPartitionDeletion(ColumnValues partitionKey)
protected void applyPartitionDeletion(Object[] partitionKey)
{
TxnId txnId = TxnId.parse(partitionKey.value(0));
tracing().erasePermits(txnId);
TxnId txnId = TxnId.parse((String)partitionKey[0]);
tracing().stopTracing(txnId);
}
@Override
protected void applyRowDeletion(ColumnValues partitionKey, ColumnValues clusteringColumns)
protected void applyRowDeletion(Object[] partitionKeys, Object[] clusteringKeys)
{
TxnId txnId = TxnId.parse(partitionKey.value(0));
tracing().erasePermits(txnId, parseEventType(clusteringColumns.value(0)));
TxnId txnId = TxnId.parse((String)partitionKeys[0]);
tracing().stopTracing(txnId);
}
@Override
protected void applyColumnDeletion(ColumnValues partitionKey, ColumnValues clusteringColumns, String columnName)
protected void applyRowUpdate(Object[] partitionKeys, @Nullable Object[] clusteringKeys, ColumnMetadata[] columns, Object[] values)
{
TxnId txnId = TxnId.parse(partitionKey.value(0));
TraceEventType eventType = parseEventType(clusteringColumns.value(0));
tracing().erasePermits(txnId, eventType);
}
TxnId txnId = TxnId.parse((String)partitionKeys[0]);
CoordinationKinds newTrace = null;
BucketMode newBucketMode = null;
boolean unsetManagedByOwner = false;
int newBucketSize = -1, newBucketSubSize = -1, newBucketSeen = -1;
float newChance = Float.NaN;
for (int i = 0 ; i < columns.length ; ++i)
{
String name = columns[i].name.toString();
switch (name)
{
default: throw new InvalidRequestException("Cannot update '" + name + '\'');
case "bucket_mode":
newBucketMode = checkBucketMode(values[i]);
break;
case "chance":
newChance = checkChance(values[i], name);
break;
case "managed_by_pattern":
if (values[i] != null && (Boolean)values[i])
throw new InvalidRequestException("Can only unset '" + name + '\'');
unsetManagedByOwner = true;
break;
case "bucket_size":
newBucketSize = checkNonNegative(values[i], name, 0);
break;
case "bucket_sub_size":
newBucketSubSize = checkNonNegative(values[i], name, 0);
break;
case "bucket_seen":
newBucketSeen = checkNonNegative(values[i], name, 0);
break;
case "trace_events":
newTrace = tryParseCoordinationKinds(values[i]);
break;
}
}
@Override
protected void applyColumnUpdate(ColumnValues partitionKey, ColumnValues clusteringColumns, Optional<ColumnValue> columnValue)
{
TxnId txnId = TxnId.parse(partitionKey.value(0));
TraceEventType eventType = parseEventType(clusteringColumns.value(0));
if (columnValue.isEmpty()) tracing().erasePermits(txnId, eventType);
else tracing().setPermits(txnId, eventType, columnValue.get().value());
if (newBucketSize == 0)
{
if (newBucketMode != null || newBucketSeen > 0 || newBucketSubSize > 0 || !Float.isNaN(newChance) || (newTrace != null && !newTrace.isEmpty()))
throw new InvalidRequestException("Setting bucket size to zero clears config; cannot set other fields.");
tracing().stopTracing(txnId);
}
else
{
if (newBucketSubSize == 0)
throw new InvalidRequestException("Cannot set bucket_sub_size to zero.");
tracing().set(txnId, newTrace, newBucketMode, newBucketSize, newBucketSubSize, newBucketSeen, newChance, unsetManagedByOwner);
}
}
@Override
public void truncate()
{
tracing().eraseAllEvents();
tracing().eraseAllBuckets();
}
}
@ -989,20 +1081,15 @@ public class AccordDebugKeyspace extends VirtualKeyspace
"Accord Transaction Traces",
"CREATE TABLE %s (\n" +
" txn_id 'TxnIdUtf8Type',\n" +
" event_type text,\n" +
" id_micros bigint,\n" +
" event text,\n" +
" at_micros bigint,\n" +
" command_store_id int,\n" +
" message text,\n" +
" PRIMARY KEY (txn_id, event_type, id_micros, at_micros)" +
" PRIMARY KEY (txn_id, id_micros, event, at_micros)" +
')', TxnIdUtf8Type.instance), FAIL, UNSORTED, UNSORTED);
}
private AccordTracing tracing()
{
return ((AccordAgent)AccordService.instance().agent()).tracing();
}
@Override
protected void applyPartitionDeletion(Object[] partitionKeys)
{
@ -1014,21 +1101,22 @@ public class AccordDebugKeyspace extends VirtualKeyspace
protected void applyRangeTombstone(Object[] partitionKeys, Object[] starts, boolean startInclusive, Object[] ends, boolean endInclusive)
{
TxnId txnId = TxnId.parse((String) partitionKeys[0]);
if (!startInclusive) throw invalidRequest("May restrict deletion by at most one event_type");
if (starts.length != 1) throw invalidRequest("Deletion restricted by lower bound on id_micros or at_micros is unsupported");
if (ends.length == 0 || (ends.length == 1 && !endInclusive)) throw invalidRequest("Range deletion must specify one event_type");
if (!ends[0].equals(starts[0])) throw invalidRequest("May restrict deletion by at most one event_type");
if (ends.length > 2) throw invalidRequest("Deletion restricted by upper bound on at_micros is unsupported");
TraceEventType eventType = parseEventType((String) starts[0]);
if (starts.length > 1 || ends.length > 1) throw invalidRequest("May only delete on txn_id and id_micros");
long minId = Long.MIN_VALUE, maxId = Long.MAX_VALUE;
if (starts.length == 1)
{
minId = ((Long)starts[0]);
if (!startInclusive && minId < Long.MAX_VALUE)
++minId;
}
if (ends.length == 1)
{
tracing().eraseEvents(txnId, eventType);
}
else
{
long before = (Long)ends[1];
tracing().eraseEventsBefore(txnId, eventType, before);
maxId = ((Long)ends[0]);
if (!endInclusive && maxId > Long.MIN_VALUE)
--maxId;
}
tracing().eraseEventsBetween(txnId, minId, maxId);
}
@Override
@ -1040,11 +1128,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace
@Override
public void collect(PartitionsCollector collector)
{
tracing().forEach(id -> true, (txnId, eventType, permits, events) -> {
tracing().forEach(id -> true, (txnId, events) -> {
events.forEach(e -> {
if (e.messages().isEmpty())
{
collector.row(txnId.toString(), eventType.name(), e.idMicros, 0L)
collector.row(txnId.toString(), e.idMicros, e.kind.name(), 0L)
.eagerCollect(columns -> {
columns.add("message", "<Initialised but no events (yet) recorded>");
});
@ -1052,7 +1140,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
else
{
e.messages().forEach(m -> {
collector.row(txnId.toString(), eventType.name(), e.idMicros, NANOSECONDS.toMicros(m.atNanos - e.atNanos))
collector.row(txnId.toString(), e.idMicros, e.kind.name(), NANOSECONDS.toMicros(m.atNanos - e.atNanos))
.eagerCollect(columns -> {
columns.add("command_store_id", m.commandStoreId)
.add("message", m.message);
@ -1064,6 +1152,226 @@ public class AccordDebugKeyspace extends VirtualKeyspace
}
}
/**
* Usage:
* collect N events (may be more than N messages)
* UPDATE system_accord_debug.txn_trace SET permits = N WHERE txn_id = ? AND event_type = ?
* SELECT * FROM system_accord_debug.txn_traces WHERE txn_id = ? AND event_type = ?
*/
public static final class TxnPatternTraceTable extends AbstractMutableLazyVirtualTable
{
private TxnPatternTraceTable()
{
super(parse(VIRTUAL_ACCORD_DEBUG, TXN_PATTERN_TRACE,
"Accord Transaction Pattern Trace Configuration",
"CREATE TABLE %s (\n" +
" id int,\n" +
" bucket_mode text,\n" +
" bucket_seen int,\n" +
" bucket_size int,\n" +
" chance float,\n" +
" current_size int,\n" +
" if_intersects text,\n" +
" if_kind text,\n" +
" on_failure text,\n" +
" on_new text,\n" +
" trace_bucket_mode text,\n" +
" trace_bucket_size int,\n" +
" trace_bucket_sub_size int,\n" +
" trace_events text,\n" +
" PRIMARY KEY (id)" +
')', Int32Type.instance), FAIL, SORTED);
}
@Override
protected void collect(PartitionsCollector collector)
{
AccordTracing tracing = tracing();
tracing.forEachPattern((state) -> {
collector.row(state.id())
.eagerCollect(columns -> {
columns.add("bucket_mode", state.mode().name())
.add("bucket_size", state.bucketSize())
.add("bucket_seen", state.bucketSeen())
.add("chance", state.pattern().chance)
.add("current_size", state.currentSize())
.add("if_intersects", state.pattern().intersects, TxnPatternTraceTable::toString)
.add("if_kind", state.pattern().kinds, TO_STRING)
.add("on_failure", state.pattern().traceFailures, TO_STRING)
.add("on_new", state.pattern().traceNew, TO_STRING)
.add("trace_bucket_mode", state.traceWithMode().name())
.add("trace_bucket_size", state.traceBucketSize())
.add("trace_bucket_sub_size", state.traceBucketSubSize())
.add("trace_events", state.traceEvents(), TO_STRING)
;
});
});
}
@Override
protected void applyPartitionDeletion(Object[] partitionKeys)
{
int id = (Integer) partitionKeys[0];
tracing().erasePattern(id);
}
@Override
protected void applyRowUpdate(Object[] partitionKeys, @Nullable Object[] clusteringKeys, ColumnMetadata[] columns, Object[] values)
{
int id = (Integer)partitionKeys[0];
Function<TracePattern, TracePattern> pattern = Function.identity();
CoordinationKinds newTraceEvents = null;
BucketMode newBucketMode = null, newTraceBucketMode = null;
int newBucketSize = -1, newTraceBucketSize = -1, newTraceBucketSubSize = -1;
int newBucketSeen = -1;
for (int i = 0 ; i < columns.length ; ++i)
{
String name = columns[i].name.toString();
switch (name)
{
default: throw new InvalidRequestException("Cannot update '" + name + '\'');
case "bucket_mode":
newBucketMode = checkBucketMode(values[i]);
break;
case "bucket_seen":
newBucketSeen = checkNonNegative(values[i], name, 0);
break;
case "bucket_size":
newBucketSize = checkNonNegative(values[i], name, 0);
break;
case "chance":
float newChance = checkChance(values[i], name);
pattern = pattern.andThen(p -> p.withChance(newChance));
break;
case "if_intersects":
Participants<?> intersects = parseParticipants(values[i]);
pattern = pattern.andThen(p -> p.withIntersects(intersects));
break;
case "if_kind":
TxnKindsAndDomains kinds = tryParseTxnKinds(values[i]);
pattern = pattern.andThen(p -> p.withKinds(kinds));
break;
case "on_failure":
CoordinationKinds traceFailures = tryParseCoordinationKinds(values[i]);
pattern = pattern.andThen(p -> p.withTraceFailures(traceFailures));
break;
case "on_new":
CoordinationKinds traceNew = tryParseCoordinationKinds(values[i]);
pattern = pattern.andThen(p -> p.withTraceNew(traceNew));
break;
case "trace_bucket_mode":
newTraceBucketMode = checkBucketMode(values[i]);
break;
case "trace_bucket_size":
newTraceBucketSize = checkNonNegative(values[i], name, 0);
break;
case "trace_bucket_sub_size":
newTraceBucketSubSize = checkNonNegative(values[i], name, 0);
break;
case "trace_events":
newTraceEvents = tryParseCoordinationKinds(values[i]);
break;
}
}
tracing().setPattern(id, pattern, newBucketMode, newBucketSeen, newBucketSize, newTraceBucketMode, newTraceBucketSize, newTraceBucketSubSize, newTraceEvents);
}
private static String toString(Participants<?> participants)
{
StringBuilder out = new StringBuilder();
for (Routable r : participants)
{
if (out.length() != 0)
out.append('|');
out.append(r);
}
return out.toString();
}
private static Participants<?> parseParticipants(Object input)
{
if (input == null)
return null;
String[] vs = ((String)input).split("\\|");
if (vs.length == 0)
return RoutingKeys.EMPTY;
if (!vs[0].endsWith("]"))
{
RoutingKey[] keys = new RoutingKey[vs.length];
for (int i = 0 ; i < keys.length ; ++i)
{
try { keys[i] = TokenKey.parse(vs[i], DatabaseDescriptor.getPartitioner()); }
catch (Throwable t) { throw new InvalidRequestException("Could not parse TokenKey " + vs[0]); }
}
return RoutingKeys.of(keys);
}
else
{
TokenRange[] ranges = new TokenRange[vs.length];
for (int i = 0 ; i < ranges.length ; ++i)
{
try { ranges[i] = TokenRange.parse(vs[0], DatabaseDescriptor.getPartitioner()); }
catch (Throwable t) { throw new InvalidRequestException("Could not parse TokenKey " + vs[0]); }
}
return Ranges.of(ranges);
}
}
@Override
public void truncate()
{
tracing().eraseAllPatterns();
}
}
public static final class TxnPatternTracesTable extends AbstractMutableLazyVirtualTable
{
private TxnPatternTracesTable()
{
super(parse(VIRTUAL_ACCORD_DEBUG, TXN_PATTERN_TRACES,
"Accord Transaction Pattern Traces",
"CREATE TABLE %s (\n" +
" id int,\n" +
" txn_id 'TxnIdUtf8Type',\n" +
" PRIMARY KEY (id, txn_id)" +
')', Int32Type.instance), FAIL, SORTED, SORTED);
}
@Override
protected void applyPartitionDeletion(Object[] partitionKeys)
{
int id = (Integer) partitionKeys[0];
tracing().erasePatternTraces(id);
}
@Override
public void truncate()
{
tracing().eraseAllPatternTraces();
}
@Override
public void collect(PartitionsCollector collector)
{
tracing().forEachPattern(state -> {
if (state.currentSize() == 0)
{
collector.row(state.id(), "")
.eagerCollect(column -> {});
}
else
{
for (int i = 0, size = state.currentSize(); i < size ; ++i)
collector.row(state.id(), state.get(i).toString())
.eagerCollect(columns -> {});
}
});
}
}
// TODO (desired): don't report null as "null"
abstract static class AbstractJournalTable extends AbstractLazyVirtualTable
{
@ -1291,7 +1599,24 @@ public class AccordDebugKeyspace extends VirtualKeyspace
public static final class TxnOpsTable extends AbstractMutableLazyVirtualTable
{
// TODO (expected): test each of these operations
enum Op { ERASE_VESTIGIAL, INVALIDATE, TRY_EXECUTE, FORCE_APPLY, FORCE_UPDATE, RECOVER, FETCH, RESET_PROGRESS_LOG }
enum Op
{
LOCALLY_ERASE_VESTIGIAL("USE WITH CAUTION: Move the command to the vestigial status, erasing its contents. This has distributed state machine implications."),
LOCALLY_INVALIDATE("USE WITH CAUTION: Move the command to the invalidated status, erasing its contents. This has distributed state machine implications."),
TRY_EXECUTE("Try to execute a stuck transaction. This is safe, and will no-op if not able to."),
FORCE_APPLY("USE WITH CAUTION: Apply the command if we have the relevant information locally."),
FORCE_UPDATE("Try to reset in-memory book-keeping related to a command."),
RECOVER("Initiate recovery for a command."),
FETCH("Initiate a fetch request for a command."),
REQUEUE_PROGRESS_LOG("Ask the progress log to queue both home and waiting states.");
final String description;
Op(String description)
{
this.description = description;
}
}
private TxnOpsTable()
{
super(parse(VIRTUAL_ACCORD_DEBUG, TXN_OPS,
@ -1315,14 +1640,16 @@ public class AccordDebugKeyspace extends VirtualKeyspace
{
TxnId txnId = TxnId.parse((String) partitionKeys[0]);
int commandStoreId = (Integer) clusteringKeys[0];
Op op = Op.valueOf((String)values[0]);
Op op = tryParse(values[0], true, Op.class, Op::valueOf);
switch (op)
{
default: throw new UnhandledEnum(op);
case ERASE_VESTIGIAL:
case LOCALLY_ERASE_VESTIGIAL:
cleanup(txnId, commandStoreId, Cleanup.VESTIGIAL);
break;
case INVALIDATE:
case LOCALLY_INVALIDATE:
cleanup(txnId, commandStoreId, Cleanup.INVALIDATE);
break;
case TRY_EXECUTE:
@ -1360,7 +1687,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
recover(txnId, route, result);
});
break;
case RESET_PROGRESS_LOG:
case REQUEUE_PROGRESS_LOG:
run(txnId, commandStoreId, safeStore -> {
((DefaultProgressLog)safeStore.progressLog()).requeue(safeStore, TxnStateKind.Waiting, txnId);
((DefaultProgressLog)safeStore.progressLog()).requeue(safeStore, TxnStateKind.Home, txnId);
@ -1381,7 +1708,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
BiConsumer<Route<?>, AsyncResult.Settable<Void>> consumer = apply.apply(command);
if (command.route() == null)
{
FetchRoute.fetchRoute(node, txnId, command.maxContactable(), LatentStoreSelector.standard(), (success, fail) -> {
FetchRoute.fetchRoute(node, txnId, command.maxParticipants(), LatentStoreSelector.standard(), (success, fail) -> {
if (fail != null) result.setFailure(fail);
else consumer.accept(success, result);
});
@ -1411,7 +1738,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
PrepareRecovery.recover(node, node.someSequentialExecutor(), txnId, NotKnownToBeInvalid, (FullRoute<?>) route, null, LatentStoreSelector.standard(), (success, fail) -> {
if (fail != null) result.setFailure(fail);
else result.setSuccess(null);
}, node.agent().trace(txnId, RECOVER));
});
}
else
{
@ -1497,21 +1824,6 @@ public class AccordDebugKeyspace extends VirtualKeyspace
}
}
private static TableId tableId(int commandStoreId, CommandStores commandStores)
{
AccordCommandStore commandStore = (AccordCommandStore) commandStores.forId(commandStoreId);
if (commandStore == null)
return null;
return commandStore.tableId();
}
private static TableMetadata tableMetadata(TableId tableId)
{
if (tableId == null)
return null;
return Schema.instance.getTableMetadata(tableId);
}
private static String printToken(RoutingKey routingKey)
{
TokenKey key = (TokenKey) routingKey;
@ -1539,10 +1851,9 @@ public class AccordDebugKeyspace extends VirtualKeyspace
return av + " (" + bv + ')';
}
private static TraceEventType parseEventType(String input)
private static CoordinationKind parseEventType(String input)
{
try { return TraceEventType.valueOf(LocalizeString.toUpperCaseLocalized(input, Locale.ENGLISH)); }
catch (Throwable t) { throw invalidRequest("event_type must be one of %s; received %s", TraceEventType.values(), input); }
return tryParse(input, false, CoordinationKind.class, CoordinationKind::valueOf);
}
private static String toStringOrNull(Object o)
@ -1559,4 +1870,80 @@ public class AccordDebugKeyspace extends VirtualKeyspace
return "<error: " + t.getLocalizedMessage() + '>';
}
}
private static BucketMode checkBucketMode(Object value)
{
try { return AccordTracing.BucketMode.valueOf(LocalizeString.toUpperCaseLocalized((String)value, Locale.ENGLISH)); }
catch (IllegalArgumentException | NullPointerException e)
{
throw new InvalidRequestException("Unknown bucket_mode '" + value + '\'');
}
}
private static int checkNonNegative(Object value, String field, int ifNull)
{
if (value == null)
return ifNull;
int v = (Integer)value;
if (v < 0)
throw new InvalidRequestException("Cannot set '" + field + "' to negative value");
return v;
}
private static float checkChance(Object value, String field)
{
if (value == null)
return 1.0f;
float v = (Float)value;
if (v <= 0 || v > 1.0f)
throw new InvalidRequestException("Cannot set '" + field + "' to value outside the range (0..1]");
return v;
}
private static <T extends Enum<T>> Set<String> toStrings(TinyEnumSet<T> set, IntFunction<T> lookup)
{
ImmutableSet.Builder<String> builder = ImmutableSet.builder();
for (T t : set.iterable(lookup))
builder.add(t.name());
return builder.build();
}
private static AccordTracing tracing()
{
return ((AccordAgent)AccordService.instance().agent()).tracing();
}
private static <E extends Enum<E>> E tryParse(Object input, boolean toUpperCase, Class<E> clazz, Function<String, E> valueOf)
{
try
{
String str = (String) input;
if (toUpperCase)
str = LocalizeString.toUpperCaseLocalized(str, Locale.ENGLISH);
return valueOf.apply(str);
}
catch (IllegalArgumentException | NullPointerException e)
{
throw new InvalidRequestException("Unknown " + clazz.getName() + ": '" + input + '\'');
}
}
private static CoordinationKinds tryParseCoordinationKinds(Object input)
{
if (input == null)
return null;
return CoordinationKinds.parse((String) input);
}
private static TxnKindsAndDomains tryParseTxnKinds(Object input)
{
if (input == null)
return null;
return TxnKindsAndDomains.parse((String) input);
}
}

2
src/java/org/apache/cassandra/metrics/AccordCoordinatorMetrics.java
View File

@ -198,8 +198,6 @@ public class AccordCoordinatorMetrics
public static class Listener implements CoordinatorEventListener
{
public static final Listener instance = new Listener();
private AccordCoordinatorMetrics forTransaction(TxnId txnId)
{
if (txnId != null)

2
src/java/org/apache/cassandra/metrics/AccordReplicaMetrics.java
View File

@ -112,8 +112,6 @@ public class AccordReplicaMetrics
public static class Listener implements ReplicaEventListener
{
public static final Listener instance = new Listener();
private AccordReplicaMetrics forTransaction(TxnId txnId)
{
if (txnId != null)

4
src/java/org/apache/cassandra/metrics/AccordSystemMetrics.java
View File

@ -132,6 +132,10 @@ public class AccordSystemMetrics
}
}
public static void touch()
{
}
private AccordSystemMetrics()
{
Invariants.expect(AccordService.isSetup());

2
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -41,6 +41,7 @@ import com.google.common.primitives.Ints;
import accord.api.ConfigurationService.EpochReady;
import accord.primitives.Txn;
import org.apache.cassandra.metrics.AccordReplicaMetrics;
import org.apache.cassandra.metrics.AccordSystemMetrics;
import org.apache.cassandra.service.accord.api.AccordViolationHandler;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.concurrent.AsyncFuture;
@ -307,6 +308,7 @@ public class AccordService implements IAccordService, Shutdownable
instance = as;
AccordReplicaMetrics.touch();
AccordSystemMetrics.touch();
AccordViolationHandler.setup();
WatermarkCollector.fetchAndReportWatermarksAsync(as.configService);

936
src/java/org/apache/cassandra/service/accord/AccordTracing.java
View File

File diff suppressed because it is too large Load Diff

17
src/java/org/apache/cassandra/service/accord/TokenRange.java
View File

@ -173,4 +173,21 @@ public class TokenRange extends Range.EndInclusive
+ TokenKey.noTableSerializer.serializedSize(t.end());
}
};
public static TokenRange parse(String str, IPartitioner partitioner)
{
TableId tableId;
{
int split = str.indexOf(':', str.startsWith("tid:") ? 4 : 0);
tableId = TableId.fromString(str.substring(0, split));
str = str.substring(split + 2, str.length() - 1);
}
String[] bounds = str.split(",");
if (bounds.length != 2)
throw new IllegalArgumentException("Invalid TokenRange: " + str);
return new TokenRange(TokenKey.parse(tableId, bounds[0], partitioner), TokenKey.parse(tableId, bounds[1], partitioner));
}
}

13
src/java/org/apache/cassandra/service/accord/api/AccordAgent.java
View File

@ -35,7 +35,7 @@ import accord.api.ReplicaEventListener;
import accord.api.ProgressLog.BlockedUntil;
import accord.api.RoutingKey;
import accord.api.Tracing;
import accord.api.TraceEventType;
import accord.coordinate.Coordination;
import accord.local.Command;
import accord.local.Node;
import accord.local.SafeCommand;
@ -43,6 +43,7 @@ import accord.local.SafeCommandStore;
import accord.local.TimeService;
import accord.messages.ReplyContext;
import accord.primitives.Keys;
import accord.primitives.Participants;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Status;
@ -62,7 +63,6 @@ import accord.utils.async.AsyncChains;
import accord.utils.async.Cancellable;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.exceptions.RequestTimeoutException;
import org.apache.cassandra.metrics.AccordCoordinatorMetrics;
import org.apache.cassandra.metrics.AccordReplicaMetrics;
import org.apache.cassandra.net.ResponseContext;
import org.apache.cassandra.service.accord.AccordService;
@ -98,6 +98,7 @@ public class AccordAgent implements Agent, OwnershipEventListener
{
private static final Logger logger = LoggerFactory.getLogger(AccordAgent.class);
private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, MINUTES);
private static final ReplicaEventListener replicaEventListener = new AccordReplicaMetrics.Listener();
private static BiConsumer<TxnId, Throwable> onFailedBarrier;
public static void setOnFailedBarrier(BiConsumer<TxnId, Throwable> newOnFailedBarrier) { onFailedBarrier = newOnFailedBarrier; }
@ -121,9 +122,9 @@ public class AccordAgent implements Agent, OwnershipEventListener
}
@Override
public @Nullable Tracing trace(TxnId txnId, TraceEventType eventType)
public @Nullable Tracing trace(TxnId txnId, Participants<?> participants, Coordination.CoordinationKind eventType)
{
return tracing.trace(txnId, eventType);
return tracing.trace(txnId, participants, eventType);
}
@Override
@ -225,13 +226,13 @@ public class AccordAgent implements Agent, OwnershipEventListener
@Override
public CoordinatorEventListener coordinatorEvents()
{
return AccordCoordinatorMetrics.Listener.instance;
return tracing;
}
@Override
public ReplicaEventListener replicaEvents()
{
return AccordReplicaMetrics.Listener.instance;
return replicaEventListener;
}
private static final long ONE_SECOND = SECONDS.toMicros(1L);

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

@ -149,12 +149,15 @@ public final class TokenKey extends AccordRoutableKey implements RoutingKey, Ran
public static TokenKey parse(String str, IPartitioner partitioner)
{
TableId tableId;
{
int split = str.indexOf(':', str.startsWith("tid:") ? 4 : 0);
tableId = TableId.fromString(str.substring(0, split));
str = str.substring(split + 1);
}
int split = str.indexOf(':', str.startsWith("tid:") ? 4 : 0);
TableId tableId = TableId.fromString(str.substring(0, split));
str = str.substring(split + 1);
return parse(tableId, str, partitioner);
}
public static TokenKey parse(TableId tableId, String str, IPartitioner partitioner)
{
if (str.endsWith("Inf"))
{
return new TokenKey(tableId, str.charAt(0) == '-' ? MIN_TABLE_SENTINEL : MAX_TABLE_SENTINEL, partitioner.getMinimumToken());

264
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.db.virtual;
import java.util.Collections;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@ -127,16 +128,16 @@ public class AccordDebugKeyspaceTest extends CQLTester
String.format("SELECT txn_id, save_status FROM %s.%s WHERE node_id = ? AND txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.JOURNAL);
private static final String SET_TRACE =
String.format("UPDATE %s.%s SET permits = ? WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
String.format("UPDATE %s.%s SET bucket_size = ?, trace_events = ? WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
private static final String SET_TRACE_REMOTE =
String.format("UPDATE %s.%s SET permits = ? WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
String.format("UPDATE %s.%s SET bucket_size = ?, trace_events = ? WHERE node_id = ? AND txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String QUERY_TRACE =
String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
String.format("SELECT txn_id, bucket_size, trace_events FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
private static final String QUERY_TRACE_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
String.format("SELECT node_id, txn_id, bucket_size, trace_events FROM %s.%s WHERE node_id = ? AND txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String UNSET_TRACE1 =
String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
@ -144,33 +145,27 @@ public class AccordDebugKeyspaceTest extends CQLTester
private static final String UNSET_TRACE1_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String UNSET_TRACE2 =
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
private static final String UNSET_TRACE2_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String QUERY_ALL_TRACES =
String.format("SELECT * FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String QUERY_TRACES =
String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String QUERY_TRACES_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES1 =
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ? AND id_micros < ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND id_micros < ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES1_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ? AND id_micros < ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES2 =
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES2_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ? AND id_micros < ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES3 =
String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String TRUNCATE_TRACES =
String.format("TRUNCATE %s.%s", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES3_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
@ -192,6 +187,15 @@ public class AccordDebugKeyspaceTest extends CQLTester
private static final String QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND shard_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.REDUNDANT_BEFORE);
private static final String SET_PATTERN_TRACE =
String.format("UPDATE %s.%s SET bucket_mode = ?, bucket_seen = ?, bucket_size = ?, chance = ?, if_intersects = ?, if_kind = ?, on_failure = ?, on_new = ?, trace_bucket_mode = ?, trace_bucket_size = ?, trace_bucket_sub_size = ?, trace_events = ? WHERE id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_PATTERN_TRACE);
private static final String UNSET_PATTERN_TRACE =
String.format("DELETE FROM %s.%s WHERE id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_PATTERN_TRACE);
private static final String QUERY_PATTERN_TRACE =
String.format("SELECT * FROM %s.%s WHERE id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_PATTERN_TRACE);
@BeforeClass
public static void setUpClass()
{
@ -239,40 +243,31 @@ public class AccordDebugKeyspaceTest extends CQLTester
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
filter.appliesTo(id);
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
execute(SET_TRACE, 0, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
execute(SET_TRACE, 1, "{WaitProgress}", id.toString());
assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "{WaitProgress}"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()), row(nodeId, id.toString(), 1, "{WaitProgress}"));
execute(SET_TRACE, 0, "{}", id.toString());
assertRows(execute(QUERY_TRACE, id.toString()));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()));
execute(SET_TRACE, 1, "{WaitProgress}", id.toString());
assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "{WaitProgress}"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()), row(nodeId, id.toString(), 1, "{WaitProgress}"));
execute(UNSET_TRACE1, id.toString());
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
execute(UNSET_TRACE2, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
filter.appliesTo(id);
assertRows(execute(QUERY_TRACE, id.toString()));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()));
execute(SET_TRACE, 1, "{WaitProgress}", id.toString());
assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "{WaitProgress}"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()), row(nodeId, id.toString(), 1, "{WaitProgress}"));
accord.node().coordinate(id, txn).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
filter.apply.awaitThrowUncheckedOnInterrupt();
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WAIT_PROGRESS").size()).isGreaterThan(0));
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS").size()).isGreaterThan(0));
execute(ERASE_TRACES1, id.toString(), "FETCH", Long.MAX_VALUE);
execute(ERASE_TRACES2, id.toString(), "FETCH");
execute(ERASE_TRACES1, id.toString(), "WAIT_PROGRESS", Long.MAX_VALUE);
Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WAIT_PROGRESS").size()).isEqualTo(0);
Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS").size()).isEqualTo(0);
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WaitProgress").size()).isGreaterThan(0));
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WaitProgress").size()).isGreaterThan(0));
execute(ERASE_TRACES1, id.toString(), Long.MAX_VALUE);
execute(ERASE_TRACES1, id.toString(), Long.MAX_VALUE);
Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WaitProgress").size()).isEqualTo(0);
Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WaitProgress").size()).isEqualTo(0);
// just check other variants don't fail
execute(ERASE_TRACES2, id.toString(), "WAIT_PROGRESS");
execute(ERASE_TRACES3, id.toString());
}
@ -280,6 +275,137 @@ public class AccordDebugKeyspaceTest extends CQLTester
{
MessagingService.instance().outboundSink.remove(filter);
}
filter = new AccordMsgFilter();
MessagingService.instance().outboundSink.add(filter);
try
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 1, 1, 1);
filter.appliesTo(id);
execute(SET_TRACE_REMOTE, 1, "{WaitProgress}", nodeId, id.toString());
assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "{WaitProgress}"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()), row(nodeId, id.toString(), 1, "{WaitProgress}"));
execute(SET_TRACE_REMOTE, 0, "{}", nodeId, id.toString());
assertRows(execute(QUERY_TRACE, id.toString()));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()));
execute(SET_TRACE_REMOTE, 1, "{WaitProgress}", nodeId, id.toString());
assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "{WaitProgress}"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()), row(nodeId, id.toString(), 1, "{WaitProgress}"));
execute(UNSET_TRACE1_REMOTE, nodeId, id.toString());
assertRows(execute(QUERY_TRACE, id.toString()));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()));
execute(SET_TRACE_REMOTE, 1, "{WaitProgress}", nodeId, id.toString());
assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "{WaitProgress}"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString()), row(nodeId, id.toString(), 1, "{WaitProgress}"));
accord.node().coordinate(id, txn).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
filter.apply.awaitThrowUncheckedOnInterrupt();
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WaitProgress").size()).isGreaterThan(0));
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WaitProgress").size()).isGreaterThan(0));
execute(ERASE_TRACES1_REMOTE, nodeId, id.toString(), Long.MAX_VALUE);
execute(ERASE_TRACES1_REMOTE, nodeId, id.toString(), Long.MAX_VALUE);
Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WaitProgress").size()).isEqualTo(0);
Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WaitProgress").size()).isEqualTo(0);
// just check other variants don't fail
execute(ERASE_TRACES3_REMOTE, nodeId, id.toString());
}
finally
{
MessagingService.instance().outboundSink.remove(filter);
}
}
@Test
public void patternTracing()
{
// simple test to confirm basic tracing functionality works, doesn't validate specific behaviours only requesting/querying/erasing
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
DatabaseDescriptor.getAccord().fetch_txn = "1s";
execute(SET_PATTERN_TRACE, "leaky", 0, 5, 1.0f, "tid:1:1|tid:1:2", "-{*X}", "-{WaitProgress}", "{}", "ring", 5, 1, "*", 1);
assertRows(execute(QUERY_PATTERN_TRACE, 1), row(1, "LEAKY", 0, 5, 1.0f, 0, "tid:1:1|tid:1:2", "-{KX,RX}", "-{WaitProgress}", "{}", "RING", 5, 1, "*"));
execute(UNSET_PATTERN_TRACE, 1);
assertRows(execute(QUERY_PATTERN_TRACE, 1));
RoutingKey matchKey;
{
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
matchKey = (RoutingKey) txn.keys().toParticipants().get(0);
}
int count = 5;
{
List<TxnId> txnIds = new ArrayList<>();
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, matchKey.toString(), "*", "{}", "*", "leaky", 1, 1, "*", 1);
for (int i = 0 ; i < count + 1 ; ++i)
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, i, 0);
getBlocking(accord.node().coordinate(id, txn));
if (i < count) assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "*"));
else assertRows(execute(QUERY_TRACE, id.toString()));
txnIds.add(id);
}
execute(UNSET_PATTERN_TRACE, 1);
for (int i = 0 ; i < count ; ++i)
assertRows(execute(QUERY_TRACE, txnIds.get(i).toString()));
}
{
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, matchKey.asRange().toString(), "{KE}", "{}", "{PreAccept}", "leaky", 1, 1, "*", 1);
for (int i = 0 ; i < count ; ++i)
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, i, 0);
getBlocking(accord.node().coordinate(id, txn));
assertRows(execute(QUERY_TRACE, id.toString()));
}
List<TxnId> txnIds = new ArrayList<>();
for (int i = 0 ; i < count + 1 ; ++i)
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.EphemeralRead, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("SELECT * FROM %s.%s WHERE k = ? AND c = ?", KEYSPACE, tableName)), 0, i);
getBlocking(accord.node().coordinate(id, txn));
if (i < count) assertRows(execute(QUERY_TRACE, id.toString()), row(id.toString(), 1, "*"));
else assertRows(execute(QUERY_TRACE, id.toString()));
txnIds.add(id);
}
execute(UNSET_PATTERN_TRACE, 1);
for (int i = 0 ; i < count ; ++i)
assertRows(execute(QUERY_TRACE, txnIds.get(i).toString()));
}
{
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 1, 1, 1);
execute(SET_PATTERN_TRACE, "leaky", 0, count, 1.0f, "" + txn.keys().get(0).toUnseekable(), "{KW}", "*", "{}", "leaky", 1, 1, "{}", 1);
AccordMsgFilter filter = new AccordMsgFilter();
filter.dropVerbs = EnumSet.allOf(Verb.class);
filter.appliesTo(id);
MessagingService.instance().outboundSink.add(filter);
try
{
boolean failed = false;
try { getBlocking(accord.node().coordinate(id, txn)); }
catch (Throwable ignore) { failed = true; }
Assertions.assertThat(failed).isTrue();
}
finally
{
MessagingService.instance().outboundSink.remove(filter);
}
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_ALL_TRACES, id.toString()).size()).isGreaterThan(0), 60);
execute(UNSET_PATTERN_TRACE, 1);
}
}
@Test
@ -332,19 +458,30 @@ public class AccordDebugKeyspaceTest extends CQLTester
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
AccordMsgFilter filter = new AccordMsgFilter();
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
String keyStr = txn.keys().get(0).toUnseekable().toString();
getBlocking(accord.node().coordinate(id, txn));
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), anyInt(), 0, "", "", any(), anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name()))));
assertRows(execute(QUERY_TXN, id.toString()), row(id.toString(), "Applied"));
assertRows(execute(QUERY_TXN_REMOTE, nodeId, id.toString()), row(id.toString(), "Applied"));
assertRows(execute(QUERY_JOURNAL, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null));
assertRows(execute(QUERY_JOURNAL_REMOTE, nodeId, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null));
assertRows(execute(QUERY_COMMANDS_FOR_KEY, keyStr), row(id.toString(), "APPLIED_DURABLE"));
assertRows(execute(QUERY_COMMANDS_FOR_KEY_REMOTE, nodeId, keyStr), row(id.toString(), "APPLIED_DURABLE"));
filter.appliesTo(id);
filter.dropVerbs = Set.of();
MessagingService.instance().outboundSink.add(filter);
try
{
String keyStr = txn.keys().get(0).toUnseekable().toString();
getBlocking(accord.node().coordinate(id, txn));
filter.apply.awaitThrowUncheckedOnInterrupt();
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), anyInt(), 0, "", "", any(), anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name()))));
assertRows(execute(QUERY_TXN, id.toString()), row(id.toString(), anyOf("Applied", "Applying")));
assertRows(execute(QUERY_TXN_REMOTE, nodeId, id.toString()), row(id.toString(), anyOf("Applied", "Applying")));
assertRows(execute(QUERY_JOURNAL, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null));
assertRows(execute(QUERY_JOURNAL_REMOTE, nodeId, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null));
assertRows(execute(QUERY_COMMANDS_FOR_KEY, keyStr), row(id.toString(), "APPLIED_DURABLE"));
assertRows(execute(QUERY_COMMANDS_FOR_KEY_REMOTE, nodeId, keyStr), row(id.toString(), "APPLIED_DURABLE"));
}
finally
{
MessagingService.instance().outboundSink.remove(filter);
}
}
@Test
@ -503,13 +640,13 @@ public class AccordDebugKeyspaceTest extends CQLTester
@Test
public void patchJournalVestigialTest()
{
testPatchJournal("ERASE_VESTIGIAL", "Vestigial");
testPatchJournal("LOCALLY_ERASE_VESTIGIAL", "Vestigial");
}
@Test
public void patchJournalInvalidateTest()
{
testPatchJournal("INVALIDATE", "Invalidated");
testPatchJournal("LOCALLY_INVALIDATE", "Invalidated");
}
@Test
@ -520,9 +657,8 @@ public class AccordDebugKeyspaceTest extends CQLTester
testPatchJournal("ERASE", "Erased");
Assert.fail("Should have thrown");
}
catch (Throwable t)
catch (InvalidRequestException t)
{
Assert.assertTrue(t.getMessage().contains("No enum constant"));
}
}
@ -630,6 +766,4 @@ public class AccordDebugKeyspaceTest extends CQLTester
return !dropVerbs.contains(msg.verb());
}
}
}