diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java index b5c86752d97f..de68cb41b8d7 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java @@ -153,6 +153,9 @@ public final int read(ByteBuffer dst) throws IOException { int read = tmp.read(dst); if (read != -1) { byteRangeSpec = byteRangeSpec.withShiftBeginOffset(read); + if (read > 0 && metricsContext != null) { + metricsContext.addResponseBytes(read); + } } return read; } catch (StorageException 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 4a01a98a715f..06c089983b90 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 @@ -819,6 +819,7 @@ public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) { throw e; } finally { if (metricsContext != null) { + metricsContext.setResponseBytes(baos.size()); metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); } } @@ -864,7 +865,7 @@ public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) { Throwable error = null; try (UnbufferedReadableByteChannel r = session.open(); - WritableByteChannel w = Files.newByteChannel(path, WRITE_OPS)) { + WritableByteChannel w = countBytes(Files.newByteChannel(path, WRITE_OPS), metricsContext)) { ByteStreams.copy(r, w); } catch (ApiException | IOException e) { error = e; @@ -887,7 +888,7 @@ public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption. Throwable error = null; try (UnbufferedReadableByteChannel r = session.open(); - WritableByteChannel w = Channels.newChannel(outputStream)) { + WritableByteChannel w = countBytes(Channels.newChannel(outputStream), metricsContext)) { ByteStreams.copy(r, w); } catch (ApiException | IOException e) { error = e; @@ -902,6 +903,33 @@ public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption. } } + private static WritableByteChannel countBytes( + WritableByteChannel delegate, @Nullable StorageMetricsContext metricsContext) { + if (metricsContext == null) { + return delegate; + } + return new WritableByteChannel() { + @Override + public int write(ByteBuffer src) throws IOException { + int written = delegate.write(src); + if (written > 0) { + metricsContext.addResponseBytes(written); + } + return written; + } + + @Override + public boolean isOpen() { + return delegate.isOpen(); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + }; + } + @Override public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts); 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 714d79c5df12..1a6bcde41bb2 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 @@ -761,11 +761,18 @@ public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) { ImmutableMap optionsMap = resolve.getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storageObject, optionsMap); - return run( - "ReadObject", + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ReadObject"); + return retrier.run( algorithm, - () -> storageRpc.load(storageObject, optionsMap), - Decoder.identity()); + () -> { + byte[] bytes = storageRpc.load(storageObject, optionsMap); + if (metricsContext != null && bytes != null) { + metricsContext.setResponseBytes(bytes.length); + } + return bytes; + }, + Decoder.identity(), + metricsContext); } @Override @@ -804,15 +811,22 @@ public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption. ImmutableMap optionsMap = Opts.unwrap(options).resolveFrom(blob).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(pb, optionsMap); - run( - "ReadObject", + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ReadObject"); + retrier.run( algorithm, callable( () -> { - storageRpc.read( - pb, optionsMap, countingOutputStream.getCount(), countingOutputStream); + try { + storageRpc.read( + pb, optionsMap, countingOutputStream.getCount(), countingOutputStream); + } finally { + if (metricsContext != null) { + metricsContext.setResponseBytes(countingOutputStream.getCount()); + } + } }), - Decoder.identity()); + Decoder.identity(), + metricsContext); } @Override 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 1b1e8f87213a..f04c684a8cc6 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 @@ -205,6 +205,29 @@ void recordOperation( bodyAttrBuilder.put(StorageClientMetrics.KEY_RPC_METHOD, operationName); metrics.getRequestBodySize().record(reqBytes, bodyAttrBuilder.build(), otelContext); } + + long respBytes = responseBytes.get(); + if ("ReadObject".equals(operationName) + && respBytes > 0 + && metrics.getResponseBodySize() != 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.getResponseBodySize().record(respBytes, bodyAttrBuilder.build(), otelContext); + } } } @@ -429,6 +452,10 @@ long addResponseBytes(long bytes) { return responseBytes.addAndGet(bytes); } + void setResponseBytes(long bytes) { + responseBytes.set(bytes); + } + long getResponseBytes() { return responseBytes.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 9527b724903a..cfb38ca18109 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 @@ -659,4 +659,95 @@ public void recordOperation_doesNotRecordRequestBodySize_whenNotWriteObject() { assertThat(bodySizeData).isNull(); } + + @Test + public void recordOperation_recordsResponseBodySize_whenReadObjectAndBytesGreaterThanZero() { + 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.setResponseBytes(4096L); + 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_RESPONSE_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(4096.0); + assertThat(point.getAttributes().get(StorageClientMetrics.KEY_RPC_METHOD)) + .isEqualTo("ReadObject"); + } + + @Test + public void recordOperation_doesNotRecordResponseBodySize_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("ReadObject"); + context.setResponseBytes(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_RESPONSE_BODY_SIZE)) + .findFirst() + .orElse(null); + + assertThat(bodySizeData).isNull(); + } + + @Test + public void recordOperation_doesNotRecordResponseBodySize_whenNotReadObject() { + 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.setResponseBytes(4096L); + 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_RESPONSE_BODY_SIZE)) + .findFirst() + .orElse(null); + + assertThat(bodySizeData).isNull(); + } }