diff --git a/modules/accord b/modules/accord index 68778350bb..71d235d56c 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 68778350bb45a1545cbe38af290d7778ffb79454 +Subproject commit 71d235d56cb315fa5ae01ec24d3d9f08dd08ac6a diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index abc68385ef..4888c9d916 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -18,34 +18,47 @@ package org.apache.cassandra.db.virtual; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Collection; import java.util.Date; import java.util.HashSet; +import java.util.Iterator; import java.util.LinkedHashMap; 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.function.Function; import java.util.stream.Collectors; +import com.google.common.collect.BoundType; +import com.google.common.collect.Range; import com.google.common.collect.Sets; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.api.RoutingKey; +import accord.api.TraceEventType; +import accord.impl.CommandChange; import accord.impl.progresslog.DefaultProgressLog; import accord.impl.progresslog.TxnStateKind; +import accord.local.Command; import accord.local.CommandStore; import accord.local.CommandStores; import accord.local.DurableBefore; import accord.local.MaxConflicts; import accord.local.RejectBefore; +import accord.local.StoreParticipants; import accord.local.durability.ShardDurability; +import accord.primitives.Participants; import accord.primitives.Status; import accord.primitives.TxnId; import accord.utils.Invariants; import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DataRange; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.marshal.AbstractType; @@ -63,13 +76,17 @@ import org.apache.cassandra.service.accord.AccordCache; import org.apache.cassandra.service.accord.AccordCommandStore; import org.apache.cassandra.service.accord.AccordCommandStores; import org.apache.cassandra.service.accord.AccordExecutor; +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.CommandStoreTxnBlockedGraph; +import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.service.accord.api.TokenKey; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.service.consensus.migration.TableMigrationState; import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.utils.LocalizeString; import static accord.local.RedundantStatus.Property.GC_BEFORE; import static accord.local.RedundantStatus.Property.LOCALLY_APPLIED; @@ -81,6 +98,8 @@ import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP; import static accord.local.RedundantStatus.Property.SHARD_APPLIED; import static java.lang.String.format; import static com.google.common.collect.ImmutableList.toImmutableList; +import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest; import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_ACCORD_DEBUG; import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime; @@ -95,6 +114,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace public static final String REDUNDANT_BEFORE = "redundant_before"; public static final String REJECT_BEFORE = "reject_before"; public static final String TXN_BLOCKED_BY = "txn_blocked_by"; + public static final String TXN = "txn"; + public static final String JOURNAL = "journal"; + public static final String TXN_TRACE = "txn_trace"; + public static final String TXN_TRACES = "txn_traces"; public static final AccordDebugKeyspace instance = new AccordDebugKeyspace(); @@ -109,11 +132,14 @@ public class AccordDebugKeyspace extends VirtualKeyspace new ProgressLogTable(), new RedundantBeforeTable(), new RejectBeforeTable(), - new TxnBlockedByTable() + new TxnBlockedByTable(), + new TxnTable(), + new JournalTable(), + new TxnTraceTable(), + new TxnTracesTable() )); } - // TODO (consider): use a different type for the three timestamps in micros public static final class DurabilityServiceTable extends AbstractVirtualTable { private DurabilityServiceTable() @@ -126,10 +152,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace " token_sort blob,\n" + " token_start text,\n" + " token_end text,\n" + - " last_started_at_micros bigint,\n" + - " cycle_started_at_micros bigint,\n" + + " last_started_at bigint,\n" + + " cycle_started_at bigint,\n" + " retries int,\n" + " min text,\n" + + " requested_by text,\n" + " active text,\n" + " waiting text,\n" + " node_offset int,\n" + @@ -160,6 +187,9 @@ public class AccordDebugKeyspace extends VirtualKeyspace .column("token_end", printToken(view.shard().range.end())) .column("last_started_at", approxTime.translate().toMillisSinceEpoch(view.lastStartedAtMicros() * 1000)) .column("cycle_started_at", approxTime.translate().toMillisSinceEpoch(view.cycleStartedAtMicros() * 1000)) + .column("retries", view.retries()) + .column("min", Objects.toString(view.min())) + .column("requested_by", Objects.toString(view.requestedBy())) .column("active", Objects.toString(view.active())) .column("waiting", Objects.toString(view.waiting())) .column("node_offset", view.nodeOffset()) @@ -170,6 +200,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace .column("endIndex", view.cycleLength()) .column("current_splits", view.currentSplits()) .column("stopping", view.stopping()) + .column("stopped", view.stopped()) ; } return ds; @@ -476,11 +507,12 @@ public class AccordDebugKeyspace extends VirtualKeyspace " token_sort blob,\n" + " token_start text,\n" + " token_end text,\n" + - " command_store_id bigint,\n" + + " command_store_id int,\n" + " start_epoch bigint,\n" + " end_epoch bigint,\n" + " gc_before text,\n" + - " shard_only_applied text,\n" + + " shard_applied text,\n" + + " majority_applied text,\n" + " locally_applied text,\n" + " locally_synced text,\n" + " locally_redundant text,\n" + @@ -577,6 +609,358 @@ 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 TxnTraceTable extends AbstractMutableVirtualTable + { + private TxnTraceTable() + { + super(parse(VIRTUAL_ACCORD_DEBUG, TXN_TRACE, + "Accord Transaction Trace Configuration", + "CREATE TABLE %s (\n" + + " txn_id text,\n" + + " event_type text,\n" + + " permits int,\n" + + " PRIMARY KEY (txn_id, event_type)" + + ')', UTF8Type.instance)); + } + + @Override + public DataSet data() + { + 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); + }); + return dataSet; + } + + private AccordTracing tracing() + { + return ((AccordAgent)AccordService.instance().agent()).tracing(); + } + + @Override + protected void applyPartitionDeletion(ColumnValues partitionKey) + { + TxnId txnId = TxnId.parse(partitionKey.value(0)); + tracing().erasePermits(txnId); + } + + @Override + protected void applyRowDeletion(ColumnValues partitionKey, ColumnValues clusteringColumns) + { + TxnId txnId = TxnId.parse(partitionKey.value(0)); + tracing().erasePermits(txnId, parseEventType(clusteringColumns.value(0))); + } + + @Override + protected void applyColumnDeletion(ColumnValues partitionKey, ColumnValues clusteringColumns, String columnName) + { + TxnId txnId = TxnId.parse(partitionKey.value(0)); + TraceEventType eventType = parseEventType(clusteringColumns.value(0)); + tracing().erasePermits(txnId, eventType); + } + + @Override + protected void applyColumnUpdate(ColumnValues partitionKey, ColumnValues clusteringColumns, Optional 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()); + } + + @Override + public void truncate() + { + tracing().eraseAllEvents(); + } + } + + public static final class TxnTracesTable extends AbstractMutableVirtualTable + { + private TxnTracesTable() + { + super(parse(VIRTUAL_ACCORD_DEBUG, TXN_TRACES, + "Accord Transaction Traces", + "CREATE TABLE %s (\n" + + " txn_id text,\n" + + " event_type text,\n" + + " id_micros bigint,\n" + + " at_micros bigint,\n" + + " command_store_id int,\n" + + " message text,\n" + + " PRIMARY KEY (txn_id, event_type, id_micros, at_micros)" + + ')', UTF8Type.instance)); + } + + private AccordTracing tracing() + { + return ((AccordAgent)AccordService.instance().agent()).tracing(); + } + + @Override + protected void applyPartitionDeletion(ColumnValues partitionKey) + { + TxnId txnId = TxnId.parse(partitionKey.value(0)); + tracing().eraseEvents(txnId); + } + + @Override + protected void applyRangeTombstone(ColumnValues partitionKey, Range range) + { + TxnId txnId = TxnId.parse(partitionKey.value(0)); + if (!range.hasLowerBound() || range.lowerBoundType() != BoundType.CLOSED) throw invalidRequest("May restrict deletion by at most one event_type"); + if (range.lowerEndpoint().size() != 1) throw invalidRequest("Deletion restricted by lower bound on id_micros or at_micros is unsupported"); + if (!range.hasUpperBound() || (range.upperBoundType() != BoundType.CLOSED && range.upperEndpoint().size() == 1)) throw invalidRequest("Range deletion must specify one event_type"); + if (!range.upperEndpoint().value(0).equals(range.lowerEndpoint().value(0))) throw invalidRequest("May restrict deletion by at most one event_type"); + if (range.upperEndpoint().size() > 2) throw invalidRequest("Deletion restricted by upper bound on at_micros is unsupported"); + TraceEventType eventType = parseEventType(range.lowerEndpoint().value(0)); + if (range.upperEndpoint().size() == 1) + { + tracing().eraseEvents(txnId, eventType); + } + else + { + long before = range.upperEndpoint().value(1); + tracing().eraseEventsBefore(txnId, eventType, before); + } + } + + @Override + public void truncate() + { + tracing().eraseAllEvents(); + } + + @Override + public DataSet data() + { + SimpleDataSet dataSet = new SimpleDataSet(metadata()); + tracing().forEach(id -> true, (txnId, eventType, permits, events) -> { + events.forEach(e -> { + e.messages().forEach(m -> { + dataSet.row(txnId.toString(), eventType.name(), e.idMicros, NANOSECONDS.toMicros(m.atNanos - e.atNanos)) + .column("command_store_id", m.commandStoreId) + .column("message", m.message); + }); + }); + }); + return dataSet; + } + } + + // TODO (desired): don't report null as "null" + public static final class TxnTable extends AbstractVirtualTable implements AbstractVirtualTable.DataSet + { + static class Entry + { + final int commandStoreId; + final Command command; + + Entry(int commandStoreId, Command command) + { + this.commandStoreId = commandStoreId; + this.command = command; + } + } + private TxnTable() + { + super(parse(VIRTUAL_ACCORD_DEBUG, TXN, + "Accord per-CommandStore Transaction State", + "CREATE TABLE %s (\n" + + " txn_id text,\n" + + " command_store_id int,\n" + + " save_status text,\n" + + " route text,\n" + + " durability text,\n" + + " execute_at text,\n" + + " executes_at_least text,\n" + + " txn text,\n" + + " deps text,\n" + + " waiting_on text,\n" + + " writes text,\n" + + " result text,\n" + + " participants_owns text,\n" + + " participants_touches text,\n" + + " participants_has_touched text,\n" + + " participants_executes text,\n" + + " participants_waits_on text,\n" + + " PRIMARY KEY (txn_id, command_store_id)" + + ')', UTF8Type.instance)); + } + + @Override + public DataSet data() + { + return this; + } + + @Override + public boolean isEmpty() + { + return false; + } + + @Override + public Partition getPartition(DecoratedKey partitionKey) + { + String txnIdStr = UTF8Type.instance.compose(partitionKey.getKey()); + TxnId txnId = TxnId.parse(txnIdStr); + + List commands = new ArrayList<>(); + AccordService.instance().node().commandStores().forEachCommandStore(store -> { + Command command = ((AccordCommandStore)store).loadCommand(txnId); + if (command != null) + commands.add(new Entry(store.id(), command)); + }); + + if (commands.isEmpty()) + return null; + + SimpleDataSet ds = new SimpleDataSet(metadata); + for (Entry e : commands) + { + Command command = e.command; + ds.row(txnIdStr, e.commandStoreId) + .column("save_status", Objects.toString(command.saveStatus())) + .column("route", Objects.toString(command.route())) + .column("participants_owns", toStr(command, StoreParticipants::owns, StoreParticipants::stillOwns)) + .column("participants_touches", toStr(command, StoreParticipants::touches, StoreParticipants::stillTouches)) + .column("participants_has_touched", Objects.toString(command.participants().hasTouched())) + .column("participants_executes", toStr(command, StoreParticipants::executes, StoreParticipants::stillExecutes)) + .column("participants_waits_on", toStr(command, StoreParticipants::waitsOn, StoreParticipants::stillWaitsOn)) + .column("durability", Objects.toString(command.durability())) + .column("execute_at", Objects.toString(command.executeAt())) + .column("executes_at_least", Objects.toString(command.executesAtLeast())) + .column("txn", Objects.toString(command.partialTxn())) + .column("deps", Objects.toString(command.partialDeps())) + .column("waiting_on", Objects.toString(command.waitingOn())) + .column("writes", Objects.toString(command.writes())) + .column("result", Objects.toString(command.result())); + } + + return ds.getPartition(partitionKey); + } + + @Override + public Iterator getPartitions(DataRange range) + { + throw new UnsupportedOperationException(); + } + } + + public static final class JournalTable extends AbstractVirtualTable implements AbstractVirtualTable.DataSet + { + static class Entry + { + final int commandStoreId; + final long segment; + final int position; + final CommandChange.Builder builder; + + Entry(int commandStoreId, long segment, int position, CommandChange.Builder builder) + { + this.commandStoreId = commandStoreId; + this.segment = segment; + this.position = position; + this.builder = builder; + } + } + + private JournalTable() + { + super(parse(VIRTUAL_ACCORD_DEBUG, JOURNAL, + "Accord per-CommandStore Journal State", + "CREATE TABLE %s (\n" + + " txn_id text,\n" + + " command_store_id int,\n" + + " segment bigint,\n" + + " segment_position int,\n" + + " save_status text,\n" + + " route text,\n" + + " durability text,\n" + + " execute_at text,\n" + + " executes_at_least text,\n" + + " txn text,\n" + + " deps text,\n" + + " writes text,\n" + + " result text,\n" + + " participants_owns text,\n" + + " participants_touches text,\n" + + " participants_has_touched text,\n" + + " participants_executes text,\n" + + " participants_waits_on text,\n" + + " PRIMARY KEY (txn_id, command_store_id, segment, segment_position)" + + ')', UTF8Type.instance)); + } + + @Override + public DataSet data() + { + return this; + } + + @Override + public boolean isEmpty() + { + return false; + } + + @Override + public Partition getPartition(DecoratedKey partitionKey) + { + String txnIdStr = UTF8Type.instance.compose(partitionKey.getKey()); + TxnId txnId = TxnId.parse(txnIdStr); + + List entries = new ArrayList<>(); + AccordService.instance().node().commandStores().forEachCommandStore(store -> { + for (AccordJournal.DebugEntry e : ((AccordCommandStore)store).debugCommand(txnId)) + entries.add(new Entry(store.id(), e.segment, e.position, e.builder)); + }); + + if (entries.isEmpty()) + return null; + + SimpleDataSet ds = new SimpleDataSet(metadata); + for (Entry e : entries) + { + CommandChange.Builder b = e.builder; + StoreParticipants participants = b.participants(); + if (participants == null) participants = StoreParticipants.empty(txnId); + ds.row(txnIdStr, e.commandStoreId, e.segment, e.position) + .column("save_status", Objects.toString(b.saveStatus())) + .column("route", Objects.toString(participants.route())) + .column("participants_owns", toStr(participants, StoreParticipants::owns, StoreParticipants::stillOwns)) + .column("participants_touches", toStr(participants, StoreParticipants::touches, StoreParticipants::stillTouches)) + .column("participants_has_touched", Objects.toString(participants.hasTouched())) + .column("participants_executes", toStr(participants, StoreParticipants::executes, StoreParticipants::stillExecutes)) + .column("participants_waits_on", toStr(participants, StoreParticipants::waitsOn, StoreParticipants::stillWaitsOn)) + .column("durability", Objects.toString(b.durability())) + .column("execute_at", Objects.toString(b.executeAt())) + .column("executes_at_least", Objects.toString(b.executesAtLeast())) + .column("txn", Objects.toString(b.partialTxn())) + .column("deps", Objects.toString(b.partialDeps())) + .column("writes", Objects.toString(b.writes())) + .column("result", Objects.toString(b.result())); + } + + return ds.getPartition(partitionKey); + } + + @Override + public Iterator getPartitions(DataRange range) + { + throw new UnsupportedOperationException(); + } + } + public static class TxnBlockedByTable extends AbstractVirtualTable { enum Reason { Self, Txn, Key } @@ -716,4 +1100,24 @@ public class AccordDebugKeyspace extends VirtualKeyspace .partitioner(new LocalPartitioner(partitionKeyType)) .build(); } + + private static String toStr(Command command, Function> a, Function> b) + { + return toStr(command.participants(), a, b); + } + + private static String toStr(StoreParticipants participants, Function> a, Function> b) + { + Participants av = a.apply(participants); + Participants bv = b.apply(participants); + if (av == bv || av.equals(bv)) + return Objects.toString(av); + return av + " (" + bv + ')'; + } + + private static TraceEventType 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); } + } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 9c07b7c5aa..e166b94b87 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -443,6 +443,12 @@ public class AccordCommandStore extends CommandStore return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), durableBefore()); } + @VisibleForTesting + public List debugCommand(TxnId txnId) + { + return (List) journal.debugCommand(id, txnId); + } + public static Command prepareToCache(Command command) { // TODO (required): validate we don't have duplicate objects diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 72bca48480..c597c11f40 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -19,13 +19,16 @@ package org.apache.cassandra.service.accord; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Iterator; +import java.util.List; import java.util.NavigableMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.Consumer; +import java.util.function.Supplier; import javax.annotation.Nullable; import com.google.common.annotations.VisibleForTesting; @@ -228,6 +231,39 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return builder.construct(redundantBefore); } + public static class DebugEntry implements Supplier + { + public final long segment; + public final int position; + public final Builder builder; + + public DebugEntry(long segment, int position, Builder builder) + { + this.segment = segment; + this.position = position; + this.builder = builder; + } + + @Override + public CommandChange.Builder get() + { + return builder; + } + } + + @Override + public List debugCommand(int commandStoreId, TxnId txnId) + { + JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId); + List result = new ArrayList<>(); + journalTable.readAll(key, (long segment, int position, JournalKey k, ByteBuffer buffer, int userVersion) -> { + Builder builder = new Builder(txnId); + new AccordJournalTable.RecordConsumerAdapter<>(builder::deserializeNext).accept(segment, position, k, buffer, userVersion); + result.add(new DebugEntry(segment, position, builder)); + }); + return result; + } + @Override public Command.Minimal loadMinimal(int commandStoreId, TxnId txnId, Load load, RedundantBefore redundantBefore, DurableBefore durableBefore) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java index b8f7afa725..d04038ffcc 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java @@ -192,7 +192,7 @@ public class AccordJournalTable implements RangeSearche void read(DataInputPlus input, Version userVersion) throws IOException; } - private static class RecordConsumerAdapter implements RecordConsumer + static class RecordConsumerAdapter implements RecordConsumer { protected final Reader reader; diff --git a/src/java/org/apache/cassandra/service/accord/AccordTracing.java b/src/java/org/apache/cassandra/service/accord/AccordTracing.java new file mode 100644 index 0000000000..9de40299a0 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/AccordTracing.java @@ -0,0 +1,324 @@ +/* + * 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.service.accord; + +import java.util.AbstractList; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BiFunction; +import java.util.function.Predicate; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import accord.api.Tracing; +import accord.api.TraceEventType; +import accord.local.CommandStore; +import accord.primitives.TxnId; +import org.apache.cassandra.utils.Clock; +import org.apache.cassandra.utils.NoSpamLogger; + +public class AccordTracing +{ + private static final int MAX_EVENTS = 10000; + private static final Logger logger = LoggerFactory.getLogger(AccordTracing.class); + private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES); + + public interface ConsumeState + { + void accept(TxnId txnId, TraceEventType eventType, int permits, List events); + } + + public static class Message + { + public final long atNanos; + public final int commandStoreId; + public final String message; + + Message(int commandStoreId, String message, long atLeastNanos) + { + this.commandStoreId = commandStoreId; + this.message = message; + this.atNanos = Math.max(atLeastNanos, Clock.Global.nanoTime()); + } + + @Override + public String toString() + { + return message; + } + } + + public static class Event implements Tracing, Comparable + { + public final long idMicros = uniqueNowMicros(); + public final long atNanos = Clock.Global.nanoTime(); + final List messages = new ArrayList<>(); + + @Override + public void trace(CommandStore commandStore, String s) + { + long prevNanos = messages.isEmpty() ? 0 : messages.get(messages.size() - 1).atNanos; + int id = commandStore == null ? -1 : commandStore.id(); + if (s.length() > 1000) + s = s.substring(0, 1000); + messages.add(new Message(id, s, prevNanos + 1)); + } + + @Override + public int compareTo(Event that) + { + return Long.compareUnsigned(this.idMicros, that.idMicros); + } + + public List messages() + { + return Collections.unmodifiableList(messages); + } + } + + static class TraceState extends AbstractList + { + int permits; + int size; + Event[] events; + + void addInternal(Event event) + { + if (events == null) events = new Event[10]; + else if (size == events.length) events = Arrays.copyOf(events, events.length * 2); + events[size++] = event; + } + + void truncate(int eraseBefore) + { + System.arraycopy(events, eraseBefore, events, 0, size - eraseBefore); + Arrays.fill(events, size - eraseBefore, size, null); + size -= eraseBefore; + } + + @Override + public Event get(int index) + { + return events[index]; + } + + @Override + public int size() + { + return size; + } + } + + private static final AtomicLong lastNowMicros = new AtomicLong(); + private static long uniqueNowMicros() + { + long nowMicros = Clock.Global.currentTimeMillis() * 1000; + while (true) + { + long last = lastNowMicros.get(); + if (last >= nowMicros) + return lastNowMicros.incrementAndGet(); + if (lastNowMicros.compareAndSet(last, nowMicros)) + return nowMicros; + } + } + + final Map> stateMap = new ConcurrentHashMap<>(); + final AtomicInteger count = new AtomicInteger(); + + public Tracing trace(TxnId txnId, TraceEventType eventType) + { + if (!stateMap.containsKey(txnId)) + return null; + + class Register implements BiFunction, EnumMap> + { + Event event; + + @Override + public EnumMap apply(TxnId id, EnumMap cur) + { + if (cur == null) + return null; + + TraceState curState = cur.get(eventType); + if (curState == null || curState.permits == 0) + return cur; + + if (count.incrementAndGet() >= MAX_EVENTS) + { + count.decrementAndGet(); + noSpamLogger.warn("Too many Accord trace events stored already; delete some to continue tracing"); + } + else + { + curState.permits--; + curState.addInternal(event = new Event()); + } + return cur; + } + } + Register register = new Register(); + stateMap.compute(txnId, register); + return register.event; + } + + public void setPermits(TxnId txnId, TraceEventType eventType, int newPermits) + { + stateMap.compute(txnId, (id, cur) -> { + if (newPermits != 0) + { + if (cur == null) + cur = new EnumMap<>(TraceEventType.class); + cur.computeIfAbsent(eventType, ignore -> new TraceState()).permits = newPermits; + } + else if (cur != null) + { + TraceState curState = cur.get(eventType); + if (curState != null) + { + if (!curState.isEmpty()) curState.permits = 0; + else + { + cur.remove(eventType); + if (cur.isEmpty()) + return null; + } + } + } + return cur; + }); + } + + public void erasePermits(TxnId txnId) + { + stateMap.compute(txnId, (id, cur) -> { + if (cur == null) + return null; + + Iterator iter = cur.values().iterator(); + while (iter.hasNext()) + { + TraceState state = iter.next(); + state.permits = 0; + if (state.isEmpty()) iter.remove(); + } + return cur.isEmpty() ? null : cur; + }); + } + + public void erasePermits(TxnId txnId, TraceEventType eventType) + { + setPermits(txnId, eventType, 0); + } + + public void eraseEvents(TxnId txnId) + { + stateMap.compute(txnId, (id, cur) -> { + if (cur == null) + return null; + + Iterator iter = cur.values().iterator(); + while (iter.hasNext()) + { + TraceState state = iter.next(); + count.addAndGet(-state.size()); + state.truncate(state.size()); + if (state.permits == 0) iter.remove(); + } + return cur.isEmpty() ? null : cur; + }); + } + + public void eraseEvents(TxnId txnId, TraceEventType eventType) + { + stateMap.compute(txnId, (id, cur) -> { + if (cur != null) + { + TraceState state = cur.get(eventType); + if (state == null) + return cur; + + count.addAndGet(-state.size()); + state.truncate(state.size()); + if (state.permits == 0) + cur.remove(eventType); + if (cur.isEmpty()) + return null; + } + return cur; + }); + } + + public void eraseEventsBefore(TxnId txnId, TraceEventType eventType, long timestamp) + { + stateMap.compute(txnId, (id, cur) -> { + if (cur != null) + { + TraceState state = cur.get(eventType); + if (state == null) + return cur; + + int i = 0; + while (i < state.size() && state.get(i).idMicros < timestamp) + ++i; + state.truncate(i); + count.addAndGet(-i); + if (cur.isEmpty()) + return null; + } + return cur; + }); + } + + public void eraseAllEvents() + { + stateMap.keySet().forEach(this::eraseEvents); + } + + public void eraseAllPermits() + { + stateMap.keySet().forEach(this::erasePermits); + } + + public void forEach(Predicate include, ConsumeState forEach) + { + stateMap.forEach((txnId, state) -> { + if (include.test(txnId)) + { + // ensure lock is held for duration of callback + stateMap.compute(txnId, (id, cur) -> { + if (cur != null) + cur.forEach((event, events) -> forEach.accept(txnId, event, events.permits, Collections.unmodifiableList(events))); + return cur; + }); + } + }); + } +} diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 8c754f0d70..dd57b0e358 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -22,6 +22,8 @@ import java.util.concurrent.CancellationException; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; +import javax.annotation.Nullable; + import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,6 +33,8 @@ import accord.api.EventListener; import accord.api.ProgressLog.BlockedUntil; import accord.api.Result; import accord.api.RoutingKey; +import accord.api.Tracing; +import accord.api.TraceEventType; import accord.local.Command; import accord.local.Node; import accord.local.SafeCommand; @@ -61,6 +65,7 @@ import org.apache.cassandra.exceptions.RequestTimeoutException; import org.apache.cassandra.metrics.AccordMetrics; import org.apache.cassandra.net.ResponseContext; import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.service.accord.AccordTracing; import org.apache.cassandra.service.accord.serializers.TableMetadatasAndKeys; import org.apache.cassandra.service.accord.txn.TxnQuery; import org.apache.cassandra.service.accord.txn.TxnRead; @@ -98,7 +103,7 @@ public class AccordAgent implements Agent if (invoke != null) invoke.accept(txnId, cause); } - + private final AccordTracing tracing = new AccordTracing(); private final RandomSource random = new DefaultRandom(); protected Node.Id self; @@ -106,6 +111,17 @@ public class AccordAgent implements Agent { } + public AccordTracing tracing() + { + return tracing; + } + + @Override + public @Nullable Tracing trace(TxnId txnId, TraceEventType eventType) + { + return tracing.trace(txnId, eventType); + } + public void setNodeId(Node.Id id) { self = id; diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index a3056c9a4f..bb30e914b4 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -51,6 +51,7 @@ import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.concurrent.Condition; +import org.assertj.core.api.Assertions; import org.awaitility.Awaitility; import static accord.api.ProtocolModifiers.Toggles.SendStableMessages.TO_ALL; @@ -65,6 +66,39 @@ public class AccordDebugKeyspaceTest extends CQLTester private static final String QUERY_TXN_BLOCKED_BY = String.format("SELECT * FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_BLOCKED_BY); + private static final String QUERY_TXN = + String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN); + + private static final String QUERY_JOURNAL = + String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, 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); + + 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); + + private static final String UNSET_TRACE1 = + String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, 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 QUERY_TRACES = + String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, 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); + + 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_TRACES3 = + String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES); + + private static final String QUERY_REDUNDANT_BEFORE = + String.format("SELECT * FROM %s.%s", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE); + @BeforeClass public static void setUpClass() { @@ -85,6 +119,63 @@ public class AccordDebugKeyspaceTest extends CQLTester { createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); assertRows(execute(QUERY_TXN_BLOCKED_BY, TxnId.NONE.toString())); + assertRows(execute(QUERY_TXN, TxnId.NONE.toString())); + assertRows(execute(QUERY_JOURNAL, TxnId.NONE.toString())); + } + + @Test + public void tracing() + { + // simple test to confirm basic tracing functionality works, doesn't validate specific behaviours only requesting/querying/erasing + AccordMsgFilter filter = new AccordMsgFilter(); + MessagingService.instance().outboundSink.add(filter); + try + { + String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); + var accord = accord(); + DatabaseDescriptor.getAccord().fetch_txn = "1s"; + TxnId id = accord.node().nextTxnId(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); + + execute(SET_TRACE, 1, id.toString(), "PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); + execute(SET_TRACE, 0, id.toString(), "PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); + execute(SET_TRACE, 1, id.toString(), "PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); + execute(UNSET_TRACE1, id.toString()); + assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); + execute(SET_TRACE, 1, id.toString(), "PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); + execute(UNSET_TRACE2, id.toString(), "PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS")); + execute(SET_TRACE, 1, id.toString(), "PROGRESS"); + assertRows(execute(QUERY_TRACE, id.toString(), "PROGRESS"), row(id.toString(), "PROGRESS", 1)); + accord.node().coordinate(id, txn); + filter.preAccept.awaitThrowUncheckedOnInterrupt(); + + filter.apply.awaitThrowUncheckedOnInterrupt(); + spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "PROGRESS").size()).isGreaterThan(0)); + execute(ERASE_TRACES1, id.toString(), "FETCH", Long.MAX_VALUE); + execute(ERASE_TRACES2, id.toString(), "FETCH"); + execute(ERASE_TRACES1, id.toString(), "PROGRESS", Long.MAX_VALUE); + Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "PROGRESS").size()).isEqualTo(0); + // just check other variants don't fail + execute(ERASE_TRACES2, id.toString(), "PROGRESS"); + execute(ERASE_TRACES3, id.toString()); + } + finally + { + MessagingService.instance().outboundSink.remove(filter); + } + } + + @Test + public void redundantBefore() + { + String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); + var accord = accord(); + Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE).size()).isGreaterThan(0); } @Test @@ -98,6 +189,8 @@ public class AccordDebugKeyspaceTest extends CQLTester spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()), row(id.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name())))); + assertRows(execute(QUERY_TXN, 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")); } @Test