Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -53,11 +53,13 @@ final class StorageGrpcMetricsInterceptor implements ClientInterceptor {
@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
if (metrics == null || metrics.getRpcClientCallDuration() == null) {
if (metrics == null
|| (metrics.getRpcClientCallDuration() == null && metrics.getTtfb() == null)) {
return next.newCall(method, callOptions);
Comment on lines +56 to 58

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Relaxing this guard to allow the interceptor to run when metrics.getRpcClientCallDuration() is null (but metrics.getTtfb() is non-null) will cause a NullPointerException in onClose because onClose lacks a null check for rpcClientCallDuration.

In StorageHttpMetricsInterceptor.java, a null check was correctly added around the recording of httpClientRequestDuration (line 90). A similar null check must be added in onClose in this file.

Please update onClose to check for null before recording:

@Override
public void onClose(Status status, Metadata trailers) {
  try {
    if (callRecorded.compareAndSet(false, true) && metrics.getRpcClientCallDuration() != null) {
      // ... recording logic ...
    }
  } finally {
    super.onClose(status, trailers);
  }
}

}

String shortMethod = extractShortMethodName(method.getFullMethodName());
boolean isReadMethod = "ReadObject".equals(shortMethod) || "BidiReadObject".equals(shortMethod);
Comment on lines +56 to +62

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

If metrics.getRpcClientCallDuration() is null but metrics.getTtfb() is non-null, the interceptor will proceed. However, the onClose method (which is not shown in this diff but is part of the intercepted call) will attempt to record the call duration using metrics.getRpcClientCallDuration().record(...) without a null check, leading to a NullPointerException.

Additionally, if the method is not a read method (e.g., WriteObject) and metrics.getRpcClientCallDuration() is null, we still intercept the call even though we will not record any metrics for it.

We can solve both issues by extracting the method check earlier and defining recordDuration and recordTtfb flags. You should also wrap the duration recording in onClose with if (recordDuration).

    if (metrics == null) {
      return next.newCall(method, callOptions);
    }

    String shortMethod = extractShortMethodName(method.getFullMethodName());
    boolean isReadMethod = "ReadObject".equals(shortMethod) || "BidiReadObject".equals(shortMethod);

    boolean recordDuration = metrics.getRpcClientCallDuration() != null;
    boolean recordTtfb = isReadMethod && metrics.getTtfb() != null;

    if (!recordDuration && !recordTtfb) {
      return next.newCall(method, callOptions);
    }

String initialServerAddress =
StorageOptions.extractServerAddress(
defaultEndpoint != null ? defaultEndpoint : next.authority());
Expand All @@ -74,9 +76,32 @@ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
public void start(Listener<RespT> responseListener, Metadata headers) {
long startTimeNanos = clock.nanoTime();
AtomicBoolean callRecorded = new AtomicBoolean(false);
AtomicBoolean ttfbRecorded = new AtomicBoolean(false);

super.start(
new SimpleForwardingClientCallListener<RespT>(responseListener) {
@Override
public void onMessage(RespT message) {
if (isReadMethod
&& metrics.getTtfb() != null
&& ttfbRecorded.compareAndSet(false, true)) {
double ttfbSeconds =
Math.max(0.0, (clock.nanoTime() - startTimeNanos) / 1_000_000_000.0);
String networkPath = detectNetworkPath(getAttributes(), defaultEndpoint);
AttributesBuilder ttfbAttrs =
metrics
.clientAttributesBuilder()
.put(StorageClientMetrics.KEY_RPC_SYSTEM_NAME, "grpc")
.put(StorageClientMetrics.KEY_RPC_METHOD, shortMethod)
.put(StorageClientMetrics.KEY_NETWORK_PATH, networkPath)
.put(
StorageClientMetrics.KEY_SERVER_ADDRESS,
serverAddress != null ? serverAddress : "");
metrics.getTtfb().record(ttfbSeconds, ttfbAttrs.build(), otelContext);
}
super.onMessage(message);
}

@Override
public void onClose(Status status, Metadata trailers) {
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,8 @@ public void initialize(HttpRequest request) throws java.io.IOException {
if (delegate != null) {
delegate.initialize(request);
}
if (metrics == null || metrics.getHttpClientRequestDuration() == null) {
if (metrics == null
|| (metrics.getHttpClientRequestDuration() == null && metrics.getTtfb() == null)) {
return;
}
Comment on lines +58 to 61

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

We can optimize this guard by checking if we actually need to record any metrics for this specific request before registering the response interceptor. If metrics.getHttpClientRequestDuration() is null and the request is not a media download request, we can return early and avoid the overhead of intercepting the response.

Using effectively final recordDuration and recordTtfb flags also allows us to simplify the checks inside the response interceptor.

Suggested change
if (metrics == null
|| (metrics.getHttpClientRequestDuration() == null && metrics.getTtfb() == null)) {
return;
}
if (metrics == null) {
return;
}
boolean recordDuration = metrics.getHttpClientRequestDuration() != null;
boolean recordTtfb = metrics.getTtfb() != null && isMediaDownloadRequest(request);
if (!recordDuration && !recordTtfb) {
return;
}


Expand All @@ -80,37 +81,54 @@ public void initialize(HttpRequest request) throws java.io.IOException {
if (existingResp != null) {
existingResp.interceptResponse(response);
}
if (recorded.compareAndSet(false, true)
&& metrics.getHttpClientRequestDuration() != null) {
if (recorded.compareAndSet(false, true)) {
double durationSeconds =
Math.max(0.0, (clock.nanoTime() - startTimeNanos.get()) / 1_000_000_000.0);
int statusCode = response.getStatusCode();
String method =
request.getRequestMethod() != null ? request.getRequestMethod() : "UNKNOWN";
String urlTemplate = extractUrlTemplate(request.getUrl());
String serverAddress = resolveServerAddress(request.getUrl(), defaultServerAddress);

AttributesBuilder attrBuilder =
metrics
.clientAttributesBuilder()
.put(StorageClientMetrics.KEY_HTTP_REQUEST_METHOD, method)
.put(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE, (long) statusCode)
.put(
StorageClientMetrics.KEY_NETWORK_PATH,
StorageClientMetrics.NETWORK_PATH_CLOUDPATH)
.put(StorageClientMetrics.KEY_URL_TEMPLATE, urlTemplate)
.put(
StorageClientMetrics.KEY_SERVER_ADDRESS,
serverAddress != null ? serverAddress : "")
.put(
StorageClientMetrics.KEY_ERROR_TYPE,
statusCode >= 400
? StorageErrorTypeMapper.mapHttpResponseCode(statusCode)
: StorageErrorTypeMapper.STATUS_OK);
if (metrics.getHttpClientRequestDuration() != null) {
String method =
request.getRequestMethod() != null ? request.getRequestMethod() : "UNKNOWN";
String urlTemplate = extractUrlTemplate(request.getUrl());

metrics
.getHttpClientRequestDuration()
.record(durationSeconds, attrBuilder.build(), Context.current());
AttributesBuilder attrBuilder =
metrics
.clientAttributesBuilder()
.put(StorageClientMetrics.KEY_HTTP_REQUEST_METHOD, method)
.put(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE, (long) statusCode)
.put(
StorageClientMetrics.KEY_NETWORK_PATH,
StorageClientMetrics.NETWORK_PATH_CLOUDPATH)
.put(StorageClientMetrics.KEY_URL_TEMPLATE, urlTemplate)
.put(
StorageClientMetrics.KEY_SERVER_ADDRESS,
serverAddress != null ? serverAddress : "")
.put(
StorageClientMetrics.KEY_ERROR_TYPE,
statusCode >= 400
? StorageErrorTypeMapper.mapHttpResponseCode(statusCode)
: StorageErrorTypeMapper.STATUS_OK);

metrics
.getHttpClientRequestDuration()
.record(durationSeconds, attrBuilder.build(), Context.current());
}

if (metrics.getTtfb() != null && statusCode < 400 && isMediaDownloadRequest(request)) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The gcp.storage.client.ttfb metric measures the time to first byte of the actual media content. Recording TTFB for 3xx status codes (such as 304 Not Modified or redirects) is misleading because no media content is returned in those responses. Restricting this to successful 2xx status codes ensures the metric accurately reflects actual media download performance.

Suggested change
if (metrics.getTtfb() != null && statusCode < 400 && isMediaDownloadRequest(request)) {
if (metrics.getTtfb() != null && statusCode >= 200 && statusCode < 300 && isMediaDownloadRequest(request)) {

AttributesBuilder ttfbAttrs =
metrics
.clientAttributesBuilder()
.put(StorageClientMetrics.KEY_RPC_SYSTEM_NAME, "http")
.put(StorageClientMetrics.KEY_RPC_METHOD, "ReadObject")
.put(
StorageClientMetrics.KEY_NETWORK_PATH,
StorageClientMetrics.NETWORK_PATH_CLOUDPATH)
.put(
StorageClientMetrics.KEY_SERVER_ADDRESS,
serverAddress != null ? serverAddress : "");
metrics.getTtfb().record(durationSeconds, ttfbAttrs.build(), Context.current());
}
}
});

Expand Down Expand Up @@ -223,4 +241,17 @@ static String extractUrlTemplate(@Nullable GenericUrl url) {
String[] parts = trimmed.split("/", 2);
return parts.length == 1 ? "/{bucket}" : "/{bucket}/{object}";
}

static boolean isMediaDownloadRequest(@Nullable HttpRequest request) {
if (request == null) {
return false;
}
if (!"GET".equalsIgnoreCase(request.getRequestMethod()) || request.getUrl() == null) {
return false;
}
GenericUrl url = request.getUrl();
String path = url.getRawPath();
return (path != null && path.startsWith("/download/storage/v1/"))
|| "media".equals(url.getFirst("alt"));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,14 @@ public final class StorageGrpcMetricsInterceptorTest {
.setResponseMarshaller(new NoopMarshaller())
.build();

private static final MethodDescriptor<Void, Void> WRITE_OBJECT_METHOD =
MethodDescriptor.<Void, Void>newBuilder()
.setType(MethodDescriptor.MethodType.CLIENT_STREAMING)
.setFullMethodName("google.storage.v2.Storage/WriteObject")
.setRequestMarshaller(new NoopMarshaller())
.setResponseMarshaller(new NoopMarshaller())
.build();

private InMemoryMetricReader reader;
private SdkMeterProvider meterProvider;
private FakeClock clock;
Expand Down Expand Up @@ -215,6 +223,79 @@ public void interceptCall_callFails_recordsErrorType() throws Exception {
.isEqualTo("storage.googleapis.com");
}

@Test
public void interceptCall_readObject_recordsTtfbOnFirstMessage() throws Exception {
StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, true, false);
StorageGrpcMetricsInterceptor interceptor =
new StorageGrpcMetricsInterceptor(metrics, clock, "storage.googleapis.com");

TestClientCall<Void, Void> testCall = new TestClientCall<>(io.grpc.Attributes.EMPTY);
TestChannel testChannel = new TestChannel(testCall, "storage.googleapis.com");

ClientCall<Void, Void> call =
interceptor.interceptCall(READ_OBJECT_METHOD, CallOptions.DEFAULT, testChannel);
call.start(new TestListener<>(), new Metadata());

clock.advanceNanos(200_000_000L); // 0.2s
testCall.deliverMessage(null); // First response message: triggers TTFB

clock.advanceNanos(300_000_000L); // 0.3s more (total 0.5s)
testCall.deliverMessage(null); // Second message: should NOT record TTFB again
testCall.close(Status.OK, new Metadata());

Collection<MetricData> metricData = reader.collectAllMetrics();
MetricData ttfbData =
metricData.stream()
.filter(m -> m.getName().equals(StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_TTFB))
.findFirst()
.orElse(null);

assertThat(ttfbData).isNotNull();
assertThat(ttfbData.getHistogramData().getPoints()).hasSize(1);
HistogramPointData ttfbPoint = ttfbData.getHistogramData().getPoints().iterator().next();
assertThat(ttfbPoint.getSum()).isWithin(1e-6).of(0.2);
assertThat(ttfbPoint.getAttributes().get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME))
.isEqualTo("grpc");
assertThat(ttfbPoint.getAttributes().get(StorageClientMetrics.KEY_RPC_METHOD))
.isEqualTo("ReadObject");

MetricData durationData =
metricData.stream()
.filter(m -> m.getName().equals(StorageClientMetrics.METRIC_RPC_CLIENT_CALL_DURATION))
.findFirst()
.orElse(null);
assertThat(durationData).isNotNull();
HistogramPointData durationPoint =
durationData.getHistogramData().getPoints().iterator().next();
assertThat(durationPoint.getSum()).isWithin(1e-6).of(0.5);
}

@Test
public void interceptCall_nonReadMethod_doesNotRecordTtfb() throws Exception {
StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, true, false);
StorageGrpcMetricsInterceptor interceptor =
new StorageGrpcMetricsInterceptor(metrics, clock, "storage.googleapis.com");

TestClientCall<Void, Void> testCall = new TestClientCall<>(io.grpc.Attributes.EMPTY);
TestChannel testChannel = new TestChannel(testCall, "storage.googleapis.com");

ClientCall<Void, Void> call =
interceptor.interceptCall(WRITE_OBJECT_METHOD, CallOptions.DEFAULT, testChannel);
call.start(new TestListener<>(), new Metadata());

clock.advanceNanos(200_000_000L);
testCall.deliverMessage(null);
testCall.close(Status.OK, new Metadata());

Collection<MetricData> metricData = reader.collectAllMetrics();
MetricData ttfbData =
metricData.stream()
.filter(m -> m.getName().equals(StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_TTFB))
.findFirst()
.orElse(null);
assertThat(ttfbData).isNull();
}

@Test
public void interceptCall_disabledMetrics_doesNotRecord() {
StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, false, false);
Expand Down Expand Up @@ -262,6 +343,12 @@ public io.grpc.Attributes getAttributes() {
return attributes;
}

void deliverMessage(RespT message) {
if (listener != null) {
listener.onMessage(message);
}
}

void close(Status status, Metadata trailers) {
if (listener != null) {
listener.onClose(status, trailers);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -334,6 +334,121 @@ public LowLevelHttpResponse execute() throws IOException {
.isEqualTo("storage.googleapis.com");
}

@Test
public void isMediaDownloadRequest_cases() throws IOException {
HttpTransport transport = new MockHttpTransport();
HttpRequest getDownload =
transport
.createRequestFactory()
.buildGetRequest(
new GenericUrl(
"https://storage.googleapis.com/download/storage/v1/b/my-b/o/my-obj"));
assertThat(StorageHttpMetricsInterceptor.isMediaDownloadRequest(getDownload)).isTrue();

HttpRequest getAltMedia =
transport
.createRequestFactory()
.buildGetRequest(
new GenericUrl(
"https://storage.googleapis.com/storage/v1/b/my-b/o/my-obj?alt=media"));
assertThat(StorageHttpMetricsInterceptor.isMediaDownloadRequest(getAltMedia)).isTrue();

HttpRequest getMetadata =
transport
.createRequestFactory()
.buildGetRequest(
new GenericUrl("https://storage.googleapis.com/storage/v1/b/my-b/o/my-obj"));
assertThat(StorageHttpMetricsInterceptor.isMediaDownloadRequest(getMetadata)).isFalse();

HttpRequest postReq =
transport
.createRequestFactory()
.buildPostRequest(
new GenericUrl(
"https://storage.googleapis.com/download/storage/v1/b/my-b/o/my-obj"),
null);
assertThat(StorageHttpMetricsInterceptor.isMediaDownloadRequest(postReq)).isFalse();

assertThat(StorageHttpMetricsInterceptor.isMediaDownloadRequest(null)).isFalse();
}

@Test
public void initialize_mediaDownload_recordsTtfb() throws IOException {
StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, true, false);
StorageHttpMetricsInterceptor interceptor =
new StorageHttpMetricsInterceptor(null, metrics, clock, "https://storage.googleapis.com");

MockLowLevelHttpResponse lowLevelResponse = new MockLowLevelHttpResponse();
lowLevelResponse.setStatusCode(200);
HttpTransport transport =
new MockHttpTransport() {
@Override
public LowLevelHttpRequest buildRequest(String method, String url) {
return new MockLowLevelHttpRequest() {
@Override
public LowLevelHttpResponse execute() throws IOException {
clock.advanceNanos(250_000_000L); // 0.25s network duration
return lowLevelResponse;
}
};
}
};

HttpRequest request =
transport
.createRequestFactory(interceptor)
.buildGetRequest(
new GenericUrl(
"https://storage.googleapis.com/download/storage/v1/b/my-b/o/my-obj"));
request.execute();

Collection<MetricData> metricData = reader.collectAllMetrics();
MetricData ttfbData =
metricData.stream()
.filter(m -> m.getName().equals(StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_TTFB))
.findFirst()
.orElse(null);

assertThat(ttfbData).isNotNull();
assertThat(ttfbData.getHistogramData().getPoints()).hasSize(1);
HistogramPointData ttfbPoint = ttfbData.getHistogramData().getPoints().iterator().next();
assertThat(ttfbPoint.getSum()).isWithin(1e-6).of(0.25);
Attributes attrs = ttfbPoint.getAttributes();
assertThat(attrs.get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)).isEqualTo("http");
assertThat(attrs.get(StorageClientMetrics.KEY_RPC_METHOD)).isEqualTo("ReadObject");
assertThat(attrs.get(StorageClientMetrics.KEY_NETWORK_PATH))
.isEqualTo(StorageClientMetrics.NETWORK_PATH_CLOUDPATH);
assertThat(attrs.get(StorageClientMetrics.KEY_SERVER_ADDRESS))
.isEqualTo("storage.googleapis.com");
}

@Test
public void initialize_nonDownloadGet_doesNotRecordTtfb() throws IOException {
StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, true, false);
StorageHttpMetricsInterceptor interceptor =
new StorageHttpMetricsInterceptor(null, metrics, clock, "https://storage.googleapis.com");

MockLowLevelHttpResponse lowLevelResponse = new MockLowLevelHttpResponse();
lowLevelResponse.setStatusCode(200);
HttpTransport transport =
new MockHttpTransport.Builder().setLowLevelHttpResponse(lowLevelResponse).build();

HttpRequest request =
transport
.createRequestFactory(interceptor)
.buildGetRequest(
new GenericUrl("https://storage.googleapis.com/storage/v1/b/my-b/o/my-obj"));
request.execute();

Collection<MetricData> metricData = reader.collectAllMetrics();
MetricData ttfbData =
metricData.stream()
.filter(m -> m.getName().equals(StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_TTFB))
.findFirst()
.orElse(null);
assertThat(ttfbData).isNull();
}

@Test
public void initialize_metricsDisabled_doesNotRecord() throws IOException {
StorageClientMetrics metrics = StorageClientMetrics.create(meterProvider, false, false);
Expand Down
Loading