From 354f72b4423e6dbe2efbb859d77c31dac2a00efe Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Thu, 22 Oct 2026 05:49:55 +0000 Subject: [PATCH 1/3] feat(storage): wire operation metrics in gRPC veneer [Generated-by: AI] --- .../google/cloud/storage/GapicCopyWriter.java | 31 +- .../cloud/storage/GrpcBlobReadChannel.java | 14 +- .../cloud/storage/GrpcBlobWriteChannel.java | 16 +- .../google/cloud/storage/GrpcStorageImpl.java | 591 ++++++++++++++---- 4 files changed, 509 insertions(+), 143 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java index dff801871c51..48a2d44bfc80 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java @@ -23,6 +23,7 @@ import com.google.cloud.storage.Retrying.RetrierWithAlg; import com.google.storage.v2.RewriteObjectRequest; import com.google.storage.v2.RewriteResponse; +import org.checkerframework.checker.nullness.qual.Nullable; final class GapicCopyWriter extends CopyWriter { @@ -32,6 +33,7 @@ final class GapicCopyWriter extends CopyWriter { private final RetrierWithAlg retrier; private final RewriteObjectRequest originalRequest; private final RewriteResponse initialResponse; + @Nullable private final StorageMetricsContext metricsContext; private RewriteResponse mostRecentResponse; @@ -41,12 +43,23 @@ final class GapicCopyWriter extends CopyWriter { RetrierWithAlg retrier, RewriteObjectRequest originalRequest, RewriteResponse initialResponse) { + this(storage, callable, retrier, originalRequest, initialResponse, null); + } + + GapicCopyWriter( + GrpcStorageImpl storage, + UnaryCallable callable, + RetrierWithAlg retrier, + RewriteObjectRequest originalRequest, + RewriteResponse initialResponse, + @Nullable StorageMetricsContext metricsContext) { this.storage = storage; this.callable = callable; this.retrier = retrier; this.initialResponse = initialResponse; this.mostRecentResponse = initialResponse; this.originalRequest = originalRequest; + this.metricsContext = metricsContext; } @Override @@ -54,6 +67,9 @@ public Blob getResult() { while (!isDone()) { copyChunk(); } + if (metricsContext != null) { + metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), null); + } BlobInfo info = Conversions.grpc().blobInfo().decode(mostRecentResponse.getResource()); return info.asBlob(storage); } @@ -78,8 +94,19 @@ public void copyChunk() { if (!isDone()) { RewriteObjectRequest req = originalRequest.toBuilder().setRewriteToken(mostRecentResponse.getRewriteToken()).build(); - GrpcCallContext retryContext = Retrying.newCallContext(); - mostRecentResponse = retrier.run(() -> callable.call(req, retryContext), Decoder.identity()); + StorageMetricsContext chunkContext = + metricsContext != null + ? metricsContext.asSubAttemptContext() + : storage.getOptions().createMetricsContext("RewriteObject"); + GrpcCallContext retryContext = Retrying.newCallContext(chunkContext); + RewriteResponse resp = + retrier.run(() -> callable.call(req, retryContext), Decoder.identity(), chunkContext); + if (resp != null) { + mostRecentResponse = resp; + } + if (isDone() && metricsContext != null) { + metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), null); + } } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java index 9dd39dfe47ea..0b30d9f4376a 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java @@ -26,6 +26,7 @@ import com.google.storage.v2.ReadObjectRequest; import com.google.storage.v2.ReadObjectResponse; import org.checkerframework.checker.nullness.qual.NonNull; +import org.checkerframework.checker.nullness.qual.Nullable; final class GrpcBlobReadChannel extends BaseStorageReadChannel { @@ -41,7 +42,18 @@ final class GrpcBlobReadChannel extends BaseStorageReadChannel { ResultRetryAlgorithm resultRetryAlgorithm, ReadObjectRequest request, boolean autoGzipDecompression) { - super(Conversions.grpc().blobInfo()); + this(read, retrier, resultRetryAlgorithm, request, autoGzipDecompression, null, null); + } + + GrpcBlobReadChannel( + ZeroCopyServerStreamingCallable read, + Retrier retrier, + ResultRetryAlgorithm resultRetryAlgorithm, + ReadObjectRequest request, + boolean autoGzipDecompression, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { + super(Conversions.grpc().blobInfo(), metricsContext, storageClientMetrics); this.read = read; this.retrier = retrier; this.resultRetryAlgorithm = resultRetryAlgorithm; diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java index 3a7bc2a790c1..23c690ff7841 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java @@ -24,6 +24,7 @@ import com.google.storage.v2.WriteObjectRequest; import com.google.storage.v2.WriteObjectResponse; import java.util.function.Supplier; +import org.checkerframework.checker.nullness.qual.Nullable; final class GrpcBlobWriteChannel extends BaseStorageWriteChannel { @@ -37,7 +38,20 @@ final class GrpcBlobWriteChannel extends BaseStorageWriteChannel> start, Hasher hasher) { - super(Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource)); + this(write, retrier, start, hasher, null, null); + } + + GrpcBlobWriteChannel( + ClientStreamingCallable write, + RetrierWithAlg retrier, + Supplier> start, + Hasher hasher, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { + super( + Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource), + metricsContext, + storageClientMetrics); this.write = write; this.retrier = retrier; this.start = start; 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 120b7a269724..9160a671f260 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 @@ -209,6 +209,8 @@ public void close() throws Exception { s.shutdownNow(); java.time.Duration terminationAwaitDuration = getOptions().getTerminationAwaitDuration(); s.awaitTermination(terminationAwaitDuration.toMillis(), TimeUnit.MILLISECONDS); + } finally { + getOptions().closeStorageClientMetrics(); } } @@ -227,11 +229,13 @@ public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) { builder.getBucketBuilder().setProject(projectNameCodec.encode(getOptions().getProjectId())); } CreateBucketRequest req = opts.createBucketsRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("CreateBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.createBucketCallable().call(req, merge), - syntaxDecoders.bucket); + syntaxDecoders.bucket, + metricsContext); } @Override @@ -255,6 +259,8 @@ public Blob create( @Override public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) { + StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); + Throwable error = null; try { requireNonNull(blobInfo, "blobInfo must be non null"); InputStream inputStreamParam = firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)); @@ -264,7 +270,12 @@ public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... op optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); Hasher hasher = optsWithDefaults.getHasher(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + GrpcCallContext merge = + Utils.merge( + grpcCallContext, + metricsContext != null + ? Retrying.newCallContext(metricsContext) + : Retrying.newCallContext()); UnbufferedWritableByteChannelSession session = ResumableMedia.gapic() .write() @@ -282,7 +293,15 @@ public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... op ApiFuture responseApiFuture = session.getResult(); return this.getBlob(responseApiFuture); } catch (IOException | ApiException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @@ -307,37 +326,48 @@ public Blob internalCreateFrom(Path path, BlobInfo info, Opts o throw new StorageException(0, path + " is a directory"); } - GrpcCallContext grpcCallContext = - opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); - WriteObjectRequest req = getWriteObjectRequest(info, opts); - - ClientStreamingCallable write = - storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext); - - ApiFuture start = startResumableWrite(grpcCallContext, req, opts); - ApiFuture session2 = - ApiFutures.transform( - start, - rw -> - ResumableSession.grpc( - retrier.withAlg(retryAlgorithmManager.idempotent()), - write, - storageClient.queryWriteStatusCallable(), - rw, - opts.getHasher()), - MoreExecutors.directExecutor()); + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + Throwable error = null; try { - GrpcResumableSession got = session2.get(); - ResumableOperationResult<@Nullable Object> put = got.put(RewindableContent.of(path)); - Object object = put.getObject(); - if (object == null) { - // if by some odd chance the put didn't get the Object, query for it - ResumableOperationResult<@Nullable Object> query = got.query(); - object = query.getObject(); + GrpcCallContext grpcCallContext = + opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); + WriteObjectRequest req = getWriteObjectRequest(info, opts); + + ClientStreamingCallable write = + storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext); + + ApiFuture start = startResumableWrite(grpcCallContext, req, opts); + ApiFuture session2 = + ApiFutures.transform( + start, + rw -> + ResumableSession.grpc( + retrier.withAlg(retryAlgorithmManager.idempotent()), + write, + storageClient.queryWriteStatusCallable(), + rw, + opts.getHasher()), + MoreExecutors.directExecutor()); + try { + GrpcResumableSession got = session2.get(); + ResumableOperationResult<@Nullable Object> put = got.put(RewindableContent.of(path)); + Object object = put.getObject(); + if (object == null) { + // if by some odd chance the put didn't get the Object, query for it + ResumableOperationResult<@Nullable Object> query = got.query(); + object = query.getObject(); + } + return codecs.blobInfo().decode(object).asBlob(this); + } catch (InterruptedException | ExecutionException e) { + throw StorageException.coalesce(e); + } + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); } - return codecs.blobInfo().decode(object).asBlob(this); - } catch (InterruptedException | ExecutionException e) { - throw StorageException.coalesce(e); } } @@ -353,35 +383,46 @@ public Blob createFrom( throws IOException { requireNonNull(blobInfo, "blobInfo must be non null"); - Opts opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts); - GrpcCallContext grpcCallContext = - opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); - WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + Throwable error = null; + try { + Opts opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts); + GrpcCallContext grpcCallContext = + opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); + WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); - ApiFuture start = startResumableWrite(grpcCallContext, req, opts); - - BufferedWritableByteChannelSession session = - ResumableMedia.gapic() - .write() - .byteChannel( - storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext)) - .setHasher(opts.getHasher()) - .setByteStringStrategy(ByteStringStrategy.noCopy()) - .resumable() - .withRetryConfig(retrier.withAlg(retryAlgorithmManager.idempotent())) - .buffered(Buffers.allocateAligned(bufferSize, _256KiB)) - .setStartAsync(start) - .build(); + ApiFuture start = startResumableWrite(grpcCallContext, req, opts); - // Specifically not in the try-with, so we don't close the provided stream - ReadableByteChannel src = - Channels.newChannel(firstNonNull(in, new ByteArrayInputStream(ZERO_BYTES))); - try (BufferedWritableByteChannel dst = session.open()) { - ByteStreams.copy(src, dst); - } catch (Exception e) { - throw StorageException.coalesce(e); + BufferedWritableByteChannelSession session = + ResumableMedia.gapic() + .write() + .byteChannel( + storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext)) + .setHasher(opts.getHasher()) + .setByteStringStrategy(ByteStringStrategy.noCopy()) + .resumable() + .withRetryConfig(retrier.withAlg(retryAlgorithmManager.idempotent())) + .buffered(Buffers.allocateAligned(bufferSize, _256KiB)) + .setStartAsync(start) + .build(); + + // Specifically not in the try-with, so we don't close the provided stream + ReadableByteChannel src = + Channels.newChannel(firstNonNull(in, new ByteArrayInputStream(ZERO_BYTES))); + try (BufferedWritableByteChannel dst = session.open()) { + ByteStreams.copy(src, dst); + } catch (Exception e) { + throw StorageException.coalesce(e); + } + return getBlob(session.getResult()); + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } - return getBlob(session.getResult()); } @Override @@ -400,11 +441,14 @@ public Bucket lockRetentionPolicy(BucketInfo bucket, BucketTargetOption... optio .setBucket(bucketNameCodec.encode(bucket.getName())); LockBucketRetentionPolicyRequest req = opts.lockBucketRetentionPolicyRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = + getOptions().createMetricsContext("LockBucketRetentionPolicy"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.lockBucketRetentionPolicyCallable().call(req, merge), - syntaxDecoders.bucket); + syntaxDecoders.bucket, + metricsContext); } @Override @@ -439,14 +483,16 @@ private Blob internalObjectRestore(BlobId blobId, Opts opts) { .setObject(blobId.getName()); ifNonNull(blobId.getGeneration(), builder::setGeneration); RestoreObjectRequest req = finalOpts.restoreObjectRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("RestoreObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.restoreObjectCallable().call(req, merge), resp -> { BlobInfo tmp = codecs.blobInfo().decode(resp); return finalOpts.clearBlobFields().decode(tmp).asBlob(this); - }); + }, + metricsContext); } @Override @@ -465,7 +511,9 @@ public Page list(BucketListOption... options) { if (!request.getReturnPartialSuccess()) { try { - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ListBuckets"); + GrpcCallContext merge = + Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(request), () -> storageClient.listBucketsPagedCallable().call(request, merge), @@ -474,7 +522,9 @@ public Page list(BucketListOption... options) { resp.getPage(), syntaxDecoders.bucket.andThen(opts.clearBucketFields()), retrier, - retryAlgorithmManager.getFor(request))); + retryAlgorithmManager.getFor(request), + () -> metricsContext != null ? metricsContext.asSubAttemptContext() : null), + metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -497,11 +547,15 @@ public Page list(String bucket, BlobListOption... options) { ListObjectsRequest.newBuilder().setParent(bucketNameCodec.encode(bucket)); ListObjectsRequest req = opts.listObjectsRequest().apply(builder).build(); try { - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ListObjects"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.listObjectsCallable().call(req, merge), - resp -> new ListObjectsWithSyntheticDirectoriesPage(grpcCallContext, req, resp)); + resp -> + new ListObjectsWithSyntheticDirectoriesPage( + grpcCallContext, req, resp, metricsContext), + metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -526,11 +580,13 @@ public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) { .map(NamedField::getGrpcName) .collect(ImmutableList.toImmutableList())); UpdateBucketRequest req = builder.build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("UpdateBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.updateBucketCallable().call(req, merge), - syntaxDecoders.bucket); + syntaxDecoders.bucket, + metricsContext); } @Override @@ -552,11 +608,13 @@ public Blob update(BlobInfo blobInfo, BlobTargetOption... options) { .map(NamedField::getGrpcName) .collect(ImmutableList.toImmutableList())); UpdateObjectRequest req = builder.build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("UpdateObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.updateObjectCallable().call(req, merge), - syntaxDecoders.blob); + syntaxDecoders.blob, + metricsContext); } @Override @@ -572,7 +630,8 @@ public boolean delete(String bucket, BucketSourceOption... options) { DeleteBucketRequest.Builder builder = DeleteBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket)); DeleteBucketRequest req = opts.deleteBucketsRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("DeleteBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return Boolean.TRUE.equals( retrier.run( retryAlgorithmManager.getFor(req), @@ -584,7 +643,8 @@ public boolean delete(String bucket, BucketSourceOption... options) { return false; } }, - Decoder.identity())); + Decoder.identity(), + metricsContext)); } @Override @@ -624,14 +684,16 @@ public Void internalObjectDelete(BlobId id, Opts opts) { .setObject(id.getName()); ifNonNull(id.getGeneration(), builder::setGeneration); DeleteObjectRequest req = finalOpts.deleteObjectsRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("DeleteObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> { storageClient.deleteObjectCallable().call(req, merge); return null; }, - Decoder.identity()); + Decoder.identity(), + metricsContext); } @Override @@ -647,11 +709,13 @@ public Blob compose(ComposeRequest composeRequest) { builder.setDestination(target); builder.setDeleteSourceObjects(composeRequest.isDeleteSourceObjects()); ComposeObjectRequest req = opts.composeObjectsRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ComposeObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.composeObjectCallable().call(req, merge), - syntaxDecoders.blob); + syntaxDecoders.blob, + metricsContext); } @Override @@ -708,12 +772,26 @@ public CopyWriter copy(CopyRequest copyRequest) { GrpcCallContext grpcCallContext = grpcCallContextMapper.apply(GrpcCallContext.createDefault()); UnaryCallable callable = storageClient.rewriteObjectCallable().withDefaultCallContext(grpcCallContext); - GrpcCallContext retryContext = Retrying.newCallContext(); + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("RewriteObject"); + StorageMetricsContext initialContext = + multiContext != null + ? multiContext.asSubAttemptContext() + : getOptions().createMetricsContext("RewriteObject"); + GrpcCallContext retryContext = Retrying.newCallContext(initialContext); RetrierWithAlg retrierWithAlg = retrier.withAlg(retryAlgorithmManager.idempotent()); return retrier.run( retryAlgorithmManager.getFor(req), () -> callable.call(req, retryContext), - (resp) -> new GapicCopyWriter(this, callable, retrierWithAlg, req, resp)); + (resp) -> { + GapicCopyWriter writer = + new GapicCopyWriter(this, callable, retrierWithAlg, req, resp, multiContext); + if (resp.getDone() && multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), null); + } + return writer; + }, + initialContext); } @Override @@ -723,14 +801,25 @@ public byte[] readAllBytes(String bucket, String blob, BlobSourceOption... optio @Override public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) { - UnbufferedReadableByteChannelSession session = unbufferedReadSession(blob, options); + StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); + UnbufferedReadableByteChannelSession session = + unbufferedReadSession(blob, options, metricsContext); ByteArrayOutputStream baos = new ByteArrayOutputStream(); + Throwable error = null; try (UnbufferedReadableByteChannel r = session.open(); WritableByteChannel w = Channels.newChannel(baos)) { ByteStreams.copy(r, w); } catch (ApiException | IOException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } return baos.toByteArray(); } @@ -749,39 +838,66 @@ public GrpcBlobReadChannel reader(String bucket, String blob, BlobSourceOption.. public GrpcBlobReadChannel reader(BlobId blob, BlobSourceOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts); ReadObjectRequest request = getReadObjectRequest(blob, opts); - GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); + GrpcCallContext subAttemptContext = + metricsContext != null + ? Retrying.newCallContext(metricsContext.asSubAttemptContext()) + : Retrying.newCallContext(); + GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(subAttemptContext); return new GrpcBlobReadChannel( readObjectCallable(grpcCallContext), retrier, retryAlgorithmManager.getFor(request), request, - !opts.autoGzipDecompression()); + !opts.autoGzipDecompression(), + metricsContext, + getOptions().getStorageClientMetrics()); } @Override public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) { + StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); + UnbufferedReadableByteChannelSession session = + unbufferedReadSession(blob, options, metricsContext); - UnbufferedReadableByteChannelSession session = unbufferedReadSession(blob, options); - + Throwable error = null; try (UnbufferedReadableByteChannel r = session.open(); WritableByteChannel w = Files.newByteChannel(path, WRITE_OPS)) { ByteStreams.copy(r, w); } catch (ApiException | IOException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @Override public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) { + StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); + UnbufferedReadableByteChannelSession session = + unbufferedReadSession(blob, options, metricsContext); - UnbufferedReadableByteChannelSession session = unbufferedReadSession(blob, options); - + Throwable error = null; try (UnbufferedReadableByteChannel r = session.open(); WritableByteChannel w = Channels.newChannel(outputStream)) { ByteStreams.copy(r, w); } catch (ApiException | IOException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @@ -792,10 +908,16 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); Hasher hasher = opts.getHasher(); + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + GrpcCallContext startContext = + multiContext != null + ? Retrying.newCallContext(multiContext.asSubAttemptContext()) + : Retrying.newCallContext(); + GrpcCallContext mergedStart = Utils.merge(grpcCallContext, startContext); // in JSON, the starting of the resumable session happens before the invocation of write can // happen. Emulate the same thing here. // 1. create the future - ApiFuture startResumableWrite = startResumableWrite(grpcCallContext, req, opts); + ApiFuture startResumableWrite = startResumableWrite(mergedStart, req, opts); // 2. await the result of the future ResumableWrite resumableWrite = ApiFutureUtils.await(startResumableWrite); // 3. wrap the result in another future container before constructing the BlobWriteChannel @@ -804,7 +926,9 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext), retrier.withAlg(retryAlgorithmManager.idempotent()), () -> wrapped, - hasher); + hasher, + multiContext, + getOptions().getStorageClientMetrics()); } @Override @@ -817,7 +941,8 @@ public BlobInfo internalDirectUpload( optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); Hasher hasher = opts.getHasher(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); RewindableContent content = RewindableContent.of(buf); return retrier.run( retryAlgorithmManager.getFor(req), @@ -839,7 +964,8 @@ public BlobInfo internalDirectUpload( } return session.getResult(); }, - this::getBlob); + this::getBlob, + metricsContext); } @Override @@ -950,9 +1076,10 @@ public List delete(Iterable blobIds) { @Override public Acl getAcl(String bucket, Entity entity, BucketSourceOption... options) { + StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetBucketACL"); try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, metricsContext); Predicate entityPredicate = bucketAclEntityOrAltEq(codecs.entity().encode(entity)); @@ -980,9 +1107,14 @@ public Acl getAcl(String bucket, Entity entity) { @Override public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options) { + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("DeleteBucketACL"); + StorageMetricsContext subContext = + multiContext != null ? multiContext.asSubAttemptContext() : null; + Throwable error = null; try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, subContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = bucketAclEntityOrAltEq(encode); @@ -1000,7 +1132,7 @@ public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... opt UpdateBucketRequest req = createUpdateBucketAclRequest(bucket, newAcls, metageneration); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getAclList().stream().filter(entityPredicate).findFirst(); @@ -1012,8 +1144,16 @@ public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... opt if (se.getCode() == 404) { return false; } else { + error = se; throw se; } + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @@ -1024,7 +1164,7 @@ public boolean deleteAcl(String bucket, Entity entity) { @Override public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) { - return updateAcl(bucket, acl, options); + return internalUpdateAcl("CreateBucketACL", bucket, acl, options); } @Override @@ -1034,9 +1174,18 @@ public Acl createAcl(String bucket, Acl acl) { @Override public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { + return internalUpdateAcl("UpdateBucketACL", bucket, acl, options); + } + + private Acl internalUpdateAcl( + String operationName, String bucket, Acl acl, BucketSourceOption... options) { + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext(operationName); + StorageMetricsContext subContext = + multiContext != null ? multiContext.asSubAttemptContext() : null; + Throwable error = null; try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, subContext); BucketAccessControl encode = codecs.bucketAcl().encode(acl); String entity = encode.getEntity(); @@ -1050,7 +1199,7 @@ public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { UpdateBucketRequest req = createUpdateBucketAclRequest(bucket, newDefaultAcls, resp.getMetageneration()); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); Optional first = updateResult.getAclList().stream() @@ -1061,7 +1210,15 @@ public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { return first.orElseThrow( () -> new StorageException(0, "Acl update call success, but not in response")); } catch (NotFoundException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @@ -1072,9 +1229,10 @@ public Acl updateAcl(String bucket, Acl acl) { @Override public List listAcls(String bucket, BucketSourceOption... options) { + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ListBucketACLs"); try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, metricsContext); return resp.getAclList().stream() .map(codecs.bucketAcl()::decode) .collect(ImmutableList.toImmutableList()); @@ -1090,8 +1248,9 @@ public List listAcls(String bucket) { @Override public Acl getDefaultAcl(String bucket, Entity entity) { + StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetDefaultObjectACL"); try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, metricsContext); Predicate entityPredicate = objectAclEntityOrAltEq(codecs.entity().encode(entity)); @@ -1114,8 +1273,13 @@ public Acl getDefaultAcl(String bucket, Entity entity) { @Override public boolean deleteDefaultAcl(String bucket, Entity entity) { + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("DeleteDefaultObjectACL"); + StorageMetricsContext subContext = + multiContext != null ? multiContext.asSubAttemptContext() : null; + Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, subContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1134,7 +1298,7 @@ public boolean deleteDefaultAcl(String bucket, Entity entity) { UpdateBucketRequest req = createUpdateDefaultAclRequest(bucket, newDefaultAcls, metageneration); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getDefaultObjectAclList().stream().filter(entityPredicate).findFirst(); @@ -1146,20 +1310,36 @@ public boolean deleteDefaultAcl(String bucket, Entity entity) { if (se.getCode() == 404) { return false; } else { + error = se; throw se; } + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @Override public Acl createDefaultAcl(String bucket, Acl acl) { - return updateDefaultAcl(bucket, acl); + return internalUpdateDefaultAcl("CreateDefaultObjectACL", bucket, acl); } @Override public Acl updateDefaultAcl(String bucket, Acl acl) { + return internalUpdateDefaultAcl("UpdateDefaultObjectACL", bucket, acl); + } + + private Acl internalUpdateDefaultAcl(String operationName, String bucket, Acl acl) { + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext(operationName); + StorageMetricsContext subContext = + multiContext != null ? multiContext.asSubAttemptContext() : null; + Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, subContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1174,7 +1354,7 @@ public Acl updateDefaultAcl(String bucket, Acl acl) { UpdateBucketRequest req = createUpdateDefaultAclRequest(bucket, newDefaultAcls, resp.getMetageneration()); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); Optional first = updateResult.getDefaultObjectAclList().stream() @@ -1185,14 +1365,24 @@ public Acl updateDefaultAcl(String bucket, Acl acl) { return first.orElseThrow( () -> new StorageException(0, "Acl update call success, but not in response")); } catch (NotFoundException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @Override public List listDefaultAcls(String bucket) { + StorageMetricsContext metricsContext = + getOptions().createMetricsContext("ListDefaultObjectACLs"); try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, metricsContext); return resp.getDefaultObjectAclList().stream() .map(codecs.objectAcl()::decode) .collect(ImmutableList.toImmutableList()); @@ -1203,9 +1393,10 @@ public List listDefaultAcls(String bucket) { @Override public Acl getAcl(BlobId blob, Entity entity) { + StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetObjectACL"); try { Object req = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(req); + Object resp = getObjectWithAcls(req, metricsContext); Predicate entityPredicate = objectAclEntityOrAltEq(codecs.entity().encode(entity)); @@ -1228,9 +1419,14 @@ public Acl getAcl(BlobId blob, Entity entity) { @Override public boolean deleteAcl(BlobId blob, Entity entity) { + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("DeleteObjectACL"); + StorageMetricsContext subContext = + multiContext != null ? multiContext.asSubAttemptContext() : null; + Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj); + Object resp = getObjectWithAcls(obj, subContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1248,7 +1444,7 @@ public boolean deleteAcl(BlobId blob, Entity entity) { UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, metageneration); - Object updateResult = updateObject(req); + Object updateResult = updateObject(req, subContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getAclList().stream().filter(entityPredicate).findFirst(); @@ -1260,21 +1456,37 @@ public boolean deleteAcl(BlobId blob, Entity entity) { if (se.getCode() == 404) { return false; } else { + error = se; throw se; } + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @Override public Acl createAcl(BlobId blob, Acl acl) { - return updateAcl(blob, acl); + return internalUpdateObjectAcl("CreateObjectACL", blob, acl); } @Override public Acl updateAcl(BlobId blob, Acl acl) { + return internalUpdateObjectAcl("UpdateObjectACL", blob, acl); + } + + private Acl internalUpdateObjectAcl(String operationName, BlobId blob, Acl acl) { + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext(operationName); + StorageMetricsContext subContext = + multiContext != null ? multiContext.asSubAttemptContext() : null; + Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj); + Object resp = getObjectWithAcls(obj, subContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1288,7 +1500,7 @@ public Acl updateAcl(BlobId blob, Acl acl) { UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, resp.getMetageneration()); - Object updateResult = updateObject(req); + Object updateResult = updateObject(req, subContext); Optional first = updateResult.getAclList().stream() @@ -1299,15 +1511,24 @@ public Acl updateAcl(BlobId blob, Acl acl) { return first.orElseThrow( () -> new StorageException(0, "Acl update call success, but not in response")); } catch (NotFoundException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @Override public List listAcls(BlobId blob) { + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ListObjectACLs"); try { Object req = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(req); + Object resp = getObjectWithAcls(req, metricsContext); return resp.getAclList().stream() .map(codecs.objectAcl()::decode) .collect(ImmutableList.toImmutableList()); @@ -1350,11 +1571,13 @@ public Policy getIamPolicy(String bucket, BucketSourceOption... options) { GetIamPolicyRequest.Builder builder = GetIamPolicyRequest.newBuilder().setResource(bucketNameCodec.encode(bucket)); GetIamPolicyRequest req = opts.getIamPolicyRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetIamPolicy"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getIamPolicyCallable().call(req, merge), - codecs.policyCodec()); + codecs.policyCodec(), + metricsContext); } @Override @@ -1367,11 +1590,13 @@ public Policy setIamPolicy(String bucket, Policy policy, BucketSourceOption... o .setResource(bucketNameCodec.encode(bucket)) .setPolicy(codecs.policyCodec().encode(policy)) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("SetIamPolicy"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.setIamPolicyCallable().call(req, merge), - codecs.policyCodec()); + codecs.policyCodec(), + metricsContext); } @Override @@ -1385,7 +1610,8 @@ public List testIamPermissions( .setResource(bucketNameCodec.encode(bucket)) .addAllPermissions(permissions) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("TestIamPermissions"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.testIamPermissionsCallable().call(req, merge), @@ -1394,7 +1620,8 @@ public List testIamPermissions( return permissions.stream() .map(heldPermissions::contains) .collect(ImmutableList.toImmutableList()); - }); + }, + metricsContext); } @Override @@ -1458,10 +1685,12 @@ public Blob moveBlob(MoveBlobRequest request) { dstOpts.moveObjectsRequest().apply(b); MoveObjectRequest req = b.build(); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("MoveObject"); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.moveObjectCallable().call(req), - syntaxDecoders.blob); + syntaxDecoders.blob, + metricsContext); } @Override @@ -1562,12 +1791,22 @@ private final class ListObjectsWithSyntheticDirectoriesPage implements Page getNextPage() { ListObjectsRequest nextPageReq = req.toBuilder().setPageToken(resp.getNextPageToken()).build(); try { - GrpcCallContext merge = Utils.merge(ctx, Retrying.newCallContext()); + StorageMetricsContext metricsContext = + parentMetricsContext != null + ? parentMetricsContext.asSubAttemptContext() + : getOptions().createMetricsContext("ListObjects"); + GrpcCallContext merge = Utils.merge(ctx, Retrying.newCallContext(metricsContext)); ListObjectsResponse nextPageResp = retrier.run( retryAlgorithmManager.getFor(nextPageReq), () -> storageClient.listObjectsCallable().call(nextPageReq, merge), - Decoder.identity()); - return new ListObjectsWithSyntheticDirectoriesPage(ctx, nextPageReq, nextPageResp); + Decoder.identity(), + metricsContext); + return new ListObjectsWithSyntheticDirectoriesPage( + ctx, nextPageReq, nextPageResp, parentMetricsContext); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -1715,16 +1960,27 @@ static final class TransformingPageDecorator< private final Decoder translator; private final Retrier retrier; private final ResultRetryAlgorithm resultRetryAlgorithm; + private final Supplier metricsContextSupplier; TransformingPageDecorator( PageT page, Decoder translator, Retrier retrier, ResultRetryAlgorithm resultRetryAlgorithm) { + this(page, translator, retrier, resultRetryAlgorithm, () -> null); + } + + TransformingPageDecorator( + PageT page, + Decoder translator, + Retrier retrier, + ResultRetryAlgorithm resultRetryAlgorithm, + Supplier metricsContextSupplier) { this.page = page; this.translator = translator; this.retrier = retrier; this.resultRetryAlgorithm = resultRetryAlgorithm; + this.metricsContextSupplier = metricsContextSupplier; } @Override @@ -1740,7 +1996,7 @@ public String getNextPageToken() { @Override public Page getNextPage() { return new TransformingPageDecorator<>( - page.getNextPage(), translator, retrier, resultRetryAlgorithm); + page.getNextPage(), translator, retrier, resultRetryAlgorithm, metricsContextSupplier); } @SuppressWarnings({"Convert2MethodRef"}) @@ -1760,7 +2016,8 @@ public Iterable iterateAll() { // prevent a javac 1.8 exception // https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8056984 Callable c = () -> prev.getNextPage(); - return retrier.run(resultRetryAlgorithm, c, Decoder.identity()); + return retrier.run( + resultRetryAlgorithm, c, Decoder.identity(), metricsContextSupplier.get()); }) .filter(Objects::nonNull) .flatMap(p -> StreamSupport.stream(p.getValues().spliterator(), false)) @@ -1894,10 +2151,19 @@ BidiWriteObjectRequest getBidiWriteObjectRequestForTakeover( private UnbufferedReadableByteChannelSession unbufferedReadSession( BlobId blob, BlobSourceOption[] options) { + return unbufferedReadSession(blob, options, null); + } + + private UnbufferedReadableByteChannelSession unbufferedReadSession( + BlobId blob, BlobSourceOption[] options, @Nullable StorageMetricsContext metricsContext) { Opts opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts); ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts); - GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(Retrying.newCallContext()); + GrpcCallContext subAttemptContext = + metricsContext != null + ? Retrying.newCallContext(metricsContext.asSubAttemptContext()) + : Retrying.newCallContext(); + GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(subAttemptContext); return ResumableMedia.gapic() .read() .byteChannel( @@ -1944,14 +2210,21 @@ private SourceObject sourceObjectEncode(SourceBlob from) { private com.google.storage.v2.ListBucketsResponse listBuckets( GrpcCallContext grpcCallContext, ListBucketsRequest request) { - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ListBuckets"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(request), () -> storageClient.listBucketsCallable().call(request, merge), - Decoder.identity()); + Decoder.identity(), + metricsContext); } private com.google.storage.v2.Bucket getBucketWithDefaultAcls(String bucketName) { + return getBucketWithDefaultAcls(bucketName, null); + } + + private com.google.storage.v2.Bucket getBucketWithDefaultAcls( + String bucketName, @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields( ImmutableSet.of( @@ -1966,15 +2239,25 @@ private com.google.storage.v2.Bucket getBucketWithDefaultAcls(String bucketName) .setName(bucketNameCodec.encode(bucketName)) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null ? metricsContext : getOptions().createMetricsContext("GetBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } private com.google.storage.v2.Bucket getBucketWithAcls( String bucketName, Opts opts) { + return getBucketWithAcls(bucketName, opts, null); + } + + private com.google.storage.v2.Bucket getBucketWithAcls( + String bucketName, + Opts opts, + @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields(ImmutableSet.of(BucketField.ACL, BucketField.METAGENERATION)); GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); @@ -1985,20 +2268,31 @@ private com.google.storage.v2.Bucket getBucketWithAcls( .setName(bucketNameCodec.encode(bucketName)) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null ? metricsContext : getOptions().createMetricsContext("GetBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } private com.google.storage.v2.Bucket updateBucket(UpdateBucketRequest req) { + return updateBucket(req, null); + } + + private com.google.storage.v2.Bucket updateBucket( + UpdateBucketRequest req, @Nullable StorageMetricsContext metricsContext) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null ? metricsContext : getOptions().createMetricsContext("UpdateBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.updateBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } private static UpdateBucketRequest createUpdateDefaultAclRequest( @@ -2036,6 +2330,10 @@ private static UpdateBucketRequest createUpdateBucketAclRequest( } private Object getObjectWithAcls(Object obj) { + return getObjectWithAcls(obj, null); + } + + private Object getObjectWithAcls(Object obj, @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields(ImmutableSet.of(BucketField.ACL, BucketField.METAGENERATION)); GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); @@ -2047,11 +2345,14 @@ private Object getObjectWithAcls(Object obj) { .setObject(obj.getName()) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null ? metricsContext : getOptions().createMetricsContext("GetObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getObjectCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } private static UpdateObjectRequest createUpdateObjectAclRequest( @@ -2073,12 +2374,20 @@ private static UpdateObjectRequest createUpdateObjectAclRequest( } private Object updateObject(UpdateObjectRequest req) { + return updateObject(req, null); + } + + private Object updateObject( + UpdateObjectRequest req, @Nullable StorageMetricsContext metricsContext) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null ? metricsContext : getOptions().createMetricsContext("UpdateObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.updateObjectCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } @NonNull @@ -2093,7 +2402,8 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { .setObject(blobId.getName()); ifNonNull(blobId.getGeneration(), builder::setGeneration); GetObjectRequest req = finalOpts.getObjectsRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); //noinspection DataFlowIssue return retrier.run( retryAlgorithmManager.getFor(req), @@ -2101,7 +2411,8 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { resp -> { BlobInfo tmp = codecs.blobInfo().decode(resp); return finalOpts.clearBlobFields().decode(tmp); - }); + }, + metricsContext); } @Nullable @@ -2128,7 +2439,8 @@ private Bucket internalBucketGet(String bucket, Opts unwrap) { GetBucketRequest.Builder builder = GetBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket)); GetBucketRequest req = opts.getBucketsRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> { @@ -2138,7 +2450,8 @@ private Bucket internalBucketGet(String bucket, Opts unwrap) { return null; } }, - syntaxDecoders.bucket.andThen(opts.clearBucketFields())); + syntaxDecoders.bucket.andThen(opts.clearBucketFields()), + metricsContext); } private ZeroCopyServerStreamingCallable readObjectCallable( From 93b51b5dbc3be0eb2f1134e9cd641ff90b2803e6 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Thu, 22 Oct 2026 07:11:20 +0000 Subject: [PATCH 2/3] fix(storage): address PR #14729 review comments on gRPC veneer metrics [Generated-by: AI] --- .../google/cloud/storage/GapicCopyWriter.java | 57 +++++++++++------ .../google/cloud/storage/GrpcStorageImpl.java | 61 ++++++++++++------- 2 files changed, 76 insertions(+), 42 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java index 48a2d44bfc80..c5d3ed703388 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java @@ -34,6 +34,7 @@ final class GapicCopyWriter extends CopyWriter { private final RewriteObjectRequest originalRequest; private final RewriteResponse initialResponse; @Nullable private final StorageMetricsContext metricsContext; + private boolean operationRecorded; private RewriteResponse mostRecentResponse; @@ -60,15 +61,26 @@ final class GapicCopyWriter extends CopyWriter { this.mostRecentResponse = initialResponse; this.originalRequest = originalRequest; this.metricsContext = metricsContext; + this.operationRecorded = initialResponse.getDone(); + } + + private void recordOperation(@Nullable Throwable error) { + if (metricsContext != null && !operationRecorded) { + metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), error); + operationRecorded = true; + } } @Override public Blob getResult() { - while (!isDone()) { - copyChunk(); - } - if (metricsContext != null) { - metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), null); + try { + while (!isDone()) { + copyChunk(); + } + recordOperation(null); + } catch (Throwable t) { + recordOperation(t); + throw t; } BlobInfo info = Conversions.grpc().blobInfo().decode(mostRecentResponse.getResource()); return info.asBlob(storage); @@ -92,20 +104,27 @@ public long getTotalBytesCopied() { @Override public void copyChunk() { if (!isDone()) { - RewriteObjectRequest req = - originalRequest.toBuilder().setRewriteToken(mostRecentResponse.getRewriteToken()).build(); - StorageMetricsContext chunkContext = - metricsContext != null - ? metricsContext.asSubAttemptContext() - : storage.getOptions().createMetricsContext("RewriteObject"); - GrpcCallContext retryContext = Retrying.newCallContext(chunkContext); - RewriteResponse resp = - retrier.run(() -> callable.call(req, retryContext), Decoder.identity(), chunkContext); - if (resp != null) { - mostRecentResponse = resp; - } - if (isDone() && metricsContext != null) { - metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), null); + try { + RewriteObjectRequest req = + originalRequest.toBuilder() + .setRewriteToken(mostRecentResponse.getRewriteToken()) + .build(); + StorageMetricsContext chunkContext = + metricsContext != null + ? metricsContext.asSubAttemptContext() + : storage.getOptions().createMetricsContext("RewriteObject"); + GrpcCallContext retryContext = Retrying.newCallContext(chunkContext); + RewriteResponse resp = + retrier.run(() -> callable.call(req, retryContext), Decoder.identity(), chunkContext); + if (resp != null) { + mostRecentResponse = resp; + } + if (isDone()) { + recordOperation(null); + } + } catch (Throwable t) { + recordOperation(t); + throw t; } } } 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 9160a671f260..4e4100a4a5d3 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 @@ -780,18 +780,25 @@ public CopyWriter copy(CopyRequest copyRequest) { : getOptions().createMetricsContext("RewriteObject"); GrpcCallContext retryContext = Retrying.newCallContext(initialContext); RetrierWithAlg retrierWithAlg = retrier.withAlg(retryAlgorithmManager.idempotent()); - return retrier.run( - retryAlgorithmManager.getFor(req), - () -> callable.call(req, retryContext), - (resp) -> { - GapicCopyWriter writer = - new GapicCopyWriter(this, callable, retrierWithAlg, req, resp, multiContext); - if (resp.getDone() && multiContext != null) { - multiContext.recordOperation(getOptions().getStorageClientMetrics(), null); - } - return writer; - }, - initialContext); + try { + return retrier.run( + retryAlgorithmManager.getFor(req), + () -> callable.call(req, retryContext), + (resp) -> { + GapicCopyWriter writer = + new GapicCopyWriter(this, callable, retrierWithAlg, req, resp, multiContext); + if (resp.getDone() && multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), null); + } + return writer; + }, + initialContext); + } catch (Throwable t) { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } } @Override @@ -909,17 +916,25 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); Hasher hasher = opts.getHasher(); StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); - GrpcCallContext startContext = - multiContext != null - ? Retrying.newCallContext(multiContext.asSubAttemptContext()) - : Retrying.newCallContext(); - GrpcCallContext mergedStart = Utils.merge(grpcCallContext, startContext); - // in JSON, the starting of the resumable session happens before the invocation of write can - // happen. Emulate the same thing here. - // 1. create the future - ApiFuture startResumableWrite = startResumableWrite(mergedStart, req, opts); - // 2. await the result of the future - ResumableWrite resumableWrite = ApiFutureUtils.await(startResumableWrite); + ResumableWrite resumableWrite; + try { + GrpcCallContext startContext = + multiContext != null + ? Retrying.newCallContext(multiContext.asSubAttemptContext()) + : Retrying.newCallContext(); + GrpcCallContext mergedStart = Utils.merge(grpcCallContext, startContext); + // in JSON, the starting of the resumable session happens before the invocation of write can + // happen. Emulate the same thing here. + // 1. create the future + ApiFuture startResumableWrite = startResumableWrite(mergedStart, req, opts); + // 2. await the result of the future + resumableWrite = ApiFutureUtils.await(startResumableWrite); + } catch (Throwable t) { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } // 3. wrap the result in another future container before constructing the BlobWriteChannel ApiFuture wrapped = ApiFutures.immediateFuture(resumableWrite); return new GrpcBlobWriteChannel( From bb8cd418643eadff2dde094c74ba0149a337b9c9 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Thu, 22 Oct 2026 11:10:25 +0000 Subject: [PATCH 3/3] fix(storage): refine gRPC veneer operation metrics and attempt tracking --- .../google/cloud/storage/GapicCopyWriter.java | 7 +- .../cloud/storage/GrpcBlobReadChannel.java | 3 +- .../google/cloud/storage/GrpcStorageImpl.java | 227 ++++++------------ 3 files changed, 84 insertions(+), 153 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java index c5d3ed703388..21c1aad1b59e 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GapicCopyWriter.java @@ -34,7 +34,6 @@ final class GapicCopyWriter extends CopyWriter { private final RewriteObjectRequest originalRequest; private final RewriteResponse initialResponse; @Nullable private final StorageMetricsContext metricsContext; - private boolean operationRecorded; private RewriteResponse mostRecentResponse; @@ -61,13 +60,11 @@ final class GapicCopyWriter extends CopyWriter { this.mostRecentResponse = initialResponse; this.originalRequest = originalRequest; this.metricsContext = metricsContext; - this.operationRecorded = initialResponse.getDone(); } private void recordOperation(@Nullable Throwable error) { - if (metricsContext != null && !operationRecorded) { + if (metricsContext != null) { metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), error); - operationRecorded = true; } } @@ -111,7 +108,7 @@ public void copyChunk() { .build(); StorageMetricsContext chunkContext = metricsContext != null - ? metricsContext.asSubAttemptContext() + ? metricsContext : storage.getOptions().createMetricsContext("RewriteObject"); GrpcCallContext retryContext = Retrying.newCallContext(chunkContext); RewriteResponse resp = diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java index 0b30d9f4376a..927ac7c65e04 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java @@ -73,7 +73,8 @@ protected LazyReadChannel newLazyReadChannel() { ReadableByteChannelSessionBuilder b = ResumableMedia.gapic() .read() - .byteChannel(read, retrier, resultRetryAlgorithm) + .byteChannel( + read, retrier.withMetricsContext(getMetricsContext()), resultRetryAlgorithm) .setHasher(Hasher.readHasher()) .setAutoGzipDecompression(autoGzipDecompression); BufferHandle bufferHandle = getBufferHandle(); 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 4e4100a4a5d3..39152543420e 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 @@ -259,50 +259,38 @@ public Blob create( @Override public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) { - StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); - Throwable error = null; - try { - requireNonNull(blobInfo, "blobInfo must be non null"); - InputStream inputStreamParam = firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)); + requireNonNull(blobInfo, "blobInfo must be non null"); + InputStream inputStreamParam = firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)); - Opts optsWithDefaults = Opts.unwrap(options).prepend(defaultOpts); - GrpcCallContext grpcCallContext = - optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); - WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); - Hasher hasher = optsWithDefaults.getHasher(); - GrpcCallContext merge = - Utils.merge( - grpcCallContext, - metricsContext != null - ? Retrying.newCallContext(metricsContext) - : Retrying.newCallContext()); - UnbufferedWritableByteChannelSession session = - ResumableMedia.gapic() - .write() - .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge)) - .setByteStringStrategy(ByteStringStrategy.noCopy()) - .setHasher(hasher) - .direct() - .unbuffered() - .setRequest(req) - .build(); + Opts optsWithDefaults = Opts.unwrap(options).prepend(defaultOpts); + GrpcCallContext grpcCallContext = + optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); + WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); + Hasher hasher = optsWithDefaults.getHasher(); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("WriteObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + return retrier.run( + Retrying.neverRetry(), + () -> { + UnbufferedWritableByteChannelSession session = + ResumableMedia.gapic() + .write() + .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge)) + .setByteStringStrategy(ByteStringStrategy.noCopy()) + .setHasher(hasher) + .direct() + .unbuffered() + .setRequest(req) + .build(); - try (UnbufferedWritableByteChannel c = session.open()) { - ByteStreams.copy(Channels.newChannel(inputStreamParam), c); - } - ApiFuture responseApiFuture = session.getResult(); - return this.getBlob(responseApiFuture); - } catch (IOException | ApiException e) { - error = e; - throw StorageException.coalesce(e); - } catch (Exception e) { - error = e; - throw e; - } finally { - if (metricsContext != null) { - metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); - } - } + try (UnbufferedWritableByteChannel c = session.open()) { + ByteStreams.copy(Channels.newChannel(inputStreamParam), c); + } + ApiFuture responseApiFuture = session.getResult(); + return this.getBlob(responseApiFuture); + }, + Decoder.identity(), + metricsContext); } @Override @@ -336,13 +324,14 @@ public Blob internalCreateFrom(Path path, BlobInfo info, Opts o ClientStreamingCallable write = storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext); - ApiFuture start = startResumableWrite(grpcCallContext, req, opts); + ApiFuture start = + startResumableWrite(grpcCallContext, req, opts, multiContext); ApiFuture session2 = ApiFutures.transform( start, rw -> ResumableSession.grpc( - retrier.withAlg(retryAlgorithmManager.idempotent()), + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext), write, storageClient.queryWriteStatusCallable(), rw, @@ -391,7 +380,8 @@ public Blob createFrom( opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); - ApiFuture start = startResumableWrite(grpcCallContext, req, opts); + ApiFuture start = + startResumableWrite(grpcCallContext, req, opts, multiContext); BufferedWritableByteChannelSession session = ResumableMedia.gapic() @@ -401,7 +391,7 @@ public Blob createFrom( .setHasher(opts.getHasher()) .setByteStringStrategy(ByteStringStrategy.noCopy()) .resumable() - .withRetryConfig(retrier.withAlg(retryAlgorithmManager.idempotent())) + .withRetryConfig(retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext)) .buffered(Buffers.allocateAligned(bufferSize, _256KiB)) .setStartAsync(start) .build(); @@ -522,8 +512,7 @@ public Page list(BucketListOption... options) { resp.getPage(), syntaxDecoders.bucket.andThen(opts.clearBucketFields()), retrier, - retryAlgorithmManager.getFor(request), - () -> metricsContext != null ? metricsContext.asSubAttemptContext() : null), + retryAlgorithmManager.getFor(request)), metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); @@ -552,9 +541,7 @@ public Page list(String bucket, BlobListOption... options) { return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.listObjectsCallable().call(req, merge), - resp -> - new ListObjectsWithSyntheticDirectoriesPage( - grpcCallContext, req, resp, metricsContext), + resp -> new ListObjectsWithSyntheticDirectoriesPage(grpcCallContext, req, resp), metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); @@ -775,9 +762,7 @@ public CopyWriter copy(CopyRequest copyRequest) { StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("RewriteObject"); StorageMetricsContext initialContext = - multiContext != null - ? multiContext.asSubAttemptContext() - : getOptions().createMetricsContext("RewriteObject"); + multiContext != null ? multiContext : getOptions().createMetricsContext("RewriteObject"); GrpcCallContext retryContext = Retrying.newCallContext(initialContext); RetrierWithAlg retrierWithAlg = retrier.withAlg(retryAlgorithmManager.idempotent()); try { @@ -848,7 +833,7 @@ public GrpcBlobReadChannel reader(BlobId blob, BlobSourceOption... options) { StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); GrpcCallContext subAttemptContext = metricsContext != null - ? Retrying.newCallContext(metricsContext.asSubAttemptContext()) + ? Retrying.newCallContext(metricsContext) : Retrying.newCallContext(); GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(subAttemptContext); @@ -918,15 +903,11 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); ResumableWrite resumableWrite; try { - GrpcCallContext startContext = - multiContext != null - ? Retrying.newCallContext(multiContext.asSubAttemptContext()) - : Retrying.newCallContext(); - GrpcCallContext mergedStart = Utils.merge(grpcCallContext, startContext); // in JSON, the starting of the resumable session happens before the invocation of write can // happen. Emulate the same thing here. // 1. create the future - ApiFuture startResumableWrite = startResumableWrite(mergedStart, req, opts); + ApiFuture startResumableWrite = + startResumableWrite(grpcCallContext, req, opts, multiContext); // 2. await the result of the future resumableWrite = ApiFutureUtils.await(startResumableWrite); } catch (Throwable t) { @@ -939,7 +920,7 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options ApiFuture wrapped = ApiFutures.immediateFuture(resumableWrite); return new GrpcBlobWriteChannel( storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext), - retrier.withAlg(retryAlgorithmManager.idempotent()), + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext), () -> wrapped, hasher, multiContext, @@ -1124,12 +1105,10 @@ public Acl getAcl(String bucket, Entity entity) { public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options) { StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("DeleteBucketACL"); - StorageMetricsContext subContext = - multiContext != null ? multiContext.asSubAttemptContext() : null; Throwable error = null; try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, subContext); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = bucketAclEntityOrAltEq(encode); @@ -1147,7 +1126,7 @@ public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... opt UpdateBucketRequest req = createUpdateBucketAclRequest(bucket, newAcls, metageneration); - com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getAclList().stream().filter(entityPredicate).findFirst(); @@ -1195,12 +1174,10 @@ public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { private Acl internalUpdateAcl( String operationName, String bucket, Acl acl, BucketSourceOption... options) { StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext(operationName); - StorageMetricsContext subContext = - multiContext != null ? multiContext.asSubAttemptContext() : null; Throwable error = null; try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, subContext); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, multiContext); BucketAccessControl encode = codecs.bucketAcl().encode(acl); String entity = encode.getEntity(); @@ -1214,7 +1191,7 @@ private Acl internalUpdateAcl( UpdateBucketRequest req = createUpdateBucketAclRequest(bucket, newDefaultAcls, resp.getMetageneration()); - com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); Optional first = updateResult.getAclList().stream() @@ -1290,11 +1267,9 @@ public Acl getDefaultAcl(String bucket, Entity entity) { public boolean deleteDefaultAcl(String bucket, Entity entity) { StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("DeleteDefaultObjectACL"); - StorageMetricsContext subContext = - multiContext != null ? multiContext.asSubAttemptContext() : null; Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, subContext); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1313,7 +1288,7 @@ public boolean deleteDefaultAcl(String bucket, Entity entity) { UpdateBucketRequest req = createUpdateDefaultAclRequest(bucket, newDefaultAcls, metageneration); - com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getDefaultObjectAclList().stream().filter(entityPredicate).findFirst(); @@ -1350,11 +1325,9 @@ public Acl updateDefaultAcl(String bucket, Acl acl) { private Acl internalUpdateDefaultAcl(String operationName, String bucket, Acl acl) { StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext(operationName); - StorageMetricsContext subContext = - multiContext != null ? multiContext.asSubAttemptContext() : null; Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, subContext); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, multiContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1369,7 +1342,7 @@ private Acl internalUpdateDefaultAcl(String operationName, String bucket, Acl ac UpdateBucketRequest req = createUpdateDefaultAclRequest(bucket, newDefaultAcls, resp.getMetageneration()); - com.google.storage.v2.Bucket updateResult = updateBucket(req, subContext); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); Optional first = updateResult.getDefaultObjectAclList().stream() @@ -1436,12 +1409,10 @@ public Acl getAcl(BlobId blob, Entity entity) { public boolean deleteAcl(BlobId blob, Entity entity) { StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("DeleteObjectACL"); - StorageMetricsContext subContext = - multiContext != null ? multiContext.asSubAttemptContext() : null; Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj, subContext); + Object resp = getObjectWithAcls(obj, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1459,7 +1430,7 @@ public boolean deleteAcl(BlobId blob, Entity entity) { UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, metageneration); - Object updateResult = updateObject(req, subContext); + Object updateResult = updateObject(req, multiContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getAclList().stream().filter(entityPredicate).findFirst(); @@ -1496,12 +1467,10 @@ public Acl updateAcl(BlobId blob, Acl acl) { private Acl internalUpdateObjectAcl(String operationName, BlobId blob, Acl acl) { StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext(operationName); - StorageMetricsContext subContext = - multiContext != null ? multiContext.asSubAttemptContext() : null; Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj, subContext); + Object resp = getObjectWithAcls(obj, multiContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1515,7 +1484,7 @@ private Acl internalUpdateObjectAcl(String operationName, BlobId blob, Acl acl) UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, resp.getMetageneration()); - Object updateResult = updateObject(req, subContext); + Object updateResult = updateObject(req, multiContext); Optional first = updateResult.getAclList().stream() @@ -1700,10 +1669,13 @@ public Blob moveBlob(MoveBlobRequest request) { dstOpts.moveObjectsRequest().apply(b); MoveObjectRequest req = b.build(); + GrpcCallContext grpcCallContext = + dstOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); StorageMetricsContext metricsContext = getOptions().createMetricsContext("MoveObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), - () -> storageClient.moveObjectCallable().call(req), + () -> storageClient.moveObjectCallable().call(req, merge), syntaxDecoders.blob, metricsContext); } @@ -1806,22 +1778,12 @@ private final class ListObjectsWithSyntheticDirectoriesPage implements Page getNextPage() { ListObjectsRequest nextPageReq = req.toBuilder().setPageToken(resp.getNextPageToken()).build(); try { - StorageMetricsContext metricsContext = - parentMetricsContext != null - ? parentMetricsContext.asSubAttemptContext() - : getOptions().createMetricsContext("ListObjects"); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("ListObjects"); GrpcCallContext merge = Utils.merge(ctx, Retrying.newCallContext(metricsContext)); ListObjectsResponse nextPageResp = retrier.run( @@ -1850,8 +1809,7 @@ public Page getNextPage() { () -> storageClient.listObjectsCallable().call(nextPageReq, merge), Decoder.identity(), metricsContext); - return new ListObjectsWithSyntheticDirectoriesPage( - ctx, nextPageReq, nextPageResp, parentMetricsContext); + return new ListObjectsWithSyntheticDirectoriesPage(ctx, nextPageReq, nextPageResp); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -1975,27 +1933,16 @@ static final class TransformingPageDecorator< private final Decoder translator; private final Retrier retrier; private final ResultRetryAlgorithm resultRetryAlgorithm; - private final Supplier metricsContextSupplier; TransformingPageDecorator( PageT page, Decoder translator, Retrier retrier, ResultRetryAlgorithm resultRetryAlgorithm) { - this(page, translator, retrier, resultRetryAlgorithm, () -> null); - } - - TransformingPageDecorator( - PageT page, - Decoder translator, - Retrier retrier, - ResultRetryAlgorithm resultRetryAlgorithm, - Supplier metricsContextSupplier) { this.page = page; this.translator = translator; this.retrier = retrier; this.resultRetryAlgorithm = resultRetryAlgorithm; - this.metricsContextSupplier = metricsContextSupplier; } @Override @@ -2011,7 +1958,7 @@ public String getNextPageToken() { @Override public Page getNextPage() { return new TransformingPageDecorator<>( - page.getNextPage(), translator, retrier, resultRetryAlgorithm, metricsContextSupplier); + page.getNextPage(), translator, retrier, resultRetryAlgorithm); } @SuppressWarnings({"Convert2MethodRef"}) @@ -2031,8 +1978,7 @@ public Iterable iterateAll() { // prevent a javac 1.8 exception // https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8056984 Callable c = () -> prev.getNextPage(); - return retrier.run( - resultRetryAlgorithm, c, Decoder.identity(), metricsContextSupplier.get()); + return retrier.run(resultRetryAlgorithm, c, Decoder.identity()); }) .filter(Objects::nonNull) .flatMap(p -> StreamSupport.stream(p.getValues().spliterator(), false)) @@ -2164,11 +2110,6 @@ BidiWriteObjectRequest getBidiWriteObjectRequestForTakeover( return opts.bidiWriteObjectRequest().apply(requestBuilder).build(); } - private UnbufferedReadableByteChannelSession unbufferedReadSession( - BlobId blob, BlobSourceOption[] options) { - return unbufferedReadSession(blob, options, null); - } - private UnbufferedReadableByteChannelSession unbufferedReadSession( BlobId blob, BlobSourceOption[] options, @Nullable StorageMetricsContext metricsContext) { @@ -2176,14 +2117,14 @@ private UnbufferedReadableByteChannelSession unbufferedReadSession( ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts); GrpcCallContext subAttemptContext = metricsContext != null - ? Retrying.newCallContext(metricsContext.asSubAttemptContext()) + ? Retrying.newCallContext(metricsContext) : Retrying.newCallContext(); GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(subAttemptContext); return ResumableMedia.gapic() .read() .byteChannel( readObjectCallable(grpcCallContext), - retrier, + retrier.withMetricsContext(metricsContext), retryAlgorithmManager.getFor(readObjectRequest)) .setAutoGzipDecompression(!opts.autoGzipDecompression()) .unbuffered() @@ -2194,14 +2135,27 @@ private UnbufferedReadableByteChannelSession unbufferedReadSession( @VisibleForTesting ApiFuture startResumableWrite( GrpcCallContext grpcCallContext, WriteObjectRequest req, Opts opts) { - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + return startResumableWrite(grpcCallContext, req, opts, null); + } + + ApiFuture startResumableWrite( + GrpcCallContext grpcCallContext, + WriteObjectRequest req, + Opts opts, + @Nullable StorageMetricsContext metricsContext) { + GrpcCallContext merge = + Utils.merge( + grpcCallContext, + metricsContext != null + ? Retrying.newCallContext(metricsContext) + : Retrying.newCallContext()); return ResumableMedia.gapic() .write() .resumableWrite( storageClient.startResumableWriteCallable().withDefaultCallContext(merge), req, opts, - retrier.withAlg(retryAlgorithmManager.getFor(req))); + retrier.withAlg(retryAlgorithmManager.getFor(req), metricsContext)); } ApiFuture startResumableWrite( @@ -2234,10 +2188,6 @@ private com.google.storage.v2.ListBucketsResponse listBuckets( metricsContext); } - private com.google.storage.v2.Bucket getBucketWithDefaultAcls(String bucketName) { - return getBucketWithDefaultAcls(bucketName, null); - } - private com.google.storage.v2.Bucket getBucketWithDefaultAcls( String bucketName, @Nullable StorageMetricsContext metricsContext) { Fields fields = @@ -2264,11 +2214,6 @@ private com.google.storage.v2.Bucket getBucketWithDefaultAcls( activeContext); } - private com.google.storage.v2.Bucket getBucketWithAcls( - String bucketName, Opts opts) { - return getBucketWithAcls(bucketName, opts, null); - } - private com.google.storage.v2.Bucket getBucketWithAcls( String bucketName, Opts opts, @@ -2293,10 +2238,6 @@ private com.google.storage.v2.Bucket getBucketWithAcls( activeContext); } - private com.google.storage.v2.Bucket updateBucket(UpdateBucketRequest req) { - return updateBucket(req, null); - } - private com.google.storage.v2.Bucket updateBucket( UpdateBucketRequest req, @Nullable StorageMetricsContext metricsContext) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); @@ -2344,10 +2285,6 @@ private static UpdateBucketRequest createUpdateBucketAclRequest( .build(); } - private Object getObjectWithAcls(Object obj) { - return getObjectWithAcls(obj, null); - } - private Object getObjectWithAcls(Object obj, @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields(ImmutableSet.of(BucketField.ACL, BucketField.METAGENERATION)); @@ -2388,10 +2325,6 @@ private static UpdateObjectRequest createUpdateObjectAclRequest( .build(); } - private Object updateObject(UpdateObjectRequest req) { - return updateObject(req, null); - } - private Object updateObject( UpdateObjectRequest req, @Nullable StorageMetricsContext metricsContext) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault();