From e50ed9d51ee715b16b1292cb18f1f0a5050df3d9 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Thu, 22 Oct 2026 07:31:11 +0000 Subject: [PATCH] feat: add rpc.client.call.duration metric for gRPC calls --- .../cloud/storage/GrpcStorageOptions.java | 15 +- .../StorageGrpcMetricsInterceptor.java | 163 +++++++++ .../StorageGrpcMetricsInterceptorTest.java | 325 ++++++++++++++++++ 3 files changed, 502 insertions(+), 1 deletion(-) create mode 100644 java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageGrpcMetricsInterceptor.java create mode 100644 java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageGrpcMetricsInterceptorTest.java diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java index ce0ec9fbc487..9cd814a15e3c 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java @@ -331,7 +331,20 @@ private Tuple> resolveSettingsAndOpts() throw Collections.singletonList(InstantiatingGrpcChannelProvider.HardBoundTokenTypes.ALTS)); } - if (!NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider)) { + StorageClientMetrics clientMetrics = getStorageClientMetrics(); + if (clientMetrics != null) { + StorageGrpcMetricsInterceptor metricsInterceptor = + new StorageGrpcMetricsInterceptor(clientMetrics, getClock(), endpoint); + GrpcInterceptorProvider combinedProvider = + NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) + ? () -> ImmutableList.of(metricsInterceptor) + : () -> + ImmutableList.builder() + .addAll(grpcInterceptorProvider.getInterceptors()) + .add(metricsInterceptor) + .build(); + channelProviderBuilder.setInterceptorProvider(combinedProvider); + } else if (!NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider)) { channelProviderBuilder.setInterceptorProvider(grpcInterceptorProvider); } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageGrpcMetricsInterceptor.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageGrpcMetricsInterceptor.java new file mode 100644 index 000000000000..a09d6c37b06e --- /dev/null +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageGrpcMetricsInterceptor.java @@ -0,0 +1,163 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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 com.google.cloud.storage; + +import com.google.api.core.ApiClock; +import io.grpc.Attributes; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ForwardingClientCall.SimpleForwardingClientCall; +import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener; +import io.grpc.Grpc; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.context.Context; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.concurrent.atomic.AtomicBoolean; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** gRPC client interceptor that records call metrics such as rpc.client.call.duration. */ +final class StorageGrpcMetricsInterceptor implements ClientInterceptor { + + private final StorageClientMetrics metrics; + private final ApiClock clock; + @Nullable private final String defaultEndpoint; + + StorageGrpcMetricsInterceptor( + StorageClientMetrics metrics, ApiClock clock, @Nullable String defaultEndpoint) { + this.metrics = metrics; + this.clock = clock; + this.defaultEndpoint = defaultEndpoint; + } + + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + if (metrics == null || metrics.getRpcClientCallDuration() == null) { + return next.newCall(method, callOptions); + } + + String shortMethod = extractShortMethodName(method.getFullMethodName()); + String initialServerAddress = + StorageOptions.extractServerAddress( + defaultEndpoint != null ? defaultEndpoint : next.authority()); + final String serverAddress = + initialServerAddress != null + ? initialServerAddress + : StorageOptions.extractServerAddress(next.authority()); + + StorageMetricsContext metricsCtx = StorageMetricsContext.from(callOptions); + Context otelContext = metricsCtx != null ? metricsCtx.getOtelContext() : Context.current(); + + return new SimpleForwardingClientCall(next.newCall(method, callOptions)) { + @Override + public void start(Listener responseListener, Metadata headers) { + long startTimeNanos = clock.nanoTime(); + AtomicBoolean callRecorded = new AtomicBoolean(false); + + super.start( + new SimpleForwardingClientCallListener(responseListener) { + @Override + public void onClose(Status status, Metadata trailers) { + try { + if (callRecorded.compareAndSet(false, true) + && metrics.getRpcClientCallDuration() != null) { + double durationSeconds = + Math.max(0.0, (clock.nanoTime() - startTimeNanos) / 1_000_000_000.0); + String networkPath = detectNetworkPath(getAttributes(), defaultEndpoint); + + AttributesBuilder attrBuilder = + metrics + .clientAttributesBuilder() + .put(StorageClientMetrics.KEY_RPC_SYSTEM_NAME, "grpc") + .put(StorageClientMetrics.KEY_RPC_METHOD, shortMethod) + .put( + StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE, + status.getCode().name()) + .put(StorageClientMetrics.KEY_NETWORK_PATH, networkPath) + .put( + StorageClientMetrics.KEY_SERVER_ADDRESS, + serverAddress != null ? serverAddress : "") + .put( + StorageClientMetrics.KEY_ERROR_TYPE, + StorageErrorTypeMapper.mapGrpcStatus(status)); + + metrics + .getRpcClientCallDuration() + .record(durationSeconds, attrBuilder.build(), otelContext); + } + } finally { + super.onClose(status, trailers); + } + } + }, + headers); + } + }; + } + + static String extractShortMethodName(@Nullable String fullMethodName) { + if (fullMethodName == null || fullMethodName.isEmpty()) { + return "UNKNOWN"; + } + int idx = fullMethodName.lastIndexOf('/'); + if (idx >= 0 && idx < fullMethodName.length() - 1) { + return fullMethodName.substring(idx + 1); + } + return fullMethodName; + } + + static String detectNetworkPath( + @Nullable Attributes callAttributes, @Nullable String defaultEndpoint) { + SocketAddress remoteAddr = + callAttributes != null ? callAttributes.get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR) : null; + if (remoteAddr instanceof InetSocketAddress) { + InetAddress inetAddr = ((InetSocketAddress) remoteAddr).getAddress(); + if (inetAddr != null) { + return isDirectPathIp(inetAddr) + ? StorageClientMetrics.NETWORK_PATH_DIRECTPATH + : StorageClientMetrics.NETWORK_PATH_CLOUDPATH; + } + } + if (defaultEndpoint != null && defaultEndpoint.trim().startsWith("google-c2p:///")) { + return StorageClientMetrics.NETWORK_PATH_DIRECTPATH; + } + return StorageClientMetrics.NETWORK_PATH_UNKNOWN; + } + + static boolean isDirectPathIp(InetAddress addr) { + byte[] b = addr.getAddress(); + if (b.length == 4) { + return (b[0] & 0xFF) == 34 && (b[1] & 0xFF) == 126 && ((b[2] & 0xFF) & 0xC0) == 0; + } + if (b.length == 16) { + return (b[0] & 0xFF) == 0x20 + && (b[1] & 0xFF) == 0x01 + && (b[2] & 0xFF) == 0x48 + && (b[3] & 0xFF) == 0x60 + && (b[4] & 0xFF) == 0x80 + && ((b[5] & 0xFF) & 0xC0) == 0x40; + } + return false; + } +} diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageGrpcMetricsInterceptorTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageGrpcMetricsInterceptorTest.java new file mode 100644 index 000000000000..6bc17c0df9da --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageGrpcMetricsInterceptorTest.java @@ -0,0 +1,325 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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 com.google.cloud.storage; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.api.core.ApiClock; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.Grpc; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.HistogramPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.util.Collection; +import java.util.concurrent.atomic.AtomicLong; +import org.junit.Before; +import org.junit.Test; + +public final class StorageGrpcMetricsInterceptorTest { + + private static final MethodDescriptor READ_OBJECT_METHOD = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.SERVER_STREAMING) + .setFullMethodName("google.storage.v2.Storage/ReadObject") + .setRequestMarshaller(new NoopMarshaller()) + .setResponseMarshaller(new NoopMarshaller()) + .build(); + + private InMemoryMetricReader reader; + private SdkMeterProvider meterProvider; + private FakeClock clock; + + @Before + public void setUp() { + reader = InMemoryMetricReader.create(); + meterProvider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + clock = new FakeClock(); + } + + @Test + public void extractShortMethodName_variousInputs() { + assertThat( + StorageGrpcMetricsInterceptor.extractShortMethodName( + "google.storage.v2.Storage/ReadObject")) + .isEqualTo("ReadObject"); + assertThat(StorageGrpcMetricsInterceptor.extractShortMethodName("ReadObject")) + .isEqualTo("ReadObject"); + assertThat(StorageGrpcMetricsInterceptor.extractShortMethodName("a/b/c")).isEqualTo("c"); + assertThat(StorageGrpcMetricsInterceptor.extractShortMethodName(null)).isEqualTo("UNKNOWN"); + assertThat(StorageGrpcMetricsInterceptor.extractShortMethodName("")).isEqualTo("UNKNOWN"); + assertThat(StorageGrpcMetricsInterceptor.extractShortMethodName("/")).isEqualTo("/"); + } + + @Test + public void isDirectPathIp_ipv4AndIpv6() throws Exception { + assertThat(StorageGrpcMetricsInterceptor.isDirectPathIp(InetAddress.getByName("34.126.0.1"))) + .isTrue(); + assertThat(StorageGrpcMetricsInterceptor.isDirectPathIp(InetAddress.getByName("34.126.63.254"))) + .isTrue(); + assertThat(StorageGrpcMetricsInterceptor.isDirectPathIp(InetAddress.getByName("34.126.64.1"))) + .isFalse(); + assertThat(StorageGrpcMetricsInterceptor.isDirectPathIp(InetAddress.getByName("142.250.1.1"))) + .isFalse(); + + assertThat( + StorageGrpcMetricsInterceptor.isDirectPathIp( + InetAddress.getByName("2001:4860:8040:0000:0000:0000:0000:0001"))) + .isTrue(); + assertThat( + StorageGrpcMetricsInterceptor.isDirectPathIp( + InetAddress.getByName("2001:4860:807f:ffff:ffff:ffff:ffff:ffff"))) + .isTrue(); + assertThat( + StorageGrpcMetricsInterceptor.isDirectPathIp( + InetAddress.getByName("2001:4860:8080:0000:0000:0000:0000:0001"))) + .isFalse(); + assertThat( + StorageGrpcMetricsInterceptor.isDirectPathIp( + InetAddress.getByName("2607:f8b0:4005:805::200e"))) + .isFalse(); + } + + @Test + public void detectNetworkPath_cases() throws Exception { + io.grpc.Attributes directPathAttrs = + io.grpc.Attributes.newBuilder() + .set( + Grpc.TRANSPORT_ATTR_REMOTE_ADDR, + new InetSocketAddress(InetAddress.getByName("34.126.1.2"), 443)) + .build(); + assertThat(StorageGrpcMetricsInterceptor.detectNetworkPath(directPathAttrs, null)) + .isEqualTo(StorageClientMetrics.NETWORK_PATH_DIRECTPATH); + + io.grpc.Attributes cloudPathAttrs = + io.grpc.Attributes.newBuilder() + .set( + Grpc.TRANSPORT_ATTR_REMOTE_ADDR, + new InetSocketAddress(InetAddress.getByName("142.250.1.1"), 443)) + .build(); + assertThat(StorageGrpcMetricsInterceptor.detectNetworkPath(cloudPathAttrs, null)) + .isEqualTo(StorageClientMetrics.NETWORK_PATH_CLOUDPATH); + + assertThat( + StorageGrpcMetricsInterceptor.detectNetworkPath( + null, "google-c2p:///storage.googleapis.com")) + .isEqualTo(StorageClientMetrics.NETWORK_PATH_DIRECTPATH); + assertThat( + StorageGrpcMetricsInterceptor.detectNetworkPath(null, "https://storage.googleapis.com")) + .isEqualTo(StorageClientMetrics.NETWORK_PATH_UNKNOWN); + assertThat(StorageGrpcMetricsInterceptor.detectNetworkPath(null, null)) + .isEqualTo(StorageClientMetrics.NETWORK_PATH_UNKNOWN); + } + + @Test + public void interceptCall_unaryCallSuccess_recordsMetricWithErrorTypeOk() throws Exception { + StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, true, false); + StorageGrpcMetricsInterceptor interceptor = + new StorageGrpcMetricsInterceptor(metrics, clock, "https://storage.googleapis.com"); + + io.grpc.Attributes callAttrs = + io.grpc.Attributes.newBuilder() + .set( + Grpc.TRANSPORT_ATTR_REMOTE_ADDR, + new InetSocketAddress(InetAddress.getByName("142.250.1.1"), 443)) + .build(); + TestClientCall testCall = new TestClientCall<>(callAttrs); + TestChannel testChannel = new TestChannel(testCall, "storage.googleapis.com:443"); + + ClientCall call = + interceptor.interceptCall(READ_OBJECT_METHOD, CallOptions.DEFAULT, testChannel); + TestListener listener = new TestListener<>(); + call.start(listener, new Metadata()); + + clock.advanceNanos(500_000_000L); // 0.5s + testCall.close(Status.OK, new Metadata()); + + Collection metricData = reader.collectAllMetrics(); + MetricData rpcDurationData = + metricData.stream() + .filter(m -> m.getName().equals(StorageClientMetrics.METRIC_RPC_CLIENT_CALL_DURATION)) + .findFirst() + .orElse(null); + + assertThat(rpcDurationData).isNotNull(); + assertThat(rpcDurationData.getHistogramData().getPoints()).hasSize(1); + HistogramPointData point = rpcDurationData.getHistogramData().getPoints().iterator().next(); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isWithin(1e-6).of(0.5); + + Attributes attrs = point.getAttributes(); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)).isEqualTo("grpc"); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_METHOD)).isEqualTo("ReadObject"); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE)).isEqualTo("OK"); + assertThat(attrs.get(StorageClientMetrics.KEY_NETWORK_PATH)) + .isEqualTo(StorageClientMetrics.NETWORK_PATH_CLOUDPATH); + assertThat(attrs.get(StorageClientMetrics.KEY_SERVER_ADDRESS)) + .isEqualTo("storage.googleapis.com"); + assertThat(attrs.get(StorageClientMetrics.KEY_ERROR_TYPE)).isEqualTo("OK"); + } + + @Test + public void interceptCall_callFails_recordsErrorType() throws Exception { + StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, true, false); + StorageGrpcMetricsInterceptor interceptor = + new StorageGrpcMetricsInterceptor(metrics, clock, "storage.googleapis.com"); + + TestClientCall testCall = new TestClientCall<>(io.grpc.Attributes.EMPTY); + TestChannel testChannel = new TestChannel(testCall, "storage.googleapis.com"); + + ClientCall call = + interceptor.interceptCall(READ_OBJECT_METHOD, CallOptions.DEFAULT, testChannel); + call.start(new TestListener<>(), new Metadata()); + + clock.advanceNanos(100_000_000L); // 0.1s + testCall.close(Status.NOT_FOUND, new Metadata()); + + Collection metricData = reader.collectAllMetrics(); + MetricData rpcDurationData = + metricData.stream() + .filter(m -> m.getName().equals(StorageClientMetrics.METRIC_RPC_CLIENT_CALL_DURATION)) + .findFirst() + .orElse(null); + + assertThat(rpcDurationData).isNotNull(); + HistogramPointData point = rpcDurationData.getHistogramData().getPoints().iterator().next(); + assertThat(point.getAttributes().get(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE)) + .isEqualTo("NOT_FOUND"); + assertThat(point.getAttributes().get(StorageClientMetrics.KEY_ERROR_TYPE)) + .isEqualTo("NOT_FOUND"); + assertThat(point.getAttributes().get(StorageClientMetrics.KEY_SERVER_ADDRESS)) + .isEqualTo("storage.googleapis.com"); + } + + @Test + public void interceptCall_disabledMetrics_doesNotRecord() { + StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, false, false); + StorageGrpcMetricsInterceptor interceptor = + new StorageGrpcMetricsInterceptor(metrics, clock, "storage.googleapis.com"); + + TestClientCall testCall = new TestClientCall<>(io.grpc.Attributes.EMPTY); + TestChannel testChannel = new TestChannel(testCall, "storage.googleapis.com"); + + ClientCall call = + interceptor.interceptCall(READ_OBJECT_METHOD, CallOptions.DEFAULT, testChannel); + call.start(new TestListener<>(), new Metadata()); + testCall.close(Status.OK, new Metadata()); + + assertThat(reader.collectAllMetrics()).isEmpty(); + } + + private static final class TestClientCall extends ClientCall { + private Listener listener; + private final io.grpc.Attributes attributes; + + TestClientCall(io.grpc.Attributes attributes) { + this.attributes = attributes; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + this.listener = responseListener; + } + + @Override + public void request(int numMessages) {} + + @Override + public void cancel(String message, Throwable cause) {} + + @Override + public void halfClose() {} + + @Override + public void sendMessage(ReqT message) {} + + @Override + public io.grpc.Attributes getAttributes() { + return attributes; + } + + void close(Status status, Metadata trailers) { + if (listener != null) { + listener.onClose(status, trailers); + } + } + } + + private static final class TestChannel extends Channel { + private final TestClientCall call; + private final String authority; + + TestChannel(TestClientCall call, String authority) { + this.call = call; + this.authority = authority; + } + + @SuppressWarnings("unchecked") + @Override + public ClientCall newCall( + MethodDescriptor methodDescriptor, CallOptions callOptions) { + return (ClientCall) call; + } + + @Override + public String authority() { + return authority; + } + } + + private static final class TestListener extends ClientCall.Listener {} + + private static final class NoopMarshaller implements MethodDescriptor.Marshaller { + @Override + public InputStream stream(Void value) { + return new ByteArrayInputStream(new byte[0]); + } + + @Override + public Void parse(InputStream stream) { + return null; + } + } + + private static final class FakeClock implements ApiClock { + private final AtomicLong nanos = new AtomicLong(1_000_000_000L); + + @Override + public long nanoTime() { + return nanos.get(); + } + + @Override + public long millisTime() { + return nanos.get() / 1_000_000L; + } + + void advanceNanos(long amount) { + nanos.addAndGet(amount); + } + } +}