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/16 23:49:54 UTC
(ozone) branch master updated: HDDS-16741. Record GrpcMetrics queue and processing times per call (#11554)
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 af7e022ddf8 HDDS-16741. Record GrpcMetrics queue and processing times per call (#11554)
af7e022ddf8 is described below
commit af7e022ddf88c8cbadaa719b9014b4360ac98fcd
Author: Andrey Yarovoy <11...@users.noreply.github.com>
AuthorDate: Fri Oct 16 19:49:48 2026 -0400
HDDS-16741. Record GrpcMetrics queue and processing times per call (#11554)
Generated-by: Claude Code (claude-opus-5-5)
---
.../hadoop/ozone/grpc/metrics/GrpcMetrics.java | 32 +++++-----
.../GrpcMetricsServerRequestInterceptor.java | 38 ++++++++----
.../TestGrpcMetricsServerRequestInterceptor.java | 69 ++++++++++++++++++++++
3 files changed, 111 insertions(+), 28 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 b7e9cacd06b..331e6c1c87f 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
@@ -80,13 +80,13 @@ public class GrpcMetrics implements MetricsSource {
// There should be no getter method to avoid
// exposing internal representation. FindBugs error raised.
- private MutableQuantiles[] grpcQueueTimeMillisQuantiles;
+ private MutableQuantiles[] grpcQueueTimeNanosQuantiles;
private final ConcurrentMutableRate grpcProcessingTime;
// There should be no getter method to avoid
// exposing internal representation. FindBugs error raised.
- private MutableQuantiles[] grpcProcessingTimeMillisQuantiles;
+ private MutableQuantiles[] grpcProcessingTimeNanosQuantiles;
@Metric("Number of active clients connected")
private MutableCounterLong numOpenClientConnections;
@@ -96,25 +96,25 @@ public GrpcMetrics(Configuration conf) {
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);
+ grpcQueueTime = new ConcurrentMutableRate("GrpcQueueTime", "Queue time in nanoseconds", false);
+ grpcProcessingTime = new ConcurrentMutableRate("GrpcProcessingTime", "Processing time in nanoseconds", false);
int[] intervals = conf.getInts(
OzoneConfigKeys.OZONE_GPRC_METRICS_PERCENTILES_INTERVALS_KEY);
grpcQuantileEnable = (intervals.length > 0);
if (grpcQuantileEnable) {
- grpcQueueTimeMillisQuantiles =
+ grpcQueueTimeNanosQuantiles =
new MutableQuantiles[intervals.length];
- grpcProcessingTimeMillisQuantiles =
+ grpcProcessingTimeNanosQuantiles =
new MutableQuantiles[intervals.length];
for (int i = 0; i < intervals.length; i++) {
int interval = intervals[i];
- grpcQueueTimeMillisQuantiles[i] = registry
+ grpcQueueTimeNanosQuantiles[i] = registry
.newQuantiles("grpcQueueTime" + interval
- + "s", "grpc queue time in millisecond", "ops",
+ + "s", "grpc queue time in nanoseconds", "ops",
"latency", interval);
- grpcProcessingTimeMillisQuantiles[i] = registry.newQuantiles(
+ grpcProcessingTimeNanosQuantiles[i] = registry.newQuantiles(
"grpcProcessingTime" + interval + "s",
- "grpc processing time in millisecond",
+ "grpc processing time in nanoseconds",
"ops", "latency", interval);
}
}
@@ -137,8 +137,8 @@ public static synchronized GrpcMetrics create(Configuration conf) {
*/
public void unRegister() {
DefaultMetricsSystem.instance().unregisterSource(SOURCE_NAME);
- MetricUtil.stop(grpcProcessingTimeMillisQuantiles);
- MetricUtil.stop(grpcQueueTimeMillisQuantiles);
+ MetricUtil.stop(grpcProcessingTimeNanosQuantiles);
+ MetricUtil.stop(grpcQueueTimeNanosQuantiles);
}
@Override
@@ -167,10 +167,10 @@ public void incrUnknownMessagesReceived() {
unknownMessagesReceived.incr();
}
- public void addGrpcQueueTime(int queueTime) {
+ public void addGrpcQueueTime(long queueTime) {
grpcQueueTime.add(queueTime);
if (grpcQuantileEnable) {
- for (MutableQuantiles q : grpcQueueTimeMillisQuantiles) {
+ for (MutableQuantiles q : grpcQueueTimeNanosQuantiles) {
if (q != null) {
q.add(queueTime);
}
@@ -178,10 +178,10 @@ public void addGrpcQueueTime(int queueTime) {
}
}
- public void addGrpcProcessingTime(int processingTime) {
+ public void addGrpcProcessingTime(long processingTime) {
grpcProcessingTime.add(processingTime);
if (grpcQuantileEnable) {
- for (MutableQuantiles q : grpcProcessingTimeMillisQuantiles) {
+ for (MutableQuantiles q : grpcProcessingTimeNanosQuantiles) {
if (q != null) {
q.add(processingTime);
}
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetricsServerRequestInterceptor.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetricsServerRequestInterceptor.java
index 55c7f93741e..03f17d65113 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetricsServerRequestInterceptor.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/ozone/grpc/metrics/GrpcMetricsServerRequestInterceptor.java
@@ -17,6 +17,7 @@
package org.apache.hadoop.ozone.grpc.metrics;
+import com.google.common.annotations.VisibleForTesting;
import com.google.protobuf.AbstractMessage;
import com.google.protobuf.Descriptors;
import io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener;
@@ -24,23 +25,30 @@
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
+import java.util.function.LongSupplier;
/**
* Interceptor to gather metrics based on grpc server request.
+ *
+ * <p>One instance serves every call on the server, so the timestamps of a call are kept on that call's listener.
+ * Queue and processing times are recorded in nanoseconds.
*/
public class GrpcMetricsServerRequestInterceptor implements ServerInterceptor {
private static final String CMD_TYPE_FIELD = "cmdType";
private final GrpcMetrics grpcMetrics;
- private long receivedTime;
- private long startTime;
- private long endTime;
+ private final LongSupplier nanoClock;
public GrpcMetricsServerRequestInterceptor(
GrpcMetrics grpcMetrics) {
- super();
+ this(grpcMetrics, System::nanoTime);
+ }
+
+ @VisibleForTesting
+ GrpcMetricsServerRequestInterceptor(GrpcMetrics grpcMetrics, LongSupplier nanoClock) {
this.grpcMetrics = grpcMetrics;
+ this.nanoClock = nanoClock;
}
@Override
@@ -49,15 +57,20 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
ServerCallHandler<ReqT, RespT> serverCallHandler) {
// received time
- receivedTime = System.nanoTime();
+ final long receivedTime = nanoClock.getAsLong();
return new SimpleForwardingServerCallListener<ReqT>(
serverCallHandler.startCall(serverCall, headers)) {
+ // gRPC calls the listener of one call serially, so these need no synchronization
+ private long startTime;
+ private boolean messageReceived;
+
@Override
public void onMessage(ReqT message) {
// start time
- startTime = System.nanoTime();
+ startTime = nanoClock.getAsLong();
+ messageReceived = true;
long messageSize = 0;
if (message instanceof AbstractMessage) {
@@ -81,17 +94,18 @@ public void onMessage(ReqT message) {
@Override
public void onComplete() {
super.onComplete();
+ // a call that completes without a request has no start time
+ if (!messageReceived) {
+ return;
+ }
// end time
- endTime = System.nanoTime();
-
- int queueTime = (int) (startTime - receivedTime);
- int processingTime = (int) (endTime - startTime);
+ long endTime = nanoClock.getAsLong();
// set metrics queue time
- grpcMetrics.addGrpcQueueTime(queueTime);
+ grpcMetrics.addGrpcQueueTime(startTime - receivedTime);
// set metrics processing time
- grpcMetrics.addGrpcProcessingTime(processingTime);
+ grpcMetrics.addGrpcProcessingTime(endTime - startTime);
}
};
}
diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/grpc/metrics/TestGrpcMetricsServerRequestInterceptor.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/grpc/metrics/TestGrpcMetricsServerRequestInterceptor.java
index ac2a4045205..c2f09e063a3 100644
--- a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/grpc/metrics/TestGrpcMetricsServerRequestInterceptor.java
+++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/ozone/grpc/metrics/TestGrpcMetricsServerRequestInterceptor.java
@@ -19,13 +19,17 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.SCMBlockLocationRequest;
import org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.Type;
@@ -33,6 +37,7 @@
import org.apache.hadoop.metrics2.lib.MetricsAnnotations;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
/**
* Tests for {@link GrpcMetricsServerRequestInterceptor}.
@@ -42,6 +47,9 @@
*/
public class TestGrpcMetricsServerRequestInterceptor {
+ private static final SCMBlockLocationRequest REQUEST =
+ SCMBlockLocationRequest.newBuilder().setCmdType(Type.GetScmInfo).setTraceID("trace").build();
+
private GrpcMetrics metrics;
private ServerCall.Listener<Object> delegate;
private ServerCall.Listener<Object> listener;
@@ -98,4 +106,65 @@ public void testNonProtobufMessageIsCountedAsUnknown() {
assertEquals(0, metrics.getReceivedBytes());
verify(delegate).onMessage(message);
}
+
+ @Test
+ public void testInterleavedCallsRecordTheirOwnTimes() {
+ GrpcMetrics mockMetrics = mock(GrpcMetrics.class);
+ AtomicLong now = new AtomicLong();
+ GrpcMetricsServerRequestInterceptor interceptor = new GrpcMetricsServerRequestInterceptor(mockMetrics, now::get);
+
+ now.set(100);
+ ServerCall.Listener<Object> first = startCall(interceptor);
+ now.set(200);
+ ServerCall.Listener<Object> second = startCall(interceptor);
+ now.set(250);
+ second.onMessage(REQUEST);
+ now.set(300);
+ first.onMessage(REQUEST);
+ now.set(400);
+ second.onComplete();
+ now.set(1000);
+ first.onComplete();
+
+ InOrder inOrder = inOrder(mockMetrics);
+ // second call: received at 200, started at 250, completed at 400
+ inOrder.verify(mockMetrics).addGrpcQueueTime(50);
+ inOrder.verify(mockMetrics).addGrpcProcessingTime(150);
+ // first call: received at 100, started at 300, completed at 1000
+ inOrder.verify(mockMetrics).addGrpcQueueTime(200);
+ inOrder.verify(mockMetrics).addGrpcProcessingTime(700);
+ }
+
+ @Test
+ public void testTimesAboveIntRangeAreNotTruncated() {
+ GrpcMetrics mockMetrics = mock(GrpcMetrics.class);
+ AtomicLong now = new AtomicLong();
+ ServerCall.Listener<Object> call = startCall(new GrpcMetricsServerRequestInterceptor(mockMetrics, now::get));
+
+ now.set(3_000_000_000L);
+ call.onMessage(REQUEST);
+ now.set(8_000_000_000L);
+ call.onComplete();
+
+ verify(mockMetrics).addGrpcQueueTime(3_000_000_000L);
+ verify(mockMetrics).addGrpcProcessingTime(5_000_000_000L);
+ }
+
+ @Test
+ public void testCallWithoutMessageRecordsNoTimes() {
+ GrpcMetrics mockMetrics = mock(GrpcMetrics.class);
+ ServerCall.Listener<Object> call = startCall(new GrpcMetricsServerRequestInterceptor(mockMetrics));
+
+ call.onComplete();
+
+ verify(mockMetrics, never()).addGrpcQueueTime(anyLong());
+ verify(mockMetrics, never()).addGrpcProcessingTime(anyLong());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static ServerCall.Listener<Object> startCall(GrpcMetricsServerRequestInterceptor interceptor) {
+ ServerCallHandler<Object, Object> handler = mock(ServerCallHandler.class);
+ when(handler.startCall(any(), any())).thenReturn(mock(ServerCall.Listener.class));
+ return interceptor.interceptCall(mock(ServerCall.class), new Metadata(), handler);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org