From e87b7d95038188f43318ecd589b1023fc3248801 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Thu, 22 Oct 2026 08:04:53 +0000 Subject: [PATCH 1/2] feat: add gcp.storage.client.response.body.size metric for read operations --- .../cloud/storage/BaseStorageReadChannel.java | 3 + .../google/cloud/storage/GrpcStorageImpl.java | 32 ++++++- .../com/google/cloud/storage/StorageImpl.java | 32 +++++-- .../cloud/storage/StorageMetricsContext.java | 27 ++++++ .../storage/StorageMetricsContextTest.java | 91 +++++++++++++++++++ 5 files changed, 174 insertions(+), 11 deletions(-) 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(); + } } From 404687b24c1a8c67db0163bf0ff12c5e1ff60a29 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Tue, 3 Nov 2026 10:37:05 +0000 Subject: [PATCH 2/2] fix(storage): address review feedback and wire blobReadSession response body size - Extract BODY_SIZE_EXCLUDED_KEYS and buildBodySizeAttributes(StorageClientMetrics) in StorageMetricsContext to deduplicate request/response body size attribute filtering. - Guard GrpcStorageImpl.readAllBytes response byte recording with error == null and move countBytes helper to the private helper section. - Wire GrpcStorageImpl.blobReadSession and BlobReadSessionAdapter to track delivered response bytes across all ReadProjectionConfig types and record gcp.storage.client.response.body.size on session close. - Track lastReadError in BaseStorageReadChannel so mid-stream read errors propagate to recordOperation on close while preserving partial read byte counts. - Replace inline FQCNs with imports in StorageMetricsContextTest and add unit tests for channel reads, StorageImpl, GrpcStorageImpl, and BlobReadSessionAdapter. [Generated-by: AI] --- .../cloud/storage/BaseStorageReadChannel.java | 11 +- .../cloud/storage/BlobReadSessionAdapter.java | 215 +++- .../google/cloud/storage/GrpcStorageImpl.java | 78 +- .../cloud/storage/StorageMetricsContext.java | 65 +- .../storage/StorageMetricsContextTest.java | 921 +++++++++++++++--- 5 files changed, 1104 insertions(+), 186 deletions(-) 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 de68cb41b8d7..dfe5f913c6de 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 @@ -45,6 +45,7 @@ abstract class BaseStorageReadChannel implements StorageReadChannel { protected final ReentrantLock lock; @Nullable private final StorageMetricsContext metricsContext; @Nullable private final StorageClientMetrics storageClientMetrics; + @Nullable private Throwable lastReadError; protected BaseStorageReadChannel(Decoder objectDecoder) { this(objectDecoder, null, null); @@ -102,7 +103,8 @@ public final void close() { } finally { try { if (metricsContext != null && storageClientMetrics != null) { - metricsContext.recordOperation(storageClientMetrics, error); + metricsContext.recordOperation( + storageClientMetrics, error != null ? error : lastReadError); } } finally { lock.unlock(); @@ -151,6 +153,7 @@ public final int read(ByteBuffer dst) throws IOException { return -1; } int read = tmp.read(dst); + lastReadError = null; if (read != -1) { byteRangeSpec = byteRangeSpec.withShiftBeginOffset(read); if (read > 0 && metricsContext != null) { @@ -166,12 +169,16 @@ public final int read(ByteBuffer dst) throws IOException { // be removed in the next major version. return -1; } else { + lastReadError = e; throw new IOException(e); } } catch (IOException e) { + lastReadError = e; throw e; } catch (Exception e) { - throw new IOException(StorageException.coalesce(e)); + Throwable coalesced = StorageException.coalesce(e); + lastReadError = coalesced; + throw new IOException(coalesced); } } finally { lock.unlock(); diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSessionAdapter.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSessionAdapter.java index 1823a13283a1..9fcec633e612 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSessionAdapter.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadSessionAdapter.java @@ -17,17 +17,35 @@ package com.google.cloud.storage; import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutureCallback; import com.google.api.core.ApiFutures; +import com.google.cloud.storage.ZeroCopySupport.DisposableByteString; import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.MoreExecutors; import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.ScatteringByteChannel; +import java.nio.channels.SeekableByteChannel; +import org.checkerframework.checker.nullness.qual.Nullable; final class BlobReadSessionAdapter implements BlobReadSession { @VisibleForTesting final ObjectReadSession session; + @Nullable private final StorageMetricsContext metricsContext; + @Nullable private final StorageClientMetrics storageClientMetrics; + @Nullable private volatile Throwable lastOperationError; BlobReadSessionAdapter(ObjectReadSession session) { + this(session, null, null); + } + + BlobReadSessionAdapter( + ObjectReadSession session, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { this.session = session; + this.metricsContext = metricsContext; + this.storageClientMetrics = storageClientMetrics; } @Override @@ -40,23 +58,208 @@ public BlobInfo getBlobInfo() { @SuppressWarnings({"rawtypes", "unchecked"}) @Override public Projection readAs(ReadProjectionConfig config) { - Projection projection = session.readAs(config); + Projection projection; + try { + projection = session.readAs(config); + } catch (Throwable t) { + lastOperationError = t; + throw t; + } if (projection instanceof ApiFuture) { - ApiFuture apiFuture = (ApiFuture) projection; - return (Projection) StorageException.coalesceAsync(apiFuture); + ApiFuture apiFuture = StorageException.coalesceAsync((ApiFuture) projection); + if (metricsContext != null) { + ApiFutures.addCallback( + apiFuture, + new ApiFutureCallback() { + @Override + public void onSuccess(Object result) { + if (result instanceof byte[]) { + int len = ((byte[]) result).length; + if (len > 0) { + metricsContext.addResponseBytes(len); + } + } else if (result instanceof DisposableByteString) { + int len = ((DisposableByteString) result).byteString().size(); + if (len > 0) { + metricsContext.addResponseBytes(len); + } + } + } + + @Override + public void onFailure(Throwable t) { + if (!(t instanceof AsyncSessionClosedException) + && !(t.getCause() instanceof AsyncSessionClosedException)) { + lastOperationError = t; + } + } + }, + MoreExecutors.directExecutor()); + } + return (Projection) apiFuture; + } + if (metricsContext != null && projection instanceof SeekableByteChannel) { + return (Projection) wrapSeekableByteChannel((SeekableByteChannel) projection); + } + if (metricsContext != null && projection instanceof ScatteringByteChannel) { + return (Projection) wrapScatteringByteChannel((ScatteringByteChannel) projection); } return projection; } @Override public void close() throws IOException { - session.close(); + Throwable priorError = lastOperationError; + Throwable error = null; + try { + session.close(); + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (metricsContext != null && storageClientMetrics != null) { + metricsContext.recordOperation(storageClientMetrics, error != null ? error : priorError); + } + } + } + + private SeekableByteChannel wrapSeekableByteChannel(SeekableByteChannel delegate) { + return new SeekableByteChannel() { + @Override + public int read(ByteBuffer dst) throws IOException { + try { + int read = delegate.read(dst); + if (read > 0 && metricsContext != null) { + metricsContext.addResponseBytes(read); + } + return read; + } catch (IOException | RuntimeException | Error e) { + lastOperationError = e; + throw e; + } + } + + @Override + public int write(ByteBuffer src) throws IOException { + return delegate.write(src); + } + + @Override + public long position() throws IOException { + return delegate.position(); + } + + @Override + public SeekableByteChannel position(long newPosition) throws IOException { + delegate.position(newPosition); + return this; + } + + @Override + public long size() throws IOException { + return delegate.size(); + } + + @Override + public SeekableByteChannel truncate(long size) throws IOException { + delegate.truncate(size); + return this; + } + + @Override + public boolean isOpen() { + return delegate.isOpen(); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + }; + } + + private ScatteringByteChannel wrapScatteringByteChannel(ScatteringByteChannel delegate) { + return new ScatteringByteChannel() { + @Override + public int read(ByteBuffer dst) throws IOException { + try { + int read = delegate.read(dst); + if (read > 0 && metricsContext != null) { + metricsContext.addResponseBytes(read); + } + return read; + } catch (IOException | RuntimeException | Error e) { + lastOperationError = e; + throw e; + } + } + + @Override + public long read(ByteBuffer[] dsts, int offset, int length) throws IOException { + try { + long read = delegate.read(dsts, offset, length); + if (read > 0 && metricsContext != null) { + metricsContext.addResponseBytes(read); + } + return read; + } catch (IOException | RuntimeException | Error e) { + lastOperationError = e; + throw e; + } + } + + @Override + public long read(ByteBuffer[] dsts) throws IOException { + try { + long read = delegate.read(dsts); + if (read > 0 && metricsContext != null) { + metricsContext.addResponseBytes(read); + } + return read; + } catch (IOException | RuntimeException | Error e) { + lastOperationError = e; + throw e; + } + } + + @Override + public boolean isOpen() { + return delegate.isOpen(); + } + + @Override + public void close() throws IOException { + delegate.close(); + } + }; } static ApiFuture wrap(ApiFuture session) { + return wrap(session, null, null); + } + + static ApiFuture wrap( + ApiFuture session, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { + ApiFuture coalesced = StorageException.coalesceAsync(session); + if (metricsContext != null && storageClientMetrics != null) { + ApiFutures.addCallback( + coalesced, + new ApiFutureCallback() { + @Override + public void onSuccess(ObjectReadSession result) {} + + @Override + public void onFailure(Throwable t) { + metricsContext.recordOperation(storageClientMetrics, t); + } + }, + MoreExecutors.directExecutor()); + } return ApiFutures.transform( - StorageException.coalesceAsync(session), - BlobReadSessionAdapter::new, + coalesced, + s -> new BlobReadSessionAdapter(s, metricsContext, storageClientMetrics), MoreExecutors.directExecutor()); } } 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 06c089983b90..3a0a19ba3254 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,7 +819,9 @@ public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) { throw e; } finally { if (metricsContext != null) { - metricsContext.setResponseBytes(baos.size()); + if (error == null) { + metricsContext.setResponseBytes(baos.size()); + } metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); } } @@ -903,33 +905,6 @@ 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); @@ -1738,10 +1713,22 @@ public ApiFuture blobReadSession(BlobId id, BlobSourceOption... opts.bidiReadObjectRequest().apply(b); BidiReadObjectRequest req = b.build(); - GrpcCallContext context = opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); - ApiFuture session = storageDataClient.readSession(req, context); - - return BlobReadSessionAdapter.wrap(session); + StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); + GrpcCallContext baseContext = + metricsContext != null + ? metricsContext.attach(GrpcCallContext.createDefault()) + : GrpcCallContext.createDefault(); + GrpcCallContext context = opts.grpcMetadataMapper().apply(baseContext); + try { + ApiFuture session = storageDataClient.readSession(req, context); + return BlobReadSessionAdapter.wrap( + session, metricsContext, getOptions().getStorageClientMetrics()); + } catch (Throwable t) { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } } @Override @@ -2449,4 +2436,31 @@ private ZeroCopyServerStreamingCallable r storageClient.readObjectCallable().withDefaultCallContext(grpcCallContext), responseContentLifecycleManager); } + + 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(); + } + }; + } } 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 f04c684a8cc6..83016675e31b 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 @@ -20,6 +20,7 @@ import com.google.api.gax.grpc.GrpcCallContext; import com.google.api.gax.rpc.ApiCallContext; +import com.google.common.collect.ImmutableSet; import io.grpc.CallOptions; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; @@ -44,6 +45,15 @@ final class StorageMetricsContext { ApiCallContext.Key.create("storage-metrics-context"); static final String HTTP_REQUEST_PROPERTY_KEY = "com.google.cloud.storage.StorageMetricsContext"; + private static final ImmutableSet> BODY_SIZE_EXCLUDED_KEYS = + ImmutableSet.of( + StorageClientMetrics.KEY_NETWORK_PATH, + StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE, + StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE, + StorageClientMetrics.KEY_ERROR_TYPE, + StorageClientMetrics.KEY_GCP_STORAGE_OPERATION, + StorageClientMetrics.KEY_GCP_CLIENT_METHOD); + private final String operationName; private final long startTimeNanos; private final Context otelContext; @@ -187,50 +197,37 @@ void recordOperation( 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); + metrics + .getRequestBodySize() + .record(reqBytes, buildBodySizeAttributes(metrics), 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); + metrics + .getResponseBodySize() + .record(respBytes, buildBodySizeAttributes(metrics), otelContext); } } } + private Attributes buildBodySizeAttributes(StorageClientMetrics metrics) { + AttributesBuilder bodyAttrBuilder = Attributes.builder(); + bodyAttrBuilder.putAll(metrics.getClientAttributes()); + baseAttributes.forEach( + (key, val) -> { + if (!BODY_SIZE_EXCLUDED_KEYS.contains(key)) { + @SuppressWarnings("unchecked") + AttributeKey objKey = (AttributeKey) key; + bodyAttrBuilder.put(objKey, val); + } + }); + bodyAttrBuilder.put(StorageClientMetrics.KEY_RPC_METHOD, operationName); + return bodyAttrBuilder.build(); + } + void recordAttemptSuccess(@Nullable StorageClientMetrics metrics) { if (metrics == null || metrics.getAttempts() == null) { return; 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 cfb38ca18109..4b6d14a87d1c 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 @@ -21,26 +21,60 @@ import com.google.api.client.http.HttpHeaders; import com.google.api.client.http.HttpResponseException; +import com.google.api.core.ApiFuture; +import com.google.api.core.ApiFutures; import com.google.api.gax.grpc.GrpcCallContext; import com.google.api.gax.grpc.GrpcStatusCode; +import com.google.api.gax.retrying.RetrySettings; import com.google.api.gax.rpc.ApiCallContext; import com.google.api.gax.rpc.ApiException; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.ReadChannel; +import com.google.cloud.RestorableState; +import com.google.cloud.storage.BufferedReadableByteChannelSession.BufferedReadableByteChannel; +import com.google.cloud.storage.ZeroCopySupport.DisposableByteString; +import com.google.cloud.storage.it.ChecksummedTestContent; +import com.google.cloud.storage.spi.StorageRpcFactory; +import com.google.cloud.storage.spi.v1.StorageRpc; +import com.google.protobuf.ByteString; +import com.google.storage.v2.BidiReadObjectRequest; +import com.google.storage.v2.BidiReadObjectResponse; +import com.google.storage.v2.BucketName; +import com.google.storage.v2.ObjectRangeData; +import com.google.storage.v2.ReadObjectRequest; +import com.google.storage.v2.ReadObjectResponse; +import com.google.storage.v2.ReadRange; +import com.google.storage.v2.StorageGrpc.StorageImplBase; import io.grpc.CallOptions; import io.grpc.Status; import io.grpc.StatusException; import io.grpc.StatusRuntimeException; +import io.grpc.stub.StreamObserver; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.common.Attributes; import io.opentelemetry.context.Context; import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.HistogramPointData; import io.opentelemetry.sdk.metrics.data.LongPointData; import io.opentelemetry.sdk.metrics.data.MetricData; import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.nio.channels.ScatteringByteChannel; +import java.nio.channels.SeekableByteChannel; +import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.Collection; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import org.checkerframework.checker.nullness.qual.Nullable; import org.junit.Test; +import org.mockito.Mockito; public final class StorageMetricsContextTest { @@ -569,185 +603,848 @@ private static LongPointData getSinglePoint(Collection metrics, Stri return metric.getLongSumData().getPoints().iterator().next(); } + private static HistogramPointData getSingleHistogramPoint( + Collection metrics, String metricName) { + MetricData metric = findMetric(metrics, metricName); + assertThat(metric).isNotNull(); + assertThat(metric.getHistogramData().getPoints()).hasSize(1); + return metric.getHistogramData().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(); + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); - StorageMetricsContext context = StorageMetricsContext.create("WriteObject"); + Attributes baseAttributes = + Attributes.builder() + .put(StorageClientMetrics.KEY_RPC_SYSTEM_NAME, "grpc") + .put(StorageClientMetrics.KEY_SERVER_ADDRESS, "storage.googleapis.com") + .put(StorageClientMetrics.KEY_SERVER_PORT, 443L) + .put( + StorageClientMetrics.KEY_NETWORK_PATH, StorageClientMetrics.NETWORK_PATH_DIRECTPATH) + .put(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE, "OK") + .put(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE, 200L) + .put(StorageClientMetrics.KEY_ERROR_TYPE, StorageErrorTypeMapper.STATUS_OK) + .put(StorageClientMetrics.KEY_GCP_STORAGE_OPERATION, "WriteObject") + .put(StorageClientMetrics.KEY_GCP_CLIENT_METHOD, "Storage.writer") + .build(); + StorageMetricsContext context = StorageMetricsContext.create("WriteObject", baseAttributes); 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(); + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); assertThat(point.getSum()).isEqualTo(2048.0); - assertThat(point.getAttributes().get(StorageClientMetrics.KEY_RPC_METHOD)) - .isEqualTo("WriteObject"); + Attributes attrs = point.getAttributes(); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_METHOD)).isEqualTo("WriteObject"); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)).isEqualTo("grpc"); + assertThat(attrs.get(StorageClientMetrics.KEY_SERVER_ADDRESS)) + .isEqualTo("storage.googleapis.com"); + assertThat(attrs.get(StorageClientMetrics.KEY_SERVER_PORT)).isEqualTo(443L); + assertThat(attrs.get(StorageClientMetrics.KEY_GCP_CLIENT_SERVICE)).isEqualTo("storage"); + assertThat(attrs.get(StorageClientMetrics.KEY_NETWORK_PATH)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_ERROR_TYPE)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_GCP_STORAGE_OPERATION)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_GCP_CLIENT_METHOD)).isNull(); } @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(); + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = 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); + Collection collected = reader.collectAllMetrics(); + MetricData bodySizeData = + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); 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(); + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = 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); + Collection collected = reader.collectAllMetrics(); + MetricData bodySizeData = + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_REQUEST_BODY_SIZE); 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(); + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); - StorageMetricsContext context = StorageMetricsContext.create("ReadObject"); + Attributes baseAttributes = + Attributes.builder() + .put(StorageClientMetrics.KEY_RPC_SYSTEM_NAME, "grpc") + .put(StorageClientMetrics.KEY_SERVER_ADDRESS, "storage.googleapis.com") + .put(StorageClientMetrics.KEY_SERVER_PORT, 443L) + .put( + StorageClientMetrics.KEY_NETWORK_PATH, StorageClientMetrics.NETWORK_PATH_DIRECTPATH) + .put(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE, "OK") + .put(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE, 200L) + .put(StorageClientMetrics.KEY_ERROR_TYPE, StorageErrorTypeMapper.STATUS_OK) + .put(StorageClientMetrics.KEY_GCP_STORAGE_OPERATION, "ReadObject") + .put(StorageClientMetrics.KEY_GCP_CLIENT_METHOD, "Storage.reader") + .build(); + StorageMetricsContext context = StorageMetricsContext.create("ReadObject", baseAttributes); 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(); + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); assertThat(point.getSum()).isEqualTo(4096.0); - assertThat(point.getAttributes().get(StorageClientMetrics.KEY_RPC_METHOD)) - .isEqualTo("ReadObject"); + Attributes attrs = point.getAttributes(); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_METHOD)).isEqualTo("ReadObject"); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_SYSTEM_NAME)).isEqualTo("grpc"); + assertThat(attrs.get(StorageClientMetrics.KEY_SERVER_ADDRESS)) + .isEqualTo("storage.googleapis.com"); + assertThat(attrs.get(StorageClientMetrics.KEY_SERVER_PORT)).isEqualTo(443L); + assertThat(attrs.get(StorageClientMetrics.KEY_GCP_CLIENT_SERVICE)).isEqualTo("storage"); + assertThat(attrs.get(StorageClientMetrics.KEY_NETWORK_PATH)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_RPC_RESPONSE_STATUS_CODE)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_HTTP_RESPONSE_STATUS_CODE)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_ERROR_TYPE)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_GCP_STORAGE_OPERATION)).isNull(); + assertThat(attrs.get(StorageClientMetrics.KEY_GCP_CLIENT_METHOD)).isNull(); } @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(); + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = 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); + Collection collected = reader.collectAllMetrics(); + MetricData bodySizeData = + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); 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(); + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = 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); + Collection collected = reader.collectAllMetrics(); + MetricData bodySizeData = + findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); assertThat(bodySizeData).isNull(); } + + @Test + public void baseStorageReadChannel_recordsResponseBodySizeOnClose_fullRead() throws IOException { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("ReadObject"); + + byte[] payload = "0123456789abcdef0123456789abcdef".getBytes(StandardCharsets.UTF_8); + try (StorageReadChannel channel = new FakeStorageReadChannel(payload, null, context, metrics)) { + ByteBuffer buf = ByteBuffer.allocate(10); + int total = 0; + int r; + while ((r = channel.read(buf)) != -1) { + total += r; + buf.clear(); + } + assertThat(total).isEqualTo(payload.length); + } + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo((double) payload.length); + LongPointData opPoint = + getSinglePoint(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS); + assertThat(opPoint.getAttributes().get(StorageClientMetrics.KEY_ERROR_TYPE)) + .isEqualTo(StorageErrorTypeMapper.STATUS_OK); + } + + @Test + public void baseStorageReadChannel_recordsResponseBodySizeOnClose_partialReadEarlyClose() + throws IOException { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("ReadObject"); + + byte[] payload = new byte[100]; + try (StorageReadChannel channel = new FakeStorageReadChannel(payload, null, context, metrics)) { + ByteBuffer buf = ByteBuffer.allocate(12); + int read = channel.read(buf); + assertThat(read).isEqualTo(12); + } + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo(12.0); + } + + @Test + public void baseStorageReadChannel_doesNotRecordResponseBodySize_whenClosedWithoutReading() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("ReadObject"); + + byte[] payload = new byte[100]; + try (StorageReadChannel ignored = new FakeStorageReadChannel(payload, null, context, metrics)) { + // Close immediately without reading any bytes + } + + Collection collected = reader.collectAllMetrics(); + assertThat( + findMetric( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE)) + .isNull(); + assertThat(findMetric(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS)) + .isNotNull(); + } + + @Test + public void baseStorageReadChannel_recordsPartialBytesAndError_whenMidStreamErrorOccurs() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("ReadObject"); + + byte[] payload = new byte[64]; + StorageException midStreamError = new StorageException(503, "Service Unavailable"); + IOException thrown = + assertThrows( + IOException.class, + () -> { + try (StorageReadChannel channel = + new FakeStorageReadChannel(payload, midStreamError, context, metrics)) { + ByteBuffer buf = ByteBuffer.allocate(16); + assertThat(channel.read(buf)).isEqualTo(16); + buf.clear(); + channel.read(buf); + } + }); + assertThat(thrown).hasCauseThat().isSameInstanceAs(midStreamError); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo(16.0); + + LongPointData opPoint = + getSinglePoint(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS); + assertThat(opPoint.getAttributes().get(StorageClientMetrics.KEY_ERROR_TYPE)) + .isEqualTo(StorageErrorTypeMapper.STATUS_UNAVAILABLE); + } + + @Test + public void storageImpl_readAllBytes_recordsResponseBodySize() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageRpc storageRpc = Mockito.mock(StorageRpc.class); + StorageRpcFactory rpcFactory = options -> storageRpc; + + byte[] payload = "response-body-bytes".getBytes(StandardCharsets.UTF_8); + Mockito.when(storageRpc.load(Mockito.any(StorageObject.class), Mockito.anyMap())) + .thenReturn(payload); + + HttpStorageOptions options = + HttpStorageOptions.newBuilder() + .setProjectId("test-proj") + .setServiceRpcFactory(rpcFactory) + .setMeterProvider(provider) + .setEnableOtelMetrics(true) + .build(); + Storage storage = options.getService(); + + byte[] actual = storage.readAllBytes(BlobId.of("b", "o")); + assertThat(actual).isEqualTo(payload); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo((double) payload.length); + } + + @Test + public void storageImpl_downloadTo_recordsResponseBodySize_acrossRetriesWithoutDoubleCounting() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageRpc storageRpc = Mockito.mock(StorageRpc.class); + StorageRpcFactory rpcFactory = options -> storageRpc; + + byte[] payload = + "0123456789abcdefghij0123456789abcdefghij0123456789".getBytes(StandardCharsets.UTF_8); + AtomicInteger callCount = new AtomicInteger(0); + Mockito.doAnswer( + invocation -> { + long position = invocation.getArgument(2); + OutputStream out = invocation.getArgument(3); + int attempt = callCount.incrementAndGet(); + if (attempt == 1) { + out.write(payload, 0, 20); + throw new StorageException(503, "Transient failure after 20 bytes"); + } + out.write(payload, (int) position, payload.length - (int) position); + return (long) (payload.length - (int) position); + }) + .when(storageRpc) + .read( + Mockito.any(StorageObject.class), + Mockito.anyMap(), + Mockito.anyLong(), + Mockito.any(OutputStream.class)); + + HttpStorageOptions options = + HttpStorageOptions.newBuilder() + .setProjectId("test-proj") + .setServiceRpcFactory(rpcFactory) + .setRetrySettings( + RetrySettings.newBuilder() + .setMaxAttempts(3) + .setInitialRetryDelayDuration(Duration.ofMillis(1)) + .setMaxRetryDelayDuration(Duration.ofMillis(5)) + .setRetryDelayMultiplier(1.0) + .build()) + .setMeterProvider(provider) + .setEnableOtelMetrics(true) + .build(); + Storage storage = options.getService(); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + storage.downloadTo(BlobId.of("b", "o"), out); + assertThat(out.toByteArray()).isEqualTo(payload); + assertThat(callCount.get()).isEqualTo(2); + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo((double) payload.length); + } + + @Test + public void blobReadSessionAdapter_recordsResponseBodySizeOnClose_acrossProjections() + throws Exception { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("ReadObject"); + + byte[] futureBytesPayload = new byte[25]; + ByteString byteStringPayload = ByteString.copyFrom(new byte[15]); + byte[] channelPayload = new byte[40]; + byte[] seekablePayload = new byte[30]; + + ObjectReadSession fakeSession = + new ObjectReadSession() { + @Override + public com.google.storage.v2.Object getResource() { + return com.google.storage.v2.Object.getDefaultInstance(); + } + + @SuppressWarnings("unchecked") + @Override + public Projection readAs(ReadProjectionConfig config) { + if (config instanceof ReadAsFutureBytes) { + return (Projection) ApiFutures.immediateFuture(futureBytesPayload); + } + if (config instanceof ReadAsFutureByteString) { + DisposableByteString dbs = + new DisposableByteString() { + @Override + public ByteString byteString() { + return byteStringPayload; + } + + @Override + public void close() {} + }; + return (Projection) ApiFutures.immediateFuture(dbs); + } + if (config instanceof ReadAsChannel) { + return (Projection) new FakeScatteringByteChannel(channelPayload); + } + if (config instanceof ReadAsSeekableChannel) { + return (Projection) new FakeSeekableByteChannel(seekablePayload); + } + throw new UnsupportedOperationException("Unexpected config: " + config); + } + + @Override + public void close() {} + }; + + ApiFuture wrapped = + BlobReadSessionAdapter.wrap(ApiFutures.immediateFuture(fakeSession), context, metrics); + try (BlobReadSession session = wrapped.get()) { + byte[] b1 = session.readAs(ReadProjectionConfigs.asFutureBytes()).get(); + assertThat(b1).hasLength(25); + + try (DisposableByteString dbs = + session.readAs(ReadProjectionConfigs.asFutureByteString()).get()) { + assertThat(dbs.byteString().size()).isEqualTo(15); + } + + // Partial read from ScatteringByteChannel (read 10 of 40 bytes) + try (ScatteringByteChannel ch = session.readAs(ReadProjectionConfigs.asChannel())) { + ByteBuffer dst = ByteBuffer.allocate(10); + assertThat(ch.read(dst)).isEqualTo(10); + } + + // Read 12 bytes from SeekableByteChannel + try (SeekableByteChannel seekable = + session.readAs(ReadProjectionConfigs.asSeekableChannel())) { + ByteBuffer dst = ByteBuffer.allocate(12); + assertThat(seekable.read(dst)).isEqualTo(12); + } + } + + Collection collected = reader.collectAllMetrics(); + HistogramPointData point = + getSingleHistogramPoint( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo(25.0 + 15.0 + 10.0 + 12.0); + } + + @Test + public void blobReadSessionAdapter_doesNotRecordResponseBodySize_whenOpenFails() { + InMemoryMetricReader reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + StorageClientMetrics metrics = StorageClientMetrics.create(provider, true, false); + StorageMetricsContext context = StorageMetricsContext.createForMultiStep("ReadObject"); + + ApiFuture failedOpen = + ApiFutures.immediateFailedFuture(new StorageException(404, "Not Found")); + ApiFuture wrapped = BlobReadSessionAdapter.wrap(failedOpen, context, metrics); + + assertThrows(ExecutionException.class, wrapped::get); + + Collection collected = reader.collectAllMetrics(); + assertThat( + findMetric( + collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE)) + .isNull(); + LongPointData opPoint = + getSinglePoint(collected, StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS); + assertThat(opPoint.getAttributes().get(StorageClientMetrics.KEY_ERROR_TYPE)) + .isEqualTo(StorageErrorTypeMapper.STATUS_NOT_FOUND); + } + + @Test + public void grpcStorageImpl_readAllBytes_recordsResponseBodySizeAndSuppressesOnMidStreamError() + throws Exception { + byte[] payload = "0123456789abcdefghij01234".getBytes(StandardCharsets.UTF_8); + ChecksummedTestContent chunk1 = ChecksummedTestContent.of(payload, 0, 10); + ChecksummedTestContent chunk2 = ChecksummedTestContent.of(payload, 10, 15); + com.google.storage.v2.Object meta = + com.google.storage.v2.Object.newBuilder() + .setBucket(BucketName.format("_", "b")) + .setName("ok") + .setGeneration(1L) + .setSize(payload.length) + .build(); + + StorageImplBase fakeService = + new StorageImplBase() { + @Override + public void readObject( + ReadObjectRequest request, StreamObserver responseObserver) { + if ("ok".equals(request.getObject())) { + responseObserver.onNext( + ReadObjectResponse.newBuilder() + .setMetadata(meta) + .setChecksummedData(chunk1.asChecksummedData()) + .build()); + responseObserver.onNext( + ReadObjectResponse.newBuilder() + .setChecksummedData(chunk2.asChecksummedData()) + .build()); + responseObserver.onCompleted(); + } else { + responseObserver.onNext( + ReadObjectResponse.newBuilder() + .setMetadata(meta.toBuilder().setName("fail").build()) + .setChecksummedData(chunk1.asChecksummedData()) + .build()); + responseObserver.onError(Status.PERMISSION_DENIED.asRuntimeException()); + } + } + }; + + InMemoryMetricReader okReader = InMemoryMetricReader.create(); + SdkMeterProvider okProvider = SdkMeterProvider.builder().registerMetricReader(okReader).build(); + InMemoryMetricReader failReader = InMemoryMetricReader.create(); + SdkMeterProvider failProvider = + SdkMeterProvider.builder().registerMetricReader(failReader).build(); + + try (FakeServer fakeServer = FakeServer.of(fakeService); + Storage okStorage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(1).build()) + .setMeterProvider(okProvider) + .setEnableOtelMetrics(true) + .build() + .getService(); + Storage failStorage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(1).build()) + .setMeterProvider(failProvider) + .setEnableOtelMetrics(true) + .build() + .getService()) { + byte[] actual = okStorage.readAllBytes(BlobId.of("b", "ok")); + assertThat(actual).isEqualTo(payload); + + HistogramPointData point = + getSingleHistogramPoint( + okReader.collectAllMetrics(), + StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(point.getCount()).isEqualTo(1L); + assertThat(point.getSum()).isEqualTo((double) payload.length); + + assertThrows(StorageException.class, () -> failStorage.readAllBytes(BlobId.of("b", "fail"))); + assertThat( + findMetric( + failReader.collectAllMetrics(), + StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE)) + .isNull(); + } + } + + @Test + public void grpcStorageImpl_downloadToAndBlobReadSession_recordResponseBodySize() + throws Exception { + byte[] payload = "0123456789abcdefghij0123456789".getBytes(StandardCharsets.UTF_8); + ChecksummedTestContent content = ChecksummedTestContent.of(payload); + com.google.storage.v2.Object meta = + com.google.storage.v2.Object.newBuilder() + .setBucket(BucketName.format("_", "b")) + .setName("o") + .setGeneration(1L) + .setSize(payload.length) + .build(); + + StorageImplBase fakeService = + new StorageImplBase() { + @Override + public void readObject( + ReadObjectRequest request, StreamObserver responseObserver) { + responseObserver.onNext( + ReadObjectResponse.newBuilder() + .setMetadata(meta) + .setChecksummedData(content.asChecksummedData()) + .build()); + responseObserver.onCompleted(); + } + + @Override + public StreamObserver bidiReadObject( + StreamObserver responseObserver) { + return new StreamObserver() { + @Override + public void onNext(BidiReadObjectRequest req) { + if (req.getReadRangesList().isEmpty()) { + responseObserver.onNext( + BidiReadObjectResponse.newBuilder().setMetadata(meta).build()); + } else { + ReadRange rr = req.getReadRangesList().get(0); + responseObserver.onNext( + BidiReadObjectResponse.newBuilder() + .addObjectDataRanges( + ObjectRangeData.newBuilder() + .setReadRange( + ReadRange.newBuilder() + .setReadId(rr.getReadId()) + .setReadOffset(0) + .setReadLength(payload.length) + .build()) + .setRangeEnd(true) + .setChecksummedData(content.asChecksummedData()) + .build()) + .build()); + } + } + + @Override + public void onError(Throwable t) { + responseObserver.onError(t); + } + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + }; + } + }; + + InMemoryMetricReader downloadReader = InMemoryMetricReader.create(); + SdkMeterProvider downloadProvider = + SdkMeterProvider.builder().registerMetricReader(downloadReader).build(); + InMemoryMetricReader sessionReader = InMemoryMetricReader.create(); + SdkMeterProvider sessionProvider = + SdkMeterProvider.builder().registerMetricReader(sessionReader).build(); + + try (FakeServer fakeServer = FakeServer.of(fakeService); + Storage downloadStorage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(1).build()) + .setMeterProvider(downloadProvider) + .setEnableOtelMetrics(true) + .build() + .getService(); + Storage sessionStorage = + fakeServer.getGrpcStorageOptions().toBuilder() + .setRetrySettings(RetrySettings.newBuilder().setMaxAttempts(1).build()) + .setMeterProvider(sessionProvider) + .setEnableOtelMetrics(true) + .build() + .getService()) { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + downloadStorage.downloadTo(BlobId.of("b", "o"), baos); + assertThat(baos.toByteArray()).isEqualTo(payload); + + HistogramPointData downloadPoint = + getSingleHistogramPoint( + downloadReader.collectAllMetrics(), + StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(downloadPoint.getCount()).isEqualTo(1L); + assertThat(downloadPoint.getSum()).isEqualTo((double) payload.length); + + try (BlobReadSession session = + sessionStorage.blobReadSession(BlobId.of("b", "o")).get(5, TimeUnit.SECONDS)) { + byte[] sessionBytes = + session.readAs(ReadProjectionConfigs.asFutureBytes()).get(5, TimeUnit.SECONDS); + assertThat(sessionBytes).isEqualTo(payload); + } + + HistogramPointData sessionPoint = + getSingleHistogramPoint( + sessionReader.collectAllMetrics(), + StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_RESPONSE_BODY_SIZE); + assertThat(sessionPoint.getCount()).isEqualTo(1L); + assertThat(sessionPoint.getSum()).isEqualTo((double) payload.length); + } + } + + private static final class FakeStorageReadChannel extends BaseStorageReadChannel { + private final byte[] data; + @Nullable private final StorageException errorAfterFirstRead; + + FakeStorageReadChannel( + byte[] data, + @Nullable StorageException errorAfterFirstRead, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { + super(s -> BlobInfo.newBuilder("b", "o").build(), metricsContext, storageClientMetrics); + this.data = data; + this.errorAfterFirstRead = errorAfterFirstRead; + } + + @Override + protected LazyReadChannel newLazyReadChannel() { + return new LazyReadChannel<>( + () -> + new BufferedReadableByteChannelSession() { + @Override + public ApiFuture openAsync() { + return ApiFutures.immediateFuture( + new BufferedReadableByteChannel() { + private int offset = 0; + private boolean open = true; + private int reads = 0; + + @Override + public int read(ByteBuffer dst) throws IOException { + reads++; + if (reads > 1 && errorAfterFirstRead != null) { + throw errorAfterFirstRead; + } + if (offset >= data.length) { + return -1; + } + int toCopy = Math.min(dst.remaining(), data.length - offset); + dst.put(data, offset, toCopy); + offset += toCopy; + return toCopy; + } + + @Override + public boolean isOpen() { + return open; + } + + @Override + public void close() { + open = false; + } + }); + } + + @Override + public ApiFuture getResult() { + return ApiFutures.immediateFuture("ok"); + } + }); + } + + @Override + public RestorableState capture() { + throw new UnsupportedOperationException(); + } + } + + private static final class FakeScatteringByteChannel implements ScatteringByteChannel { + private final byte[] data; + private int offset = 0; + private boolean open = true; + + FakeScatteringByteChannel(byte[] data) { + this.data = data; + } + + @Override + public int read(ByteBuffer dst) { + if (offset >= data.length) { + return -1; + } + int toCopy = Math.min(dst.remaining(), data.length - offset); + dst.put(data, offset, toCopy); + offset += toCopy; + return toCopy; + } + + @Override + public long read(ByteBuffer[] dsts, int off, int length) { + long total = 0; + for (int i = off; i < off + length; i++) { + int r = read(dsts[i]); + if (r == -1) { + return total == 0 ? -1 : total; + } + total += r; + } + return total; + } + + @Override + public long read(ByteBuffer[] dsts) { + return read(dsts, 0, dsts.length); + } + + @Override + public boolean isOpen() { + return open; + } + + @Override + public void close() { + open = false; + } + } + + private static final class FakeSeekableByteChannel implements SeekableByteChannel { + private final byte[] data; + private int position = 0; + private boolean open = true; + + FakeSeekableByteChannel(byte[] data) { + this.data = data; + } + + @Override + public int read(ByteBuffer dst) { + if (position >= data.length) { + return -1; + } + int toCopy = Math.min(dst.remaining(), data.length - position); + dst.put(data, position, toCopy); + position += toCopy; + return toCopy; + } + + @Override + public int write(ByteBuffer src) { + throw new UnsupportedOperationException(); + } + + @Override + public long position() { + return position; + } + + @Override + public SeekableByteChannel position(long newPosition) { + this.position = (int) newPosition; + return this; + } + + @Override + public long size() { + return data.length; + } + + @Override + public SeekableByteChannel truncate(long size) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean isOpen() { + return open; + } + + @Override + public void close() { + open = false; + } + } }