You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by we...@apache.org on 2026/10/14 05:31:00 UTC
(ozone) branch master updated: HDDS-16740. Convert GrpcMetrics latency counters to lock-free ConcurrentMutableRate (#11498)
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 5b2dd0ff08c HDDS-16740. Convert GrpcMetrics latency counters to lock-free ConcurrentMutableRate (#11498)
5b2dd0ff08c is described below
commit 5b2dd0ff08c20808563ed1d06fb1a32c8f0dabe5
Author: Andrey Yarovoy <11...@users.noreply.github.com>
AuthorDate: Wed Oct 14 01:30:54 2026 -0400
HDDS-16740. Convert GrpcMetrics latency counters to lock-free ConcurrentMutableRate (#11498)
Generated-by: Claude Code (Claude Opus 5.5)
---
.../hadoop/ozone/grpc/metrics/GrpcMetrics.java | 26 +++-
.../hadoop/ozone/grpc/metrics/TestGrpcMetrics.java | 164 +++++++++++++++++++++
2 files changed, 182 insertions(+), 8 deletions(-)
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetrics.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetrics.java
index 5e9921e507c..bc39d9bf6ca 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetrics.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetrics.java
@@ -29,15 +29,20 @@
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.OzoneConfigKeys;
import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.util.ConcurrentMutableRate;
import org.apache.hadoop.ozone.util.MetricUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Class which maintains metrics related to using GRPC.
+ *
+ * The two latency rates are kept outside the {@link MetricsRegistry} as
+ * lock-free {@link ConcurrentMutableRate}s, because they are updated on every
+ * gRPC call and {@code MutableStat.add} is synchronized. {@link #getMetrics}
+ * snapshots them next to the registry, keeping the emitted metric names unchanged.
*/
@Metrics(about = "GRPC Metrics", context = OzoneConsts.OZONE)
public class GrpcMetrics implements MetricsSource {
@@ -70,15 +75,13 @@ public class GrpcMetrics implements MetricsSource {
@Metric("Number of unknown messages received")
private MutableCounterLong unknownMessagesReceived;
- @Metric("Queue time")
- private MutableRate grpcQueueTime;
+ private final ConcurrentMutableRate grpcQueueTime;
// There should be no getter method to avoid
// exposing internal representation. FindBugs error raised.
private MutableQuantiles[] grpcQueueTimeMillisQuantiles;
- @Metric("Processing time")
- private MutableRate grpcProcessingTime;
+ private final ConcurrentMutableRate grpcProcessingTime;
// There should be no getter method to avoid
// exposing internal representation. FindBugs error raised.
@@ -90,6 +93,10 @@ public class GrpcMetrics implements MetricsSource {
public GrpcMetrics(Configuration conf) {
this.registry = new MetricsRegistry("grpc");
this.requestType = "NoRequest";
+ // Names match what MutableMetricsFactory derived from the field names, so
+ // the emitted metrics are unchanged
+ grpcQueueTime = new ConcurrentMutableRate("GrpcQueueTime", "Queue time", false);
+ grpcProcessingTime = new ConcurrentMutableRate("GrpcProcessingTime", "Processing time", false);
int[] intervals = conf.getInts(
OzoneConfigKeys.OZONE_GPRC_METRICS_PERCENTILES_INTERVALS_KEY);
grpcQuantileEnable = (intervals.length > 0);
@@ -135,7 +142,10 @@ public void unRegister() {
@Override
public synchronized void getMetrics(MetricsCollector collector, boolean all) {
- registry.snapshot(collector.addRecord(registry.info()), all);
+ MetricsRecordBuilder builder = collector.addRecord(registry.info());
+ registry.snapshot(builder, all);
+ grpcQueueTime.snapshot(builder, all);
+ grpcProcessingTime.snapshot(builder, all);
MetricsRecordBuilder recordBuilder = collector.addRecord(SOURCE_NAME);
recordBuilder.tag(LATEST_REQUEST_TYPE, requestType);
}
@@ -202,11 +212,11 @@ public long getUnknownMessagesReceived() {
return unknownMessagesReceived.value();
}
- MutableRate getGrpcQueueTime() {
+ ConcurrentMutableRate getGrpcQueueTime() {
return grpcQueueTime;
}
- MutableRate getGrpcProcessingTime() {
+ ConcurrentMutableRate getGrpcProcessingTime() {
return grpcProcessingTime;
}
diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/grpc/metrics/TestGrpcMetrics.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/grpc/metrics/TestGrpcMetrics.java
new file mode 100644
index 00000000000..6381962e71a
--- /dev/null
+++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/grpc/metrics/TestGrpcMetrics.java
@@ -0,0 +1,164 @@
+/*
+ * 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.ozone.grpc.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 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.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsSource;
+import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl;
+import org.apache.hadoop.metrics2.lib.MetricsAnnotations;
+import org.apache.hadoop.ozone.OzoneConfigKeys;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link GrpcMetrics}.
+ *
+ * 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 that they stay in the same record as the registry-backed metrics.
+ */
+public class TestGrpcMetrics {
+
+ private GrpcMetrics metrics;
+ private MetricsSource source;
+
+ @BeforeEach
+ public void setUp() {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.set(OzoneConfigKeys.OZONE_GPRC_METRICS_PERCENTILES_INTERVALS_KEY, "60");
+ metrics = new GrpcMetrics(conf);
+ // Same reflection DefaultMetricsSystem.register() performs: it populates the
+ // @Metric counter fields into the registry
+ source = MetricsAnnotations.makeSource(metrics);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ metrics.unRegister();
+ }
+
+ @Test
+ public void testGrpcMetricsIsUsedAsItsOwnSource() {
+ assertSame(metrics, source);
+ }
+
+ @Test
+ public void testLatencyRateNamesAreUnchanged() {
+ metrics.addGrpcQueueTime(10);
+ metrics.addGrpcQueueTime(20);
+ metrics.addGrpcProcessingTime(100);
+ metrics.addGrpcProcessingTime(300);
+
+ Map<String, Number> values = snapshotMetrics();
+
+ assertEquals(2L, values.get("GrpcQueueTimeNumOps").longValue());
+ assertEquals(15.0, values.get("GrpcQueueTimeAvgTime").doubleValue(), 0.001);
+ assertEquals(2L, values.get("GrpcProcessingTimeNumOps").longValue());
+ assertEquals(200.0, values.get("GrpcProcessingTimeAvgTime").doubleValue(), 0.001);
+
+ // the rates are not extended, matching @Metric.always() defaulting to false
+ assertThat(values).doesNotContainKeys("GrpcQueueTimeStdevTime", "GrpcProcessingTimeStdevTime");
+ }
+
+ @Test
+ public void testRatesAreInTheRegistryRecord() {
+ metrics.incrSentBytes(42);
+ metrics.incrReceivedBytes(24);
+ metrics.inrcNumOpenClientConnections();
+ metrics.addGrpcProcessingTime(5);
+ metrics.setRequestType("CreateKey");
+
+ MetricsCollectorImpl collector = new MetricsCollectorImpl();
+ source.getMetrics(collector, true);
+ assertEquals(2, collector.getRecords().size());
+
+ MetricsRecord record = collector.getRecords().get(0);
+ assertEquals("grpc", record.name());
+ 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("NumOpenClientConnections").longValue());
+ assertEquals(1L, values.get("GrpcProcessingTimeNumOps").longValue());
+ assertThat(values).containsKeys("GrpcQueueTimeNumOps", "GrpcProcessingTime60sNumOps");
+
+ MetricsRecord tagRecord = collector.getRecords().get(1);
+ assertEquals(GrpcMetrics.class.getSimpleName(), tagRecord.name());
+ assertThat(tagRecord.tags()).anyMatch(tag ->
+ tag.name().equals("LatestRequestType") && tag.value().equals("CreateKey"));
+ }
+
+ @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.addGrpcProcessingTime(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("GrpcProcessingTimeNumOps").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: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org