From 1261ba159c058b0106d13bc9582d5745274a3666 Mon Sep 17 00:00:00 2001 From: maoling Date: Mon, 31 Mar 2025 23:39:00 +0800 Subject: [PATCH] Add additional metrics around hints patch by Ling Mao; reviewed by Stefan Miklosovic, Maxim Muzafarov for CASSANDRA-20499 Co-authored-by: Stefan Miklosovic --- CHANGES.txt | 1 + .../pages/managing/operating/metrics.adoc | 44 ++++--------------- .../cassandra/hints/HintVerbHandler.java | 11 ++++- .../apache/cassandra/hints/HintsCatalog.java | 6 +++ .../apache/cassandra/hints/HintsReader.java | 16 +++++-- .../apache/cassandra/hints/HintsService.java | 20 +++++++++ .../metrics/CassandraMetricsRegistry.java | 5 +++ .../metrics/HintsServiceMetrics.java | 22 ++++++++++ .../test/metrics/HintsServiceMetricsTest.java | 39 ++++++++++++++++ 9 files changed, 124 insertions(+), 40 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index c54c0d3ffc..7bec00d66f 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.1 + * Add additional metrics around hints (CASSANDRA-20499) * Support for add and replace in IntervalTree (CASSANDRA-20513) * Enable single_sstable_uplevel by default for LCS (CASSANDRA-18509) * Introduce NativeAccessor to avoid new ByteBuffer allocation on flush for each NativeCell (CASSANDRA-20173) diff --git a/doc/modules/cassandra/pages/managing/operating/metrics.adoc b/doc/modules/cassandra/pages/managing/operating/metrics.adoc index 6fba0f8117..6b8587300a 100644 --- a/doc/modules/cassandra/pages/managing/operating/metrics.adoc +++ b/doc/modules/cassandra/pages/managing/operating/metrics.adoc @@ -880,41 +880,15 @@ Reported name format: [cols=",,",options="header",] |=== |Name |Type |Description -|HintsSucceeded a| -____ -Meter -____ - -a| -____ -A meter of the hints successfully delivered -____ - -|HintsFailed a| -____ -Meter -____ - -a| -____ -A meter of the hints that failed deliver -____ - -|HintsTimedOut a| -____ -Meter -____ - -a| -____ -A meter of the hints that timed out -____ - -|Hint_delays |Histogram |Histogram of hint delivery delays (in -milliseconds) - -|Hint_delays- |Histogram |Histogram of hint delivery delays (in -milliseconds) per peer +|HintsSucceeded|Meter|A meter of the hints successfully delivered +|HintsFailed |Meter|A meter of the hints that failed deliver +|HintsTimedOut |Meter|A meter of the hints that timed out +|Hint_delays |Histogram |Histogram of hint delivery delays (in milliseconds) +|Hint_delays- |Histogram |Histogram of hint delivery delays (in milliseconds) per peer +|HintsFileSize |Gauge| Total size of hints of a node +|HintsThrottle |Counter|Increased when rate limiter is acquired +|HintsApplySucceeded|Meter|A meter of succeeded applications of a hint +|HintsApplyFailed|Meter|A meter of failed applications of a hint |=== == SSTable Index Metrics diff --git a/src/java/org/apache/cassandra/hints/HintVerbHandler.java b/src/java/org/apache/cassandra/hints/HintVerbHandler.java index c91219d0d5..d552473adf 100644 --- a/src/java/org/apache/cassandra/hints/HintVerbHandler.java +++ b/src/java/org/apache/cassandra/hints/HintVerbHandler.java @@ -26,6 +26,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.db.partitions.PartitionUpdate; import org.apache.cassandra.exceptions.RetryOnDifferentSystemException; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.HintsServiceMetrics; import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; @@ -106,7 +107,15 @@ public final class HintVerbHandler implements IVerbHandler try { // the common path - the node is both the destination and a valid replica for the hint. - hint.applyFuture().addCallback(o -> respond(message), e -> logger.debug("Failed to apply hint", e)); + hint.applyFuture().addCallback( + o -> { + HintsServiceMetrics.hintsApplySucceeded.mark(); + respond(message); + }, + e -> { + HintsServiceMetrics.hintsApplyFailed.mark(); + logger.debug("Failed to apply hint", e); + }); } catch (RetryOnDifferentSystemException e) { diff --git a/src/java/org/apache/cassandra/hints/HintsCatalog.java b/src/java/org/apache/cassandra/hints/HintsCatalog.java index 902b239dec..87e60336a7 100644 --- a/src/java/org/apache/cassandra/hints/HintsCatalog.java +++ b/src/java/org/apache/cassandra/hints/HintsCatalog.java @@ -20,6 +20,7 @@ package org.apache.cassandra.hints; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; @@ -87,6 +88,11 @@ final class HintsCatalog } } + Collection storesCollection() + { + return stores.values(); + } + Stream stores() { return stores.values().stream(); diff --git a/src/java/org/apache/cassandra/hints/HintsReader.java b/src/java/org/apache/cassandra/hints/HintsReader.java index 2738f023f7..117e1ccdd8 100644 --- a/src/java/org/apache/cassandra/hints/HintsReader.java +++ b/src/java/org/apache/cassandra/hints/HintsReader.java @@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.exceptions.UnknownTableException; import org.apache.cassandra.io.FSReadError; +import org.apache.cassandra.metrics.HintsServiceMetrics; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.AbstractIterator; @@ -231,8 +232,7 @@ class HintsReader implements AutoCloseable, Iterable private Hint readHint(int size) throws IOException { - if (rateLimiter != null) - rateLimiter.acquire(size); + applyThrottleRateLimit(size); input.limit(size); Hint hint; @@ -338,8 +338,7 @@ class HintsReader implements AutoCloseable, Iterable private ByteBuffer readBuffer(int size) throws IOException { - if (rateLimiter != null) - rateLimiter.acquire(size); + applyThrottleRateLimit(size); input.limit(size); ByteBuffer buffer = Hint.serializer.readBufferIfLive(input, now, size, descriptor.messagingVersion()); @@ -364,4 +363,13 @@ class HintsReader implements AutoCloseable, Iterable } return true; } + + private void applyThrottleRateLimit(int size) + { + if (rateLimiter != null) + { + rateLimiter.acquire(size); + HintsServiceMetrics.hintsThrottle.inc(size); + } + } } diff --git a/src/java/org/apache/cassandra/hints/HintsService.java b/src/java/org/apache/cassandra/hints/HintsService.java index e3ae1907d7..67990f5049 100644 --- a/src/java/org/apache/cassandra/hints/HintsService.java +++ b/src/java/org/apache/cassandra/hints/HintsService.java @@ -292,6 +292,26 @@ public final class HintsService implements HintsServiceMBean return store.getTotalFileSize(); } + /** + * Get the total hints file size of current node + * + * Does not use stream on purpose as this is called in metrics + * which might be considered to be a hot path when queried frequently. + * + * The computation is reasonably fast as we have cached sizes since CASSANDRA-19477. + */ + public long getTotalHintsSize() + { + long size = 0; + for (HintsStore store : catalog.storesCollection()) + { + if (store == null) + continue; + size += store.getTotalFileSize(); + } + return size; + } + /** * Gracefully and blockingly shut down the service. * diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java index 58540f903f..44bcf2d6cf 100644 --- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java +++ b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java @@ -328,6 +328,11 @@ public class CassandraMetricsRegistry extends MetricRegistry return histogram; } + public > T gauge(MetricName name, T gauge) + { + return register(name, gauge); + } + public > T gauge(MetricName name, MetricName alias, T gauge) { T gaugeLoc = register(name, gauge); diff --git a/src/java/org/apache/cassandra/metrics/HintsServiceMetrics.java b/src/java/org/apache/cassandra/metrics/HintsServiceMetrics.java index c3203e74f4..2954d4add9 100644 --- a/src/java/org/apache/cassandra/metrics/HintsServiceMetrics.java +++ b/src/java/org/apache/cassandra/metrics/HintsServiceMetrics.java @@ -17,16 +17,20 @@ */ package org.apache.cassandra.metrics; +import java.io.Serializable; import java.net.UnknownHostException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; import com.codahale.metrics.Histogram; import com.codahale.metrics.Meter; import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.LoadingCache; import org.apache.cassandra.concurrent.ImmediateExecutor; +import org.apache.cassandra.hints.HintsService; import org.apache.cassandra.locator.InetAddressAndPort; import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; @@ -62,6 +66,12 @@ public final class HintsServiceMetrics public static final Meter hintsTimedOut = Metrics.meter(factory.createMetricName("HintsTimedOut")); public static final Meter hintsRetryDifferentSystem = Metrics.meter(factory.createMetricName("HintsRetryDifferentSystem")); + public static final Gauge hintsFileSize = Metrics.gauge(factory.createMetricName("HintsFileSize"), new TotalHintsSizeGauge()); + // Corresponding to the hinted_handoff_throttle_in_kb configuration + public static final Counter hintsThrottle = Metrics.counter(factory.createMetricName("HintsThrottle")); + + public static final Meter hintsApplySucceeded = Metrics.meter(factory.createMetricName("HintsApplySucceeded")); + public static final Meter hintsApplyFailed = Metrics.meter(factory.createMetricName("HintsApplyFailed")); /** Histogram of all hint delivery delays */ private static final Histogram globalDelayHistogram = Metrics.histogram(factory.createMetricName("Hint_delays"), false); @@ -71,6 +81,18 @@ public final class HintsServiceMetrics .executor(ImmediateExecutor.INSTANCE) .build(address -> Metrics.histogram(factory.createMetricName("Hint_delays-"+address.toString().replace(':', '.')), false)); + // because at the time of static hintsFileSize being initialized, + // HintsService.instance is null / is not initialized yet so usage of method reference is not possible, + // so this is the workaround. + private static class TotalHintsSizeGauge implements Gauge, Serializable + { + @Override + public Long getValue() + { + return HintsService.instance.getTotalHintsSize(); + } + } + public static void updateDelayMetrics(InetAddressAndPort endpoint, long delay) { if (delay <= 0) diff --git a/test/distributed/org/apache/cassandra/distributed/test/metrics/HintsServiceMetricsTest.java b/test/distributed/org/apache/cassandra/distributed/test/metrics/HintsServiceMetricsTest.java index d834919cb8..42b274e628 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/metrics/HintsServiceMetricsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/metrics/HintsServiceMetricsTest.java @@ -108,6 +108,8 @@ public class HintsServiceMetricsTest extends TestBaseImpl dropWritesForNode2.set(true); for (int i = 0; i < NUM_ROWS / 2; i++) coordinator.execute(withKeyspace("INSERT INTO %s.t (k, v) VALUES (?, ?)"), QUORUM, i, i); + // some hints have created for node1, so file size must be greater than 0 + waitUntilAsserted(() -> assertThat(countHintsFileSize(node1)).isGreaterThan(0)); dropWritesForNode2.set(false); // write the second half of the rows with the third node dropping mutations requests, @@ -115,8 +117,15 @@ public class HintsServiceMetricsTest extends TestBaseImpl dropWritesForNode3.set(true); for (int i = NUM_ROWS / 2; i < NUM_ROWS; i++) coordinator.execute(withKeyspace("INSERT INTO %s.t (k, v) VALUES (?, ?)"), QUORUM, i, i); + // another hints have created for node1, so file size must be greater than 0 + waitUntilAsserted(() -> assertThat(countHintsFileSize(node1)).isGreaterThan(0)); dropWritesForNode3.set(false); + // Hints Throttle happens in the delivery process, so must be greater than 0 + waitUntilAsserted(() -> assertThat(countHintsThrottle(node1)).isGreaterThan(0)); + waitUntilAsserted(() -> assertThat(countHintsApplySucceeded(node1)).isEqualTo(0)); + waitUntilAsserted(() -> assertThat(countHintsApplyFailed(node1)).isEqualTo(0)); + // wait until all the hints have been successfully applied to the nodes that have been dropping mutations waitUntilAsserted(() -> assertThat(countRows(node2)).isEqualTo(countRows(node3)).isEqualTo(NUM_ROWS)); @@ -143,6 +152,13 @@ public class HintsServiceMetricsTest extends TestBaseImpl assertThat(countHintsFailed(node)).isEqualTo(0); assertThat(countHintsTimedOut(node)).isEqualTo(0); assertThat(countHintsRetryDifferentSystem(node)).isEqualTo(0); + + assertThat(countHintsFileSize(node)).isEqualTo(0); + assertThat(countHintsThrottle(node)).isEqualTo(0); + // node two and three must apply these hints which belongs to them, so must be greater than 0 + assertThat(countHintsApplySucceeded(node)).isGreaterThan(0); + assertThat(countHintsApplyFailed(node)).isEqualTo(0); + assertThat(countGlobalDelays(node)).isEqualTo(0); cluster.forEach(target -> assertThat(countEndpointDelays(node, target)).isEqualTo(0)); } @@ -187,6 +203,29 @@ public class HintsServiceMetricsTest extends TestBaseImpl return node.callOnInstance(() -> HintsServiceMetrics.hintsRetryDifferentSystem.getCount()); } + private static Long countHintsFileSize(IInvokableInstance node) + { + return node.callOnInstance(HintsServiceMetrics.hintsFileSize::getValue); + } + + @SuppressWarnings("Convert2MethodRef") + private static Long countHintsApplySucceeded(IInvokableInstance node) + { + return node.callOnInstance(() -> HintsServiceMetrics.hintsApplySucceeded.getCount()); + } + + @SuppressWarnings("Convert2MethodRef") + private static Long countHintsApplyFailed(IInvokableInstance node) + { + return node.callOnInstance(() -> HintsServiceMetrics.hintsApplyFailed.getCount()); + } + + @SuppressWarnings("Convert2MethodRef") + private static Long countHintsThrottle(IInvokableInstance node) + { + return node.callOnInstance(() -> HintsServiceMetrics.hintsThrottle.getCount()); + } + private static Long countGlobalDelays(IInvokableInstance node) { return getHistogramCount(node, "org.apache.cassandra.metrics.HintsService.Hint_delays");