Improve Observability:

 - Track all active Coordinations
 - Refactor Replica/Coordinator metrics and report Coordinator exhausted/preempted/timeout
 - DurabilityQueue metrics and visibility
Also Fix:
 - WaitingState can get cause distributed stall when asked to wait for CanApply if not yet PreCommitted; track separate querying state and advance this to the next achievable state rather than the desired final state
 - Stalled coordinators should not prevent recovery
 - Edge case with fetch unable to make progress when pre-bootstrap and all peers have GC'd
 - Dependency initialisation for sync points across certain ownership changes
 - SyncPoint propagation may not include all of the epochs required on the receiving node for ranges they have lost but not closed, and receiving node does not validate them
 - Stable tracker accounting with LocalExecute
 - Do not prune non-durable APPLIED as must be reported in dependencies until durably applied (so as not to break recovery)
 - Ensure we cannot race with replies when initiating Coordination
 - ProgressLog does not guarantee to clear home or waiting states when erased or invalidated by compaction
 - WaitingState on non-home shard cannot guarantee progress once home shard is Erased
 - WaitingOnSync handles retired ranges incorrectly
Also Improve:
 - Standardise failure accounting, use null to represent single reply timeouts
 - BurnTest record/replay to/from file

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20878

Author: belliottsmith

modules/accord

@@ -41,6 +41,13 @@
 import com.google.common.collect.BoundType;
 import com.google.common.collect.Range;
 import com.google.common.collect.Sets;
+
+import accord.coordinate.AbstractCoordination;
+import accord.coordinate.Coordination;
+import accord.coordinate.Coordinations;
+import accord.coordinate.PrepareRecovery;
+import accord.coordinate.tracking.AbstractTracker;
+import accord.utils.SortedListMap;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.EmptyIterators;
 import org.apache.cassandra.db.filter.ClusteringIndexFilter;
@@ -57,7 +64,6 @@
 import accord.coordinate.FetchData;
 import accord.coordinate.FetchRoute;
 import accord.coordinate.MaybeRecover;
-import accord.coordinate.RecoverWithRoute;
 import accord.impl.CommandChange;
 import accord.impl.progresslog.DefaultProgressLog;
 import accord.impl.progresslog.TxnStateKind;
@@ -151,6 +157,8 @@
 
 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 DURABILITY_SERVICE = "durability_service";
@@ -173,6 +181,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace
     private AccordDebugKeyspace()
     {
         super(VIRTUAL_ACCORD_DEBUG, List.of(
+            new ExecutorsTable(),
+            new CoordinationsTable(),
             new CommandsForKeyTable(),
             new CommandsForKeyUnmanagedTable(),
             new DurabilityServiceTable(),
@@ -192,6 +202,118 @@ private AccordDebugKeyspace()
         ));
     }
 
