CASSANDRA-20986: Add metrics for Mutation Tracking

src/java/org/apache/cassandra/journal/Journal.java

@@ -832,6 +832,22 @@ private void shutDown() throws InterruptedException
         }
     }
 
+    public long getDiskSpaceUsed()
+    {
+        long totalSize = 0;
+
+        try (ReferencedSegments<K, V> refs = selectAndReference(s -> true))
+        {
+            for (Segment<K, V> segment : refs.all())
+            {
+                File dataFile = segment.descriptor.fileFor(Component.DATA);
+                if (dataFile.exists())
+                    totalSize += dataFile.length();
+            }
+        }
+        return totalSize;
+    }
+
     private ActiveSegment<K, V> createSegment()
     {
         Descriptor descriptor = Descriptor.create(directory, nextSegmentId.getAndIncrement(), params.userVersion());

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

@@ -139,6 +139,7 @@ public class CassandraMetricsRegistry extends MetricRegistry
                                    .add(KeyspaceMetrics.TYPE_NAME)
                                    .add(MemtablePool.TYPE_NAME)
                                    .add(MessagingMetrics.TYPE_NAME)
+                                   .add(MutationTrackingMetrics.TYPE_NAME)
                                    .add(MutualTlsMetrics.TYPE_NAME)
                                    .add(PaxosMetrics.TYPE_NAME)
                                    .add(ReadRepairMetrics.TYPE_NAME)

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

@@ -0,0 +1,57 @@
+/*
+ * 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.metrics;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Histogram;
+import org.apache.cassandra.replication.MutationJournal;
+import org.apache.cassandra.replication.MutationTrackingService;
+
+import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
+
+public class MutationTrackingMetrics
+{
+    public static final String TYPE_NAME = "MutationTracking";
+    private static final MetricNameFactory factory = new DefaultNameFactory(TYPE_NAME);
+
+    public static final MutationTrackingMetrics instance = new MutationTrackingMetrics();
+
+    public final Counter broadcastOffsetsDiscovered; // Newly-witnessed offsets discovered via broadcast
+    public final Counter writeTimeOffsetsDiscovered; // Newly-witnessed offsets discovered at write time
+    public final Histogram readSummarySize; // Read summary sizes
+    public final Gauge<Long> unreconciledMutationCount; // Number of unreconciled mutations
+    public final Gauge<Long> journalDiskSpaceUsed; // Size of MutationJournal on disk
+
+    @SuppressWarnings("Convert2MethodRef")
+    private MutationTrackingMetrics()
+    {
+        broadcastOffsetsDiscovered = Metrics.counter(factory.createMetricName("broadcastOffsetsDiscovered"));
+        writeTimeOffsetsDiscovered = Metrics.counter(factory.createMetricName("WriteTimeOffsetsDiscovered"));
+        readSummarySize = Metrics.histogram(factory.createMetricName("ReadSummarySize"), false);
+        unreconciledMutationCount = Metrics.register(
+                factory.createMetricName("UnreconciledMutationCount"),
+                () -> MutationTrackingService.instance.getUnreconciledMutationCount()
+        );
+        journalDiskSpaceUsed = Metrics.register(
+                factory.createMetricName("JournalDiskSpaceUsed"),
+                () -> MutationJournal.instance.getDiskSpaceUsed()
+        );
+    }
+}

src/java/org/apache/cassandra/replication/CoordinatorLog.java

@@ -45,6 +45,7 @@
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.metrics.MutationTrackingMetrics;
 import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.tcm.ClusterMetadata;
@@ -208,8 +209,14 @@ void updateReplicatedOffsets(Offsets offsets, boolean persisted, int onNodeId)
 
     private void updateWitnessedReplicatedOffsets(Offsets offsets, int onNodeId)
     {
+        // Track newly-witnessed offsets from broadcasts (use array for lambda)
+        int[] newlyWitnessedCount = {0};
+
         witnessedOffsets.get(onNodeId).addAll(offsets, (ignore, start, end) ->
         {
+            // Count the newly-witnessed offsets in this range
+            newlyWitnessedCount[0] += (end - start + 1);
+
             for (int offset = start; offset <= end; ++offset)
             {
                 // TODO (desired): use the fact that Offsets are ordered to optimise this look up
@@ -220,6 +227,9 @@ private void updateWitnessedReplicatedOffsets(Offsets offsets, int onNodeId)
                 }
             }
         });
+
+        // Record metric for newly witnessed offsets only
+        MutationTrackingMetrics.instance.broadcastOffsetsDiscovered.inc(newlyWitnessedCount[0]);
     }
 
     private void updatePersistedReplicatedOffsets(Offsets offsets, int onNodeId)
@@ -272,6 +282,19 @@ Offsets.Immutable collectReconciledOffsets()
         }
     }
 
+    public long getUnreconciledCount()
+    {
+        lock.readLock().lock();
+        try
+        {
+            return unreconciledMutations.size();
+        }
+        finally
+        {
+            lock.readLock().unlock();
+        }
+    }
+
     boolean startWriting(Mutation mutation)
     {
         lock.writeLock().lock();
@@ -301,6 +324,9 @@ void finishWriting(Mutation mutation)
             if (!witnessedOffsets.get(localNodeId).add(offset))
                 return;
 
+            // Track write-time discovery of newly-witnessed offset
+            MutationTrackingMetrics.instance.writeTimeOffsetsDiscovered.inc();
+
             unreconciledMutations.finishWriting(mutation);
 
             if (remoteReplicasWitnessed(offset))

src/java/org/apache/cassandra/replication/MutationJournal.java

@@ -746,4 +746,9 @@ public int countStaticSegmentsForTesting()
     {
         return journal.countStaticSegmentsForTesting();
     }
+
+    public long getDiskSpaceUsed()
+    {
+        return journal.getDiskSpaceUsed();
+    }
 }

src/java/org/apache/cassandra/replication/MutationTrackingService.java

@@ -54,6 +54,7 @@
 import org.apache.cassandra.exceptions.RequestFailure;
 import org.apache.cassandra.gms.FailureDetector;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.metrics.MutationTrackingMetrics;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.net.Verb;
@@ -330,7 +331,9 @@ public MutationSummary createSummaryForKey(DecoratedKey key, TableId tableId, bo
         shardLock.readLock().lock();
         try
         {
-            return getOrCreateShards(tableId).createSummaryForKey(key, tableId, includePending);
+            MutationSummary summary = getOrCreateShards(tableId).createSummaryForKey(key, tableId, includePending);
+            MutationTrackingMetrics.instance.readSummarySize.update(summary.size());
+            return summary;
         }
         finally
         {
@@ -370,6 +373,20 @@ void forEachKeyspace(Consumer<KeyspaceShards> consumer)
         }
     }
 
+    public long getUnreconciledMutationCount()
+    {
+        if (!isStarted())
+            return 0L;
+
+        final long[] count = {0L};
+        forEachKeyspace(ks -> {
+            ks.forEachShard(shard -> {
+                count[0] += shard.getUnreconciledCount();
+            });
+        });
+        return count[0];
+    }
+
     public void collectLocallyMissingMutations(MutationSummary remoteSummary, Log2OffsetsMap.Mutable into)
     {
         shardLock.readLock().lock();

src/java/org/apache/cassandra/replication/Shard.java

@@ -297,6 +297,16 @@ private CoordinatorLog getOrCreate(CoordinatorLogId logId)
         return getOrCreate(logId.asLong());
     }
 
+    public long getUnreconciledCount()
+    {
+        long count = 0;
+        for (CoordinatorLog log : logs.values())
+        {
+            count += log.getUnreconciledCount();
+        }
+        return count;
+    }
+
     @Nonnull
     private CoordinatorLog get(CoordinatorLogId logId)
     {

src/java/org/apache/cassandra/replication/UnreconciledMutations.java

@@ -228,6 +228,11 @@ boolean isEmpty()
         return statesMap.isEmpty();
     }
 
+    public int size()
+    {
+        return statesMap.size();
+    }
+
     static UnreconciledMutations loadFromJournal(Node2OffsetsMap witnessedOffsets, int localNodeId)
     {
         UnreconciledMutations result = new UnreconciledMutations();