This is an automated email from the ASF dual-hosted git repository.
jojochuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 7d06fc7a2a6 HDDS-16304. Convert RpcMetrics latency counters to
lock-free ConcurrentMutableStat (#11425)
7d06fc7a2a6 is described below
commit 7d06fc7a2a681c45722483657aabcab8188dbf71
Author: Andrey Yarovoy <[email protected]>
AuthorDate: Wed Oct 7 16:47:59 2026 -0400
HDDS-16304. Convert RpcMetrics latency counters to lock-free
ConcurrentMutableStat (#11425)
Generated-by: Claude Code (claude-opus)
---
.../org/apache/hadoop/ipc_/metrics/RpcMetrics.java | 41 ++++-
.../apache/hadoop/ipc_/metrics/TestRpcMetrics.java | 195 +++++++++++++++++++++
2 files changed, 229 insertions(+), 7 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/metrics/RpcMetrics.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/metrics/RpcMetrics.java
index b4726d04bbd..83d831de142 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/metrics/RpcMetrics.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ipc_/metrics/RpcMetrics.java
@@ -22,6 +22,9 @@
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.ipc_.Server;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.MetricsCollector;
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.metrics2.MetricsSource;
import org.apache.hadoop.metrics2.MetricsTag;
import org.apache.hadoop.metrics2.annotation.Metric;
import org.apache.hadoop.metrics2.annotation.Metrics;
@@ -29,16 +32,22 @@
import org.apache.hadoop.metrics2.lib.MetricsRegistry;
import org.apache.hadoop.metrics2.lib.MutableCounterLong;
import org.apache.hadoop.metrics2.lib.MutableQuantiles;
-import org.apache.hadoop.metrics2.lib.MutableRate;
+import org.apache.hadoop.ozone.util.ConcurrentMutableRate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* This class is for maintaining the various RPC statistics
* and publishing them through the metrics interfaces.
+ *
+ * The three latency rates are kept outside the {@link MetricsRegistry} as
+ * lock-free {@link ConcurrentMutableRate}s, because they are updated on every
+ * RPC by every handler thread and {@code MutableStat.add} is synchronized.
+ * This class therefore implements {@link MetricsSource} itself and snapshots
+ * them next to the registry, keeping the emitted metric names unchanged.
*/
@Metrics(about="Aggregate RPC metrics", context="rpc")
-public class RpcMetrics {
+public class RpcMetrics implements MetricsSource {
static final Logger LOG = LoggerFactory.getLogger(RpcMetrics.class);
final Server server;
@@ -55,6 +64,11 @@ public class RpcMetrics {
registry = new MetricsRegistry("rpc")
.tag("port", "RPC port", port)
.tag("serverName", "Name of the RPC server", server.getServerName());
+ // Names match what MutableMetricsFactory derived from the field names, so
+ // the emitted metrics are unchanged
+ rpcQueueTime = new ConcurrentMutableRate("RpcQueueTime", "Queue time",
false);
+ rpcLockWaitTime = new ConcurrentMutableRate("RpcLockWaitTime", "Lock wait
time", false);
+ rpcProcessingTime = new ConcurrentMutableRate("RpcProcessingTime",
"Processing time", false);
int[] intervals = conf.getInts(
CommonConfigurationKeys.RPC_METRICS_PERCENTILES_INTERVALS_KEY);
rpcQuantileEnable = (intervals.length > 0) && conf.getBoolean(
@@ -94,11 +108,11 @@ public static RpcMetrics create(Server server,
Configuration conf) {
@Metric("Number of received bytes") MutableCounterLong receivedBytes;
@Metric("Number of sent bytes") MutableCounterLong sentBytes;
- @Metric("Queue time") MutableRate rpcQueueTime;
+ final ConcurrentMutableRate rpcQueueTime;
MutableQuantiles[] rpcQueueTimeQuantiles;
- @Metric("Lock wait time") MutableRate rpcLockWaitTime;
+ final ConcurrentMutableRate rpcLockWaitTime;
MutableQuantiles[] rpcLockWaitTimeQuantiles;
- @Metric("Processing time") MutableRate rpcProcessingTime;
+ final ConcurrentMutableRate rpcProcessingTime;
MutableQuantiles[] rpcProcessingTimeQuantiles;
@Metric("Number of authentication failures")
MutableCounterLong rpcAuthenticationFailures;
@@ -243,10 +257,10 @@ public void incrSlowRpc() {
rpcSlowCalls.incr();
}
/**
- * Returns a MutableRate Counter.
+ * Returns the processing time rate counter.
* @return Mutable Rate
*/
- public MutableRate getRpcProcessingTime() {
+ public ConcurrentMutableRate getRpcProcessingTime() {
return rpcProcessingTime;
}
@@ -285,4 +299,17 @@ public long getRpcSlowCalls() {
public MetricsTag getTag(String tagName) {
return registry.getTag(tagName);
}
+
+ /**
+ * Emits the registry (tags, counters, gauges and quantiles) and the three
+ * latency rates that are held outside of it
+ */
+ @Override
+ public void getMetrics(MetricsCollector collector, boolean all) {
+ MetricsRecordBuilder builder = collector.addRecord(registry.info());
+ registry.snapshot(builder, all);
+ rpcQueueTime.snapshot(builder, all);
+ rpcLockWaitTime.snapshot(builder, all);
+ rpcProcessingTime.snapshot(builder, all);
+ }
}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ipc_/metrics/TestRpcMetrics.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ipc_/metrics/TestRpcMetrics.java
new file mode 100644
index 00000000000..ae9856daf84
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ipc_/metrics/TestRpcMetrics.java
@@ -0,0 +1,195 @@
+/*
+ * 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.hadoop.ipc_.metrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc_.Server;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
+import org.apache.hadoop.metrics2.lib.MetricsAnnotations;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link RpcMetrics}.
+ *
+ * The latency rates are held outside the {@link
org.apache.hadoop.metrics2.lib.MetricsRegistry}
+ * as lock-free {@link org.apache.hadoop.ozone.util.ConcurrentMutableRate}s,
so these tests pin the
+ * emitted metric names and the hybrid {@link MetricsSource} wiring that keeps
the registry-backed
+ * metrics in the same record.
+ */
+public class TestRpcMetrics {
+
+ private static final int PORT = 9999;
+ private static final String SERVER_NAME = "TestProtocol";
+
+ private RpcMetrics metrics;
+ private MetricsSource source;
+
+ @BeforeEach
+ public void setUp() {
+ Server server = mock(Server.class);
+ when(server.getListenerAddress()).thenReturn(new InetSocketAddress(PORT));
+ when(server.getServerName()).thenReturn(SERVER_NAME);
+ when(server.getNumOpenConnections()).thenReturn(7);
+ when(server.getNumOpenConnectionsPerUser()).thenReturn("{\"alice\":7}");
+ when(server.getCallQueueLen()).thenReturn(3);
+ when(server.getNumDroppedConnections()).thenReturn(5L);
+
+ metrics = new RpcMetrics(server, new Configuration());
+ // Same reflection DefaultMetricsSystem.register() performs: it populates
the
+ // @Metric counter fields and the @Metric methods into the registry
+ source = MetricsAnnotations.makeSource(metrics);
+ }
+
+ @Test
+ public void testRpcMetricsIsUsedAsItsOwnSource() {
+ assertSame(metrics, source);
+ }
+
+ @Test
+ public void testLatencyRateNamesAreUnchanged() {
+ metrics.addRpcQueueTime(10);
+ metrics.addRpcQueueTime(20);
+ metrics.addRpcLockWaitTime(4);
+ metrics.addRpcProcessingTime(100);
+ metrics.addRpcProcessingTime(300);
+
+ Map<String, Number> values = snapshotMetrics();
+
+ assertEquals(2L, values.get("RpcQueueTimeNumOps").longValue());
+ assertEquals(15.0, values.get("RpcQueueTimeAvgTime").doubleValue(), 0.001);
+ assertEquals(1L, values.get("RpcLockWaitTimeNumOps").longValue());
+ assertEquals(4.0, values.get("RpcLockWaitTimeAvgTime").doubleValue(),
0.001);
+ assertEquals(2L, values.get("RpcProcessingTimeNumOps").longValue());
+ assertEquals(200.0, values.get("RpcProcessingTimeAvgTime").doubleValue(),
0.001);
+
+ // the rates are not extended, matching @Metric.always() defaulting to
false
+ assertThat(values).doesNotContainKeys("RpcQueueTimeStdevTime",
+ "RpcProcessingTimeStdevTime");
+ }
+
+ @Test
+ public void testRegistryMetricsAndTagsStayInTheSameRecord() {
+ metrics.incrSentBytes(42);
+ metrics.incrReceivedBytes(24);
+ metrics.incrSlowRpc();
+
+ MetricsCollectorImpl collector = new MetricsCollectorImpl();
+ source.getMetrics(collector, true);
+ assertEquals(1, collector.getRecords().size());
+ MetricsRecord record = collector.getRecords().get(0);
+ assertEquals("rpc", record.name());
+
+ Map<String, String> tags = new HashMap<>();
+ for (MetricsTag tag : record.tags()) {
+ tags.put(tag.name(), tag.value());
+ }
+ assertEquals(String.valueOf(PORT), tags.get("port"));
+ assertEquals(SERVER_NAME, tags.get("serverName"));
+ assertEquals("rpc", tags.get("Context"));
+ // a String valued @Metric method is emitted as a tag, not as a gauge
+ assertEquals("{\"alice\":7}", tags.get("NumOpenConnectionsPerUser"));
+
+ Map<String, Number> values = new HashMap<>();
+ for (AbstractMetric metric : record.metrics()) {
+ values.put(metric.name(), metric.value());
+ }
+ assertEquals(42L, values.get("SentBytes").longValue());
+ assertEquals(24L, values.get("ReceivedBytes").longValue());
+ assertEquals(1L, values.get("RpcSlowCalls").longValue());
+ assertEquals(7, values.get("NumOpenConnections").intValue());
+ assertEquals(3, values.get("CallQueueLength").intValue());
+ assertEquals(5L, values.get("NumDroppedConnections").longValue());
+ assertThat(values).containsKey("RpcProcessingTimeNumOps");
+ }
+
+ @Test
+ public void testProcessingStatsFeedSlowCallDetection() {
+ metrics.addRpcProcessingTime(100);
+ metrics.addRpcProcessingTime(200);
+ metrics.addRpcProcessingTime(300);
+
+ assertEquals(3, metrics.getProcessingSampleCount());
+ assertEquals(200.0, metrics.getProcessingMean(), 0.001);
+ assertTrue(metrics.getProcessingStdDev() > 0.0,
+ "slow call detection needs a positive deviation");
+ }
+
+ @Test
+ public void testConcurrentProcessingTimeAddsAreCountedExactly()
+ throws InterruptedException {
+ int threads = 14;
+ int addsPerThread = 1000;
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ try {
+ for (int t = 0; t < threads; t++) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ for (int i = 0; i < addsPerThread; i++) {
+ metrics.addRpcProcessingTime(1);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ start.countDown();
+ assertTrue(done.await(30, TimeUnit.SECONDS));
+ } finally {
+ pool.shutdown();
+ }
+
+ assertEquals((long) threads * addsPerThread,
+ snapshotMetrics().get("RpcProcessingTimeNumOps").longValue());
+ }
+
+ private Map<String, Number> snapshotMetrics() {
+ MetricsCollectorImpl collector = new MetricsCollectorImpl();
+ source.getMetrics(collector, true);
+ Map<String, Number> values = new HashMap<>();
+ for (MetricsRecord record : collector.getRecords()) {
+ for (AbstractMetric metric : record.metrics()) {
+ values.put(metric.name(), metric.value());
+ }
+ }
+ return values;
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