+    // TODO (desired): human readable packed key tracker (but requires loading Txn, so might be preferable to only do conditionally)
+    public static final class ExecutorsTable extends AbstractVirtualTable
+    {
+        private ExecutorsTable()
+        {
+            super(parse(VIRTUAL_ACCORD_DEBUG, EXECUTORS,
+                        "Accord Executor State",
+                        "CREATE TABLE %s (\n" +
+                        "  executor_id int,\n" +
+                        "  status text,\n" +
+                        "  position int,\n" +
+                        "  unique_position int,\n" +
+                        "  description text,\n" +
+                        "  command_store_id int,\n" +
+                        "  txn_id 'TxnIdUtf8Type',\n" +
+                        "  txn_id_additional 'TxnIdUtf8Type',\n" +
+                        "  keys text,\n" +
+                        "  keysLoad text,\n" +
+                        "  keysLoadFor text,\n" +
+                        "  PRIMARY KEY (executor_id, status, position, unique_position)" +
+                        ')', UTF8Type.instance));
+        }
+
+        @Override
+        public DataSet data()
+        {
+            AccordCommandStores commandStores = (AccordCommandStores) AccordService.instance().node().commandStores();
+            SimpleDataSet ds = new SimpleDataSet(metadata());
+
+            for (AccordExecutor executor : commandStores.executors())
+            {
+                int uniquePos = 0;
+                int executorId = executor.executorId();
+                AccordExecutor.TaskInfo prev = null;
+                for (AccordExecutor.TaskInfo info : executor.taskSnapshot())
+                {
+                    if (prev != null && info.status() == prev.status() && info.position() == prev.position()) ++uniquePos;
+                    else uniquePos = 0;
+                    prev = info;
+                    PreLoadContext preLoadContext = info.preLoadContext();
+                    ds.row(executorId, info.status(), info.position(), uniquePos)
+                      .column("description", info.describe())
+                      .column("command_store_id", info.commandStoreId())
+                      .column("txn_id", preLoadContext == null ? null : preLoadContext.primaryTxnId())
+                      .column("txn_id_additional", preLoadContext == null ? null : preLoadContext.additionalTxnId())
+                      .column("keys", preLoadContext == null ? null : preLoadContext.keys())
+                      .column("keysLoad", preLoadContext == null ? null : preLoadContext.loadKeys())
+                      .column("keysLoadFor", preLoadContext == null ? null : preLoadContext.loadKeysFor())
+                    ;
+                }
+            }
+            return ds;
+        }
+    }
+
+    // TODO (desired): human readable packed key tracker (but requires loading Txn, so might be preferable to only do conditionally)
+    public static final class CoordinationsTable extends AbstractVirtualTable
+    {
+        private CoordinationsTable()
+        {
+            super(parse(VIRTUAL_ACCORD_DEBUG, COORDINATIONS,
+                        "Accord Coordination State",
+                        "CREATE TABLE %s (\n" +
+                        "  txn_id int,\n" +
+                        "  kind text,\n" +
+                        "  coordination_id int,\n" +
+                        "  description text,\n" +
+                        "  nodes text,\n" +
+                        "  nodes_inflight text,\n" +
+                        "  nodes_contacted text,\n" +
+                        "  participants text,\n" +
+                        "  replies text,\n" +
+                        "  tracker text,\n" +
+                        "  PRIMARY KEY (txn_id, kind, coordination_id)" +
+                        ')', UTF8Type.instance));
+        }
+
+        @Override
+        public DataSet data()
+        {
+            Coordinations coordinations = AccordService.instance().node().coordinations();
+            SimpleDataSet ds = new SimpleDataSet(metadata());
+            for (Coordination c : coordinations)
+            {
+                ds.row(c.txnId(), c.kind().toString(), c.coordinationId())
+                  .column("nodes", toStringOrNull(c.nodes()))
+                  .column("nodes_inflight", toStringOrNull(c.inflight()))
+                  .column("nodes_contacted", toStringOrNull(c.contacted()))
+                  .column("description", c.describe())
+                  .column("participants", toStringOrNull(c.scope()))
+                  .column("replies", summarise(c.replies()))
+                  .column("tracker", summarise(c.tracker()));
+            }
+            return ds;
+        }
+
+        private static String summarise(@Nullable SortedListMap<Node.Id, ?> replies)
+        {
+            if (replies == null)
+                return null;
+            return AbstractCoordination.summariseReplies(replies, 60);
+        }
+
+        private static String summarise(@Nullable AbstractTracker<?> tracker)
+        {
+            if (tracker == null)
+                return null;
+            return tracker.summariseTracker();
+        }
+    }
+
+
     // TODO (desired): don't report null as "null"
     public static final class CommandsForKeyTable extends AbstractVirtualTable implements AbstractVirtualTable.DataSet
     {
@@ -529,7 +651,7 @@ public DataSet data()
         private static void addRow(SimpleDataSet ds, int executorId, String scope, AccordCache.ImmutableStats stats)
         {
             ds.row(executorId, scope)
-              .column("queries", stats.queries)
+              .column("queries", stats.hits + stats.misses)
               .column("hits", stats.hits)
               .column("misses", stats.misses);
         }
@@ -1365,7 +1487,7 @@ private void recover(TxnId txnId, @Nullable Route<?> route, AsyncResult.Settable
             Node node = AccordService.instance().node();
             if (Route.isFullRoute(route))
             {
-                RecoverWithRoute.recover(node, node.someSequentialExecutor(), txnId, NotKnownToBeInvalid, (FullRoute<?>) route, null, LatentStoreSelector.standard(), (success, fail) -> {
+                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));

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -18,35 +18,94 @@
 
 package org.apache.cassandra.metrics;
 
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
+import java.util.function.Function;
+import java.util.function.ToLongFunction;
 
-import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Gauge;
+import org.apache.cassandra.service.accord.AccordExecutor;
+import org.apache.cassandra.service.accord.IAccordService;
 
-import static org.apache.cassandra.metrics.CacheMetrics.TYPE_NAME;
+import static org.apache.cassandra.metrics.AccordMetricUtils.fromAccordService;
 import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
 
-public class AccordCacheMetrics extends CacheAccessMetrics
+public class AccordCacheMetrics
 {
-    public static final String OBJECT_SIZE = "ObjectSize";
+    public static final String ACCORD_CACHE = "AccordCache";
+    public static final AccordCacheMetrics CommandsCacheMetrics = new AccordCacheMetrics("Commands");
+    public static final AccordCacheMetrics CommandsForKeyCacheMetrics = new AccordCacheMetrics("CommandsForKey");
+    public static final AccordCacheGlobalMetrics Global = new AccordCacheGlobalMetrics();
 
-    public final Histogram objectSize;
+    // not sure why we create these wrapper objects that can only be instantiated once, but it's a pattern in this package so...
+    public static class AccordCacheGlobalMetrics
+    {
+        final Gauge<Long> usedBytes;
+        final Gauge<Long> unreferencedBytes;
+
+        public AccordCacheGlobalMetrics()
+        {
+            DefaultNameFactory factory = new DefaultNameFactory("AccordCache");
+            this.usedBytes = Metrics.gauge(factory.createMetricName("UsedBytes"), fromAccordService(sumExecutors(executor -> executor.cacheUnsafe().weightedSize()), 0L));
+            this.unreferencedBytes = Metrics.gauge(factory.createMetricName("UnreferencedBytes"), fromAccordService(sumExecutors(executor -> executor.cacheUnsafe().unreferencedBytes()), 0L));
+        }
+
+        private static Function<IAccordService, Long> sumExecutors(ToLongFunction<AccordExecutor> f)
+        {
+            return service -> {
+                long sum = 0;
+                for (AccordExecutor executor : service.executors())
+                    sum += f.applyAsLong(executor);
+                return sum;
+            };
+        }
+    }
+
+    public static class Shard
+    {
+        public final ShardedHitRate.HitRateShard hitRate;
+        public final LogLinearHistogram objectSize;
+
+        public Shard(ShardedHitRate.HitRateShard hitRate, LogLinearHistogram objectSize)
+        {
+            this.hitRate = hitRate;
+            this.objectSize = objectSize;
+        }
+    }
 
-    private final Map<String, CacheAccessMetrics> instanceMetrics = new ConcurrentHashMap<>(2);
+    public final ShardedHitRate hitRate = new ShardedHitRate();
+    public final ShardedHistogram objectSize;
 
-    private final String scope;
+    public final Gauge<Long> hits;
+    public final Gauge<Long> misses;
+    public final Gauge<Long> requests;
+    public final Gauge<Double> requestRate1m;
+    public final Gauge<Double> requestRate5m;
+    public final Gauge<Double> requestRate15m;
+    public final Gauge<Double> hitRateAllTime;
+    public final Gauge<Double> hitRate1m;
+    public final Gauge<Double> hitRate5m;
+    public final Gauge<Double> hitRate15m;
+    private final String subTypeName;
 
-    public AccordCacheMetrics(String scope)
+    public AccordCacheMetrics(String subTypeName)
     {
-        super(new DefaultNameFactory(TYPE_NAME, scope));
-        objectSize = Metrics.histogram(factory.createMetricName(OBJECT_SIZE), false);
-        this.scope = scope;
+        DefaultNameFactory factory = new DefaultNameFactory("AccordCache", subTypeName);
+        this.objectSize = Metrics.shardedHistogram(factory.createMetricName("EntrySize"));
+        this.hits = Metrics.gauge(factory.createMetricName("Hits"), hitRate::totalHits);
+        this.misses = Metrics.gauge(factory.createMetricName("Misses"), hitRate::totalMisses);
+        this.requests = Metrics.gauge(factory.createMetricName("Requests"), hitRate::totalRequests);
+        this.requestRate1m = Metrics.gauge(factory.createMetricName("Requests"), () -> hitRate.requestsPerSecond(1));
+        this.requestRate5m = Metrics.gauge(factory.createMetricName("Requests"), () -> hitRate.requestsPerSecond(5));
+        this.requestRate15m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIFTEEN_MINUTE + "RequestRate"), () -> hitRate.requestsPerSecond(15));
+        this.hitRate1m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.ONE_MINUTE + "HitRate"), () -> hitRate.hitRate(1));
+        this.hitRate5m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIVE_MINUTE + "HitRate"), () -> hitRate.hitRate(5));
+        this.hitRate15m = Metrics.gauge(factory.createMetricName(RatioGaugeSet.FIFTEEN_MINUTE + "HitRate"), () -> hitRate.hitRate(15));
+        this.hitRateAllTime = Metrics.gauge(factory.createMetricName("Misses"), hitRate::hitRateAllTime);
+        this.subTypeName = subTypeName;
     }
 
-    public CacheAccessMetrics forInstance(Class<?> klass)
+    public Shard newShard(Lock guardedBy)
     {
-        // cannot make Class<?> hashCode deterministic, as cannot rewrite - so cannot safely use as Map key if want deterministic simulation
-        // (or we need to create extra hoops to catch this specific case in method rewriting)
-        return instanceMetrics.computeIfAbsent(klass.getSimpleName(), k -> new CacheAccessMetrics(new DefaultNameFactory(TYPE_NAME, String.format("%s-%s", scope, k))));
+        return new Shard(hitRate.newShard(guardedBy), objectSize.newShard(guardedBy));
     }
 }

src/java/org/apache/cassandra/metrics/AccordCacheMetrics.java

@@ -72,5 +72,4 @@ public void release()
         Metrics.remove(factory.createMetricName("Preempted"));
         Metrics.remove(factory.createMetricName("TopologyMismatches"));
     }
-
 }