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..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 @@ -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,18 +43,41 @@ 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; + } + + private void recordOperation(@Nullable Throwable error) { + if (metricsContext != null) { + metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), error); + } } @Override public Blob getResult() { - while (!isDone()) { - copyChunk(); + 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); @@ -76,10 +101,28 @@ public long getTotalBytesCopied() { @Override 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()); + try { + RewriteObjectRequest req = + originalRequest.toBuilder() + .setRewriteToken(mostRecentResponse.getRewriteToken()) + .build(); + StorageMetricsContext chunkContext = + metricsContext != null + ? metricsContext + : 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/GrpcBlobReadChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java index 9dd39dfe47ea..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 @@ -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; @@ -61,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/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..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 @@ -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,35 +259,38 @@ public Blob create( @Override public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) { - 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, 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) { - throw StorageException.coalesce(e); - } + try (UnbufferedWritableByteChannel c = session.open()) { + ByteStreams.copy(Channels.newChannel(inputStreamParam), c); + } + ApiFuture responseApiFuture = session.getResult(); + return this.getBlob(responseApiFuture); + }, + Decoder.identity(), + metricsContext); } @Override @@ -307,37 +314,49 @@ 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, multiContext); + ApiFuture session2 = + ApiFutures.transform( + start, + rw -> + ResumableSession.grpc( + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext), + 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 +372,47 @@ 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, multiContext); - // 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(), multiContext)) + .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 +431,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 +473,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 +501,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 +512,8 @@ public Page list(BucketListOption... options) { resp.getPage(), syntaxDecoders.bucket.andThen(opts.clearBucketFields()), retrier, - retryAlgorithmManager.getFor(request))); + retryAlgorithmManager.getFor(request)), + metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -497,11 +536,13 @@ 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); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -526,11 +567,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 +595,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 +617,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 +630,8 @@ public boolean delete(String bucket, BucketSourceOption... options) { return false; } }, - Decoder.identity())); + Decoder.identity(), + metricsContext)); } @Override @@ -624,14 +671,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 +696,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 +759,31 @@ 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 : 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)); + 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 @@ -723,14 +793,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 +830,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) + : 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,19 +900,31 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); Hasher hasher = opts.getHasher(); - // 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); - // 2. await the result of the future - ResumableWrite resumableWrite = ApiFutureUtils.await(startResumableWrite); + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + ResumableWrite resumableWrite; + try { + // 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, multiContext); + // 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( storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext), - retrier.withAlg(retryAlgorithmManager.idempotent()), + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext), () -> wrapped, - hasher); + hasher, + multiContext, + getOptions().getStorageClientMetrics()); } @Override @@ -817,7 +937,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 +960,8 @@ public BlobInfo internalDirectUpload( } return session.getResult(); }, - this::getBlob); + this::getBlob, + metricsContext); } @Override @@ -950,9 +1072,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 +1103,12 @@ public Acl getAcl(String bucket, Entity entity) { @Override public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options) { + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("DeleteBucketACL"); + 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, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = bucketAclEntityOrAltEq(encode); @@ -1000,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); + 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(); @@ -1012,8 +1138,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 +1158,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 +1168,16 @@ 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); + 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, multiContext); BucketAccessControl encode = codecs.bucketAcl().encode(acl); String entity = encode.getEntity(); @@ -1050,7 +1191,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, multiContext); Optional first = updateResult.getAclList().stream() @@ -1061,7 +1202,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 +1221,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 +1240,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 +1265,11 @@ public Acl getDefaultAcl(String bucket, Entity entity) { @Override public boolean deleteDefaultAcl(String bucket, Entity entity) { + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("DeleteDefaultObjectACL"); + Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1134,7 +1288,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, multiContext); // 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 +1300,34 @@ 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); + Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, multiContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1174,7 +1342,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, multiContext); Optional first = updateResult.getDefaultObjectAclList().stream() @@ -1185,14 +1353,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 +1381,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 +1407,12 @@ public Acl getAcl(BlobId blob, Entity entity) { @Override public boolean deleteAcl(BlobId blob, Entity entity) { + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("DeleteObjectACL"); + Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj); + Object resp = getObjectWithAcls(obj, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1248,7 +1430,7 @@ public boolean deleteAcl(BlobId blob, Entity entity) { UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, metageneration); - Object updateResult = updateObject(req); + 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(); @@ -1260,21 +1442,35 @@ 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); + Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj); + Object resp = getObjectWithAcls(obj, multiContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1288,7 +1484,7 @@ public Acl updateAcl(BlobId blob, Acl acl) { UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, resp.getMetageneration()); - Object updateResult = updateObject(req); + Object updateResult = updateObject(req, multiContext); Optional first = updateResult.getAclList().stream() @@ -1299,15 +1495,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 +1555,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 +1574,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 +1594,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 +1604,8 @@ public List testIamPermissions( return permissions.stream() .map(heldPermissions::contains) .collect(ImmutableList.toImmutableList()); - }); + }, + metricsContext); } @Override @@ -1458,10 +1669,15 @@ 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), - syntaxDecoders.blob); + () -> storageClient.moveObjectCallable().call(req, merge), + syntaxDecoders.blob, + metricsContext); } @Override @@ -1585,12 +1801,14 @@ public Page getNextPage() { ListObjectsRequest nextPageReq = req.toBuilder().setPageToken(resp.getNextPageToken()).build(); try { - GrpcCallContext merge = Utils.merge(ctx, Retrying.newCallContext()); + StorageMetricsContext metricsContext = 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()); + Decoder.identity(), + metricsContext); return new ListObjectsWithSyntheticDirectoriesPage(ctx, nextPageReq, nextPageResp); } catch (Exception e) { throw StorageException.coalesce(e); @@ -1893,16 +2111,20 @@ BidiWriteObjectRequest getBidiWriteObjectRequestForTakeover( } private UnbufferedReadableByteChannelSession unbufferedReadSession( - BlobId blob, BlobSourceOption[] options) { + 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) + : 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() @@ -1913,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( @@ -1944,14 +2179,17 @@ 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) { + private com.google.storage.v2.Bucket getBucketWithDefaultAcls( + String bucketName, @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields( ImmutableSet.of( @@ -1966,15 +2204,20 @@ 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) { + String bucketName, + Opts opts, + @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields(ImmutableSet.of(BucketField.ACL, BucketField.METAGENERATION)); GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); @@ -1985,20 +2228,27 @@ 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) { + 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( @@ -2035,7 +2285,7 @@ private static UpdateBucketRequest createUpdateBucketAclRequest( .build(); } - private Object getObjectWithAcls(Object obj) { + private Object getObjectWithAcls(Object obj, @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields(ImmutableSet.of(BucketField.ACL, BucketField.METAGENERATION)); GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); @@ -2047,11 +2297,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( @@ -2072,13 +2325,17 @@ private static UpdateObjectRequest createUpdateObjectAclRequest( .build(); } - private Object updateObject(UpdateObjectRequest req) { + 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 +2350,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 +2359,8 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { resp -> { BlobInfo tmp = codecs.blobInfo().decode(resp); return finalOpts.clearBlobFields().decode(tmp); - }); + }, + metricsContext); } @Nullable @@ -2128,7 +2387,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 +2398,8 @@ private Bucket internalBucketGet(String bucket, Opts unwrap) { return null; } }, - syntaxDecoders.bucket.andThen(opts.clearBucketFields())); + syntaxDecoders.bucket.andThen(opts.clearBucketFields()), + metricsContext); } private ZeroCopyServerStreamingCallable readObjectCallable(