Repository navigation
feat(storage): add rpc.client.call.duration metric for gRPC calls - #14743
nidhiii-27 wants to merge 1 commit into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces StorageGrpcMetricsInterceptor, a new gRPC client interceptor designed to record call metrics such as RPC duration, and integrates it into GrpcStorageOptions. It also adds comprehensive unit tests to verify its behavior. The review feedback is highly constructive, focusing on preventing a potential NullPointerException when combining interceptors, simplifying redundant server address resolution, and optimizing performance on the hot path by precomputing the DirectPath endpoint status in the constructor rather than performing string operations on every RPC completion.
| GrpcInterceptorProvider combinedProvider = | ||
| NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) | ||
| ? () -> ImmutableList.of(metricsInterceptor) | ||
| : () -> | ||
| ImmutableList.<ClientInterceptor>builder() | ||
| .addAll(grpcInterceptorProvider.getInterceptors()) | ||
| .add(metricsInterceptor) | ||
| .build(); |
There was a problem hiding this comment.
If grpcInterceptorProvider is null, calling grpcInterceptorProvider.getInterceptors() will throw a NullPointerException when the interceptor provider is invoked. We should add a null check to safely handle this case.
| GrpcInterceptorProvider combinedProvider = | |
| NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) | |
| ? () -> ImmutableList.of(metricsInterceptor) | |
| : () -> | |
| ImmutableList.<ClientInterceptor>builder() | |
| .addAll(grpcInterceptorProvider.getInterceptors()) | |
| .add(metricsInterceptor) | |
| .build(); | |
| GrpcInterceptorProvider combinedProvider = | |
| grpcInterceptorProvider == null | |
| || NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) | |
| ? () -> ImmutableList.of(metricsInterceptor) | |
| : () -> | |
| ImmutableList.<ClientInterceptor>builder() | |
| .addAll(grpcInterceptorProvider.getInterceptors()) | |
| .add(metricsInterceptor) | |
| .build(); |
| String initialServerAddress = | ||
| StorageOptions.extractServerAddress( | ||
| defaultEndpoint != null ? defaultEndpoint : next.authority()); | ||
| final String serverAddress = | ||
| initialServerAddress != null | ||
| ? initialServerAddress | ||
| : StorageOptions.extractServerAddress(next.authority()); |
There was a problem hiding this comment.
When defaultEndpoint is null, StorageOptions.extractServerAddress(next.authority()) is called twice if it returns null. We can simplify this logic to avoid redundant calls and improve readability.
String targetEndpoint = defaultEndpoint != null ? defaultEndpoint : next.authority();
String resolvedAddress = StorageOptions.extractServerAddress(targetEndpoint);
final String serverAddress =
resolvedAddress != null
? resolvedAddress
: (defaultEndpoint != null ? StorageOptions.extractServerAddress(next.authority()) : null);| @Nullable private final String defaultEndpoint; | ||
|
|
||
| StorageGrpcMetricsInterceptor( | ||
| StorageClientMetrics metrics, ApiClock clock, @Nullable String defaultEndpoint) { | ||
| this.metrics = metrics; | ||
| this.clock = clock; | ||
| this.defaultEndpoint = defaultEndpoint; | ||
| } |
There was a problem hiding this comment.
To avoid the overhead of string trimming and prefix matching on every single RPC completion, we can precompute whether the defaultEndpoint is a DirectPath endpoint once in the constructor.
@Nullable private final String defaultEndpoint;
private final boolean isDirectPathEndpoint;
StorageGrpcMetricsInterceptor(
StorageClientMetrics metrics, ApiClock clock, @Nullable String defaultEndpoint) {
this.metrics = metrics;
this.clock = clock;
this.defaultEndpoint = defaultEndpoint;
this.isDirectPathEndpoint =
defaultEndpoint != null && defaultEndpoint.trim().startsWith("google-c2p:///");
}| double durationSeconds = | ||
| Math.max(0.0, (clock.nanoTime() - startTimeNanos) / 1_000_000_000.0); | ||
| String networkPath = detectNetworkPath(getAttributes(), defaultEndpoint); |
There was a problem hiding this comment.
Update the call to detectNetworkPath to pass the precomputed isDirectPathEndpoint boolean instead of the raw defaultEndpoint string.
| double durationSeconds = | |
| Math.max(0.0, (clock.nanoTime() - startTimeNanos) / 1_000_000_000.0); | |
| String networkPath = detectNetworkPath(getAttributes(), defaultEndpoint); | |
| double durationSeconds = | |
| Math.max(0.0, (clock.nanoTime() - startTimeNanos) / 1_000_000_000.0); | |
| String networkPath = detectNetworkPath(getAttributes(), isDirectPathEndpoint); |
| 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; | ||
| } |
There was a problem hiding this comment.
We can update detectNetworkPath to accept the precomputed isDirectPathEndpoint boolean instead of the raw defaultEndpoint string, avoiding string operations on the hot path.
| 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 String detectNetworkPath( | |
| @Nullable Attributes callAttributes, boolean isDirectPathEndpoint) { | |
| 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 (isDirectPathEndpoint) { | |
| return StorageClientMetrics.NETWORK_PATH_DIRECTPATH; | |
| } | |
| return StorageClientMetrics.NETWORK_PATH_UNKNOWN; | |
| } |
| 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); |
There was a problem hiding this comment.
Update the tests to match the new signature of detectNetworkPath which now accepts a boolean instead of a string.
| 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); | |
| assertThat(StorageGrpcMetricsInterceptor.detectNetworkPath(directPathAttrs, false)) | |
| .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, false)) | |
| .isEqualTo(StorageClientMetrics.NETWORK_PATH_CLOUDPATH); | |
| assertThat(StorageGrpcMetricsInterceptor.detectNetworkPath(null, true)) | |
| .isEqualTo(StorageClientMetrics.NETWORK_PATH_DIRECTPATH); | |
| assertThat(StorageGrpcMetricsInterceptor.detectNetworkPath(null, false)) | |
| .isEqualTo(StorageClientMetrics.NETWORK_PATH_UNKNOWN); |
There was a problem hiding this comment.
Code Review
This pull request introduces StorageGrpcMetricsInterceptor to record gRPC client call metrics (such as duration, status codes, and network paths) and integrates it into GrpcStorageOptions. It also adds comprehensive unit tests to verify the interceptor's behavior. The review feedback highlights two key improvement opportunities: adding a null check for grpcInterceptorProvider to prevent a potential NullPointerException when combining providers, and simplifying the redundant logic used to resolve the serverAddress.
| GrpcInterceptorProvider combinedProvider = | ||
| NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) | ||
| ? () -> ImmutableList.of(metricsInterceptor) | ||
| : () -> | ||
| ImmutableList.<ClientInterceptor>builder() | ||
| .addAll(grpcInterceptorProvider.getInterceptors()) | ||
| .add(metricsInterceptor) | ||
| .build(); |
There was a problem hiding this comment.
If grpcInterceptorProvider is null, NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) evaluates to false. This leads to executing the else branch of the ternary operator, where calling grpcInterceptorProvider.getInterceptors() will throw a NullPointerException. Adding a null check for grpcInterceptorProvider prevents this potential runtime crash.
| GrpcInterceptorProvider combinedProvider = | |
| NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) | |
| ? () -> ImmutableList.of(metricsInterceptor) | |
| : () -> | |
| ImmutableList.<ClientInterceptor>builder() | |
| .addAll(grpcInterceptorProvider.getInterceptors()) | |
| .add(metricsInterceptor) | |
| .build(); | |
| GrpcInterceptorProvider combinedProvider = | |
| grpcInterceptorProvider == null | |
| || NoopGrpcInterceptorProvider.INSTANCE.equals(grpcInterceptorProvider) | |
| ? () -> ImmutableList.of(metricsInterceptor) | |
| : () -> | |
| ImmutableList.<ClientInterceptor>builder() | |
| .addAll(grpcInterceptorProvider.getInterceptors()) | |
| .add(metricsInterceptor) | |
| .build(); |
| String initialServerAddress = | ||
| StorageOptions.extractServerAddress( | ||
| defaultEndpoint != null ? defaultEndpoint : next.authority()); | ||
| final String serverAddress = | ||
| initialServerAddress != null | ||
| ? initialServerAddress | ||
| : StorageOptions.extractServerAddress(next.authority()); | ||
|
|
There was a problem hiding this comment.
The current logic for resolving serverAddress is redundant and hard to read. If defaultEndpoint is null, StorageOptions.extractServerAddress(next.authority()) is evaluated twice. Simplifying this logic avoids redundant method calls and improves readability.
| String initialServerAddress = | |
| StorageOptions.extractServerAddress( | |
| defaultEndpoint != null ? defaultEndpoint : next.authority()); | |
| final String serverAddress = | |
| initialServerAddress != null | |
| ? initialServerAddress | |
| : StorageOptions.extractServerAddress(next.authority()); | |
| String shortMethod = extractShortMethodName(method.getFullMethodName()); | |
| String serverAddress = null; | |
| if (defaultEndpoint != null) { | |
| serverAddress = StorageOptions.extractServerAddress(defaultEndpoint); | |
| } | |
| if (serverAddress == null) { | |
| serverAddress = StorageOptions.extractServerAddress(next.authority()); | |
| } |
Instruments gRPC storage calls to record
rpc.client.call.durationOpenTelemetry histogram metrics.StorageGrpcMetricsInterceptorto measure per-RPC duration and attachrpc.system.name,rpc.method,rpc.response.status_code,network.path,server.address, anderror.typeattributes.StorageGrpcMetricsInterceptorintoGrpcStorageOptionsand initializesrpc.client.call.durationinStorageClientMetrics.StorageGrpcMetricsInterceptorTestandStorageClientMetricsTest.Stacked on #14729.