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

Add additional metrics around hints 

patch by Ling Mao; reviewed by Stefan Miklosovic, Maxim Muzafarov for CASSANDRA-20499

Co-authored-by: Stefan Miklosovic <smiklosovic@apache.org>
This commit is contained in:
maoling 2025-03-31 23:39:00 +08:00 committed by Stefan Miklosovic
parent 3a506eb104
commit 1261ba159c
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
9 changed files with 124 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

44
doc/modules/cassandra/pages/managing/operating/metrics.adoc
View File

@ -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-<PeerIP> |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-<PeerIP> |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

11
src/java/org/apache/cassandra/hints/HintVerbHandler.java
View File

@ -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<HintMessage>
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)
{

6
src/java/org/apache/cassandra/hints/HintsCatalog.java
View File

@ -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<HintsStore> storesCollection()
{
return stores.values();
}
Stream<HintsStore> stores()
{
return stores.values().stream();

16
src/java/org/apache/cassandra/hints/HintsReader.java
View File

@ -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<HintsReader.Page>
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<HintsReader.Page>
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<HintsReader.Page>
}
return true;
}
private void applyThrottleRateLimit(int size)
{
if (rateLimiter != null)
{
rateLimiter.acquire(size);
HintsServiceMetrics.hintsThrottle.inc(size);
}
}
}

20
src/java/org/apache/cassandra/hints/HintsService.java
View File

@ -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.
*

5
src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java
View File

@ -328,6 +328,11 @@ public class CassandraMetricsRegistry extends MetricRegistry
return histogram;
}
public <T extends Gauge<?>> T gauge(MetricName name, T gauge)
{
return register(name, gauge);
}
public <T extends Gauge<?>> T gauge(MetricName name, MetricName alias, T gauge)
{
T gaugeLoc = register(name, gauge);

22
src/java/org/apache/cassandra/metrics/HintsServiceMetrics.java
View File

@ -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<Long> 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<Long>, Serializable
{
@Override
public Long getValue()
{
return HintsService.instance.getTotalHintsSize();
}
}
public static void updateDelayMetrics(InetAddressAndPort endpoint, long delay)
{
if (delay <= 0)

39
test/distributed/org/apache/cassandra/distributed/test/metrics/HintsServiceMetricsTest.java
View File

@ -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");