From 0f5016162511c0b1bbea63dc6f9af77a66f03df4 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Thu, 22 Oct 2026 07:56:25 +0000 Subject: [PATCH] feat: add gcp.storage.client.request.body.size metric for write operations --- .../storage/BaseStorageWriteChannel.java | 3 + .../google/cloud/storage/GrpcStorageImpl.java | 17 +++- .../com/google/cloud/storage/StorageImpl.java | 14 ++- .../cloud/storage/StorageMetricsContext.java | 27 ++++++ .../storage/StorageMetricsContextTest.java | 91 +++++++++++++++++++ 5 files changed, 147 insertions(+), 5 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java index 4751f671101b..0c4394da3914 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java @@ -138,6 +138,9 @@ public final int write(ByteBuffer src) throws IOException { return 0; } int write = tmp.write(src); + if (write > 0 && metricsContext != null) { + metricsContext.addRequestBytes(write); + } return write; } catch (StorageException e) { throw new IOException(e); diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java index 39152543420e..4a01a98a715f 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java @@ -284,7 +284,10 @@ public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... op .build(); try (UnbufferedWritableByteChannel c = session.open()) { - ByteStreams.copy(Channels.newChannel(inputStreamParam), c); + long copied = ByteStreams.copy(Channels.newChannel(inputStreamParam), c); + if (metricsContext != null) { + metricsContext.setRequestBytes(copied); + } } ApiFuture responseApiFuture = session.getResult(); return this.getBlob(responseApiFuture); @@ -315,6 +318,9 @@ public Blob internalCreateFrom(Path path, BlobInfo info, Opts o } StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + if (multiContext != null) { + multiContext.setRequestBytes(Files.size(path)); + } Throwable error = null; try { GrpcCallContext grpcCallContext = @@ -400,7 +406,10 @@ public Blob createFrom( ReadableByteChannel src = Channels.newChannel(firstNonNull(in, new ByteArrayInputStream(ZERO_BYTES))); try (BufferedWritableByteChannel dst = session.open()) { - ByteStreams.copy(src, dst); + long copied = ByteStreams.copy(src, dst); + if (multiContext != null) { + multiContext.setRequestBytes(copied); + } } catch (Exception e) { throw StorageException.coalesce(e); } @@ -938,6 +947,10 @@ public BlobInfo internalDirectUpload( WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); Hasher hasher = opts.getHasher(); StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); + int totalBytes = buf.remaining(); + if (metricsContext != null) { + metricsContext.setRequestBytes(totalBytes); + } GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); RewindableContent content = RewindableContent.of(buf); return retrier.run( diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java index cba4437e6e04..714d79c5df12 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java @@ -210,6 +210,9 @@ public Blob createFrom(BlobInfo blobInfo, Path path, int bufferSize, BlobWriteOp return create(blobInfo, null, options); } StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + if (multiContext != null) { + multiContext.setRequestBytes(size); + } Throwable error = null; try { Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); @@ -1921,6 +1924,7 @@ public BlobInfo internalCreateFrom(Path path, BlobInfo info, Opts opts, ByteBuffer buf) { + int totalBytes = buf.remaining(); BlobInfo.Builder builder = opts.blobInfoMapper().apply(info.toBuilder().clearMd5().clearCrc32c()); @Nullable Crc32cLengthKnown hash = opts.getHasher().hash(buf.duplicate()); @@ -1934,14 +1938,18 @@ public BlobInfo internalDirectUpload(BlobInfo info, Opts opts, ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsCreate(encoded, optionsMap); RewindableContent content = RewindableContent.of(buf); - return run( - "WriteObject", + StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); + if (metricsContext != null) { + metricsContext.setRequestBytes(totalBytes); + } + return retrier.run( algorithm, () -> { content.rewindTo(0); return storageRpc.create(encoded, new RewindableContentInputStream(content), optionsMap); }, - Conversions.json().blobInfo()); + Conversions.json().blobInfo(), + metricsContext); } /** diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java index 9f0b046507f2..1b1e8f87213a 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageMetricsContext.java @@ -182,6 +182,29 @@ void recordOperation( .build(); metrics.getGcpClientRequestDuration().record(getElapsedSeconds(), durationAttributes); } + + long reqBytes = requestBytes.get(); + if ("WriteObject".equals(operationName) + && reqBytes > 0 + && metrics.getRequestBodySize() != null) { + AttributesBuilder bodyAttrBuilder = Attributes.builder(); + bodyAttrBuilder.putAll(metrics.getClientAttributes()); + baseAttributes.forEach( + (key, val) -> { + if (!key.equals(StorageClientMetrics.KEY_NETWORK_PATH) + && !key.equals(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE) + && !key.equals(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE) + && !key.equals(StorageClientMetrics.KEY_ERROR_TYPE) + && !key.equals(StorageClientMetrics.KEY_GCP_STORAGE_OPERATION) + && !key.equals(StorageClientMetrics.KEY_GCP_CLIENT_METHOD)) { + @SuppressWarnings("unchecked") + AttributeKey objKey = (AttributeKey) key; + bodyAttrBuilder.put(objKey, val); + } + }); + bodyAttrBuilder.put(StorageClientMetrics.KEY_RPC_METHOD, operationName); + metrics.getRequestBodySize().record(reqBytes, bodyAttrBuilder.build(), otelContext); + } } } @@ -394,6 +417,10 @@ long addRequestBytes(long bytes) { return requestBytes.addAndGet(bytes); } + void setRequestBytes(long bytes) { + requestBytes.set(bytes); + } + long getRequestBytes() { return requestBytes.get(); } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java index e8beec487a8e..9527b724903a 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/StorageMetricsContextTest.java @@ -568,4 +568,95 @@ private static LongPointData getSinglePoint(Collection metrics, Stri assertThat(metric.getLongSumData().getPoints()).hasSize(1); return metric.getLongSumData().getPoints().iterator().next(); } + + @Test + public void recordOperation_recordsRequestBodySize_whenWriteObjectAndBytesGreaterThanZero() { + io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader reader = + io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader.create(); + io.opentelemetry.sdk.metrics.SdkMeterProvider provider = + io.opentelemetry.sdk.metrics.SdkMeterProvider.builder() + .registerMetricReader(reader) + .build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + + StorageMetricsContext context = StorageMetricsContext.create("WriteObject"); + context.setRequestBytes(2048L); + context.recordOperation(metrics, null); + + java.util.Collection collected = + reader.collectAllMetrics(); + io.opentelemetry.sdk.metrics.data.MetricData bodySizeData = + collected.stream() + .filter( + m -> + m.getName() + .equals(StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE)) + .findFirst() + .orElse(null); + + assertThat(bodySizeData).isNotNull(); + assertThat(bodySizeData.getHistogramData().getPoints()).hasSize(1); + io.opentelemetry.sdk.metrics.data.HistogramPointData point = + bodySizeData.getHistogramData().getPoints().iterator().next(); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo(2048.0); + assertThat(point.getAttributes().get(StorageClientMetrics.KEY_RPC_METHOD)) + .isEqualTo("WriteObject"); + } + + @Test + public void recordOperation_doesNotRecordRequestBodySize_whenBytesIsZero() { + io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader reader = + io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader.create(); + io.opentelemetry.sdk.metrics.SdkMeterProvider provider = + io.opentelemetry.sdk.metrics.SdkMeterProvider.builder() + .registerMetricReader(reader) + .build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + + StorageMetricsContext context = StorageMetricsContext.create("WriteObject"); + context.setRequestBytes(0L); + context.recordOperation(metrics, null); + + java.util.Collection collected = + reader.collectAllMetrics(); + io.opentelemetry.sdk.metrics.data.MetricData bodySizeData = + collected.stream() + .filter( + m -> + m.getName() + .equals(StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE)) + .findFirst() + .orElse(null); + + assertThat(bodySizeData).isNull(); + } + + @Test + public void recordOperation_doesNotRecordRequestBodySize_whenNotWriteObject() { + io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader reader = + io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader.create(); + io.opentelemetry.sdk.metrics.SdkMeterProvider provider = + io.opentelemetry.sdk.metrics.SdkMeterProvider.builder() + .registerMetricReader(reader) + .build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + + StorageMetricsContext context = StorageMetricsContext.create("ReadObject"); + context.setRequestBytes(2048L); + context.recordOperation(metrics, null); + + java.util.Collection collected = + reader.collectAllMetrics(); + io.opentelemetry.sdk.metrics.data.MetricData bodySizeData = + collected.stream() + .filter( + m -> + m.getName() + .equals(StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE)) + .findFirst() + .orElse(null); + + assertThat(bodySizeData).isNull(); + } }