From 406226c9f2502000e7baae792b57c824520a2cc5 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Tue, 20 Oct 2026 14:05:00 +0000 Subject: [PATCH 1/3] feat(storage): wire veneer operation metrics in StorageImpl and GrpcStorageImpl - Wire operation-level metrics context across GrpcStorageImpl and StorageImpl - Wire RewriteObject metrics context in GapicCopyWriter.copyChunk() - Map operation names matching Go SDK / HLD PascalCase specification (CreateBucket, ListBuckets, GetBucket, UpdateBucket, DeleteBucket, LockBucketRetentionPolicy, ListObjects, GetObject, UpdateObject, DeleteObject, RestoreObject, ComposeObject, RewriteObject, MoveObject, WriteObject, ReadObject, GetIamPolicy, SetIamPolicy, TestIamPermissions, GetServiceAccount, CreateHMACKey, GetHMACKey, ListHMACKeys, UpdateHMACKey, DeleteHMACKey, CreateNotification, GetNotification, ListNotifications, DeleteNotification, GetBucketACL, DeleteBucketACL, CreateBucketACL, UpdateBucketACL, ListBucketACLs, GetDefaultObjectACL, DeleteDefaultObjectACL, CreateDefaultObjectACL, UpdateDefaultObjectACL, ListDefaultObjectACLs, GetObjectACL, DeleteObjectACL, CreateObjectACL, UpdateObjectACL, ListObjectACLs) - Explicitly merge GrpcCallContext with Retrying.newCallContext(metricsContext) across all unary/call sites in GrpcStorageImpl without wrapper helpers - Supply metrics context in pagination decorators (TransformingPageDecorator, ListObjectsWithSyntheticDirectoriesPage, ListBucketsWithPartialSuccessPage) - Leave streaming channel methods untouched for follow-up PR [Generated-by: AI] --- .../google/cloud/storage/GapicCopyWriter.java | 7 +- .../google/cloud/storage/GrpcStorageImpl.java | 167 ++++++++++++------ .../com/google/cloud/storage/StorageImpl.java | 146 ++++++++++++--- 3 files changed, 242 insertions(+), 78 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..2e2736606c88 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 @@ -78,8 +78,11 @@ 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 metricsContext = + storage.getOptions().createMetricsContext("RewriteObject"); + GrpcCallContext retryContext = Retrying.newCallContext(metricsContext); + mostRecentResponse = + retrier.run(() -> callable.call(req, retryContext), Decoder.identity(), metricsContext); } } 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..05c0ca964b84 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 @@ -227,11 +227,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 @@ -400,11 +402,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 +444,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 +472,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 +483,9 @@ public Page list(BucketListOption... options) { resp.getPage(), syntaxDecoders.bucket.andThen(opts.clearBucketFields()), retrier, - retryAlgorithmManager.getFor(request))); + retryAlgorithmManager.getFor(request), + () -> getOptions().createMetricsContext("ListBuckets")), + metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -497,11 +508,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 +539,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 +567,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 +589,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 +602,8 @@ public boolean delete(String bucket, BucketSourceOption... options) { return false; } }, - Decoder.identity())); + Decoder.identity(), + metricsContext)); } @Override @@ -624,14 +643,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 +668,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 +731,14 @@ public CopyWriter copy(CopyRequest copyRequest) { GrpcCallContext grpcCallContext = grpcCallContextMapper.apply(GrpcCallContext.createDefault()); UnaryCallable callable = storageClient.rewriteObjectCallable().withDefaultCallContext(grpcCallContext); - GrpcCallContext retryContext = Retrying.newCallContext(); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("RewriteObject"); + GrpcCallContext retryContext = Retrying.newCallContext(metricsContext); 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) -> new GapicCopyWriter(this, callable, retrierWithAlg, req, resp), + metricsContext); } @Override @@ -817,7 +842,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 +865,8 @@ public BlobInfo internalDirectUpload( } return session.getResult(); }, - this::getBlob); + this::getBlob, + metricsContext); } @Override @@ -1350,11 +1377,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 +1396,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 +1416,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 +1426,8 @@ public List testIamPermissions( return permissions.stream() .map(heldPermissions::contains) .collect(ImmutableList.toImmutableList()); - }); + }, + metricsContext); } @Override @@ -1458,10 +1491,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 @@ -1585,12 +1620,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); @@ -1715,16 +1752,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 +1788,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 +1808,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)) @@ -1944,11 +1993,13 @@ 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) { @@ -1966,11 +2017,13 @@ private com.google.storage.v2.Bucket getBucketWithDefaultAcls(String bucketName) .setName(bucketNameCodec.encode(bucketName)) .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), () -> storageClient.getBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + metricsContext); } private com.google.storage.v2.Bucket getBucketWithAcls( @@ -1985,20 +2038,24 @@ private com.google.storage.v2.Bucket getBucketWithAcls( .setName(bucketNameCodec.encode(bucketName)) .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), () -> storageClient.getBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + metricsContext); } private com.google.storage.v2.Bucket updateBucket(UpdateBucketRequest req) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); - 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), - Decoder.identity()); + Decoder.identity(), + metricsContext); } private static UpdateBucketRequest createUpdateDefaultAclRequest( @@ -2047,11 +2104,13 @@ private Object getObjectWithAcls(Object obj) { .setObject(obj.getName()) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getObjectCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + metricsContext); } private static UpdateObjectRequest createUpdateObjectAclRequest( @@ -2074,11 +2133,13 @@ private static UpdateObjectRequest createUpdateObjectAclRequest( private Object updateObject(UpdateObjectRequest req) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); - 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), - Decoder.identity()); + Decoder.identity(), + metricsContext); } @NonNull @@ -2093,7 +2154,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 +2163,8 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { resp -> { BlobInfo tmp = codecs.blobInfo().decode(resp); return finalOpts.clearBlobFields().decode(tmp); - }); + }, + metricsContext); } @Nullable @@ -2128,7 +2191,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 +2202,8 @@ private Bucket internalBucketGet(String bucket, Opts unwrap) { return null; } }, - syntaxDecoders.bucket.andThen(opts.clearBucketFields())); + syntaxDecoders.bucket.andThen(opts.clearBucketFields()), + metricsContext); } private ZeroCopyServerStreamingCallable readObjectCallable( diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java index 7060db7e4719..21efe541bfe3 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java @@ -143,6 +143,7 @@ public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsCreate(bucketPb, optionsMap); return run( + "CreateBucket", algorithm, () -> storageRpc.create(bucketPb, optionsMap), (b) -> Conversions.json().bucketInfo().decode(b).asBucket(this)); @@ -319,6 +320,7 @@ public Blob restore(BlobId blob, BlobRestoreOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsRestore(obj, optionsMap); return run( + "RestoreObject", algorithm, () -> storageRpc.restore(obj, optionsMap), (x) -> { @@ -438,7 +440,8 @@ private static Page listBuckets( .asBucket(serviceOptions.getService())); return new PageImpl<>( new BucketPageFetcher(serviceOptions, cursor, optionsMap, retrier), cursor, buckets); - }); + }, + serviceOptions.createMetricsContext("ListBuckets")); } private static Page listBlobs( @@ -466,7 +469,8 @@ private static Page listBlobs( new BlobPageFetcher(bucket, serviceOptions, cursor, optionsMap, retrier), cursor, blobs); - }); + }, + serviceOptions.createMetricsContext("ListObjects")); } @Override @@ -506,6 +510,7 @@ public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsUpdate(bucketPb, optionsMap); return run( + "UpdateBucket", algorithm, () -> storageRpc.patch(bucketPb, optionsMap), (x) -> Conversions.json().bucketInfo().decode(x).asBucket(this)); @@ -592,6 +597,7 @@ public Blob update(BlobInfo blobInfo, BlobTargetOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsUpdate(pb, optionsMap); return run( + "UpdateObject", algorithm, () -> storageRpc.patch(pb, optionsMap), (x) -> { @@ -613,7 +619,11 @@ public boolean delete(String bucket, BucketSourceOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsDelete(bucketPb, optionsMap); - return run(algorithm, () -> storageRpc.delete(bucketPb, optionsMap), Decoder.identity()); + return run( + "DeleteBucket", + algorithm, + () -> storageRpc.delete(bucketPb, optionsMap), + Decoder.identity()); } @Override @@ -628,7 +638,11 @@ public boolean delete(BlobId blob, BlobSourceOption... options) { Opts.unwrap(options).resolveFrom(blob).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsDelete(storageObject, optionsMap); - return run(algorithm, () -> storageRpc.delete(storageObject, optionsMap), Decoder.identity()); + return run( + "DeleteObject", + algorithm, + () -> storageRpc.delete(storageObject, optionsMap), + Decoder.identity()); } @Override @@ -660,6 +674,7 @@ public Blob compose(final ComposeRequest composeRequest) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsCompose(sources, targetPb, targetOptions); return run( + "ComposeObject", algorithm, () -> storageRpc.compose(sources, targetPb, targetOptions), (x) -> { @@ -691,6 +706,7 @@ public CopyWriter copy(final CopyRequest copyRequest) { copyRequest.getMegabytesCopiedPerChunk()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsRewrite(rewriteRequest); return run( + "RewriteObject", algorithm, () -> storageRpc.openRewrite(rewriteRequest), (r) -> new HttpCopyWriter(getOptions(), r, retrier)); @@ -709,7 +725,11 @@ public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) { ImmutableMap optionsMap = resolve.getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storageObject, optionsMap); - return run(algorithm, () -> storageRpc.load(storageObject, optionsMap), Decoder.identity()); + return run( + "ReadObject", + algorithm, + () -> storageRpc.load(storageObject, optionsMap), + Decoder.identity()); } @Override @@ -747,6 +767,7 @@ public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption. Opts.unwrap(options).resolveFrom(blob).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(pb, optionsMap); run( + "ReadObject", algorithm, callable( () -> { @@ -1279,7 +1300,11 @@ public Acl getAcl(final String bucket, final Entity entity, BucketSourceOption.. String pb = codecs.entity().encode(entity); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclGet(pb, optionsMap); - return run(algorithm, () -> storageRpc.getAcl(bucket, pb, optionsMap), codecs.bucketAcl()); + return run( + "GetBucketACL", + algorithm, + () -> storageRpc.getAcl(bucket, pb, optionsMap), + codecs.bucketAcl()); } @Override @@ -1293,7 +1318,11 @@ public boolean deleteAcl( final String pb = codecs.entity().encode(entity); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclDelete(pb, optionsMap); - return run(algorithm, () -> storageRpc.deleteAcl(bucket, pb, optionsMap), Decoder.identity()); + return run( + "DeleteBucketACL", + algorithm, + () -> storageRpc.deleteAcl(bucket, pb, optionsMap), + Decoder.identity()); } @Override @@ -1307,7 +1336,11 @@ public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclCreate(aclPb, optionsMap); - return run(algorithm, () -> storageRpc.createAcl(aclPb, optionsMap), codecs.bucketAcl()); + return run( + "CreateBucketACL", + algorithm, + () -> storageRpc.createAcl(aclPb, optionsMap), + codecs.bucketAcl()); } @Override @@ -1321,7 +1354,11 @@ public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclUpdate(aclPb, optionsMap); - return run(algorithm, () -> storageRpc.patchAcl(aclPb, optionsMap), codecs.bucketAcl()); + return run( + "UpdateBucketACL", + algorithm, + () -> storageRpc.patchAcl(aclPb, optionsMap), + codecs.bucketAcl()); } @Override @@ -1335,6 +1372,7 @@ public List listAcls(final String bucket, BucketSourceOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclList(bucket, optionsMap); return run( + "ListBucketACLs", algorithm, () -> storageRpc.listAcls(bucket, optionsMap), (answer) -> @@ -1354,34 +1392,51 @@ public List listAcls(final String bucket) { public Acl getDefaultAcl(final String bucket, final Entity entity) { String pb = codecs.entity().encode(entity); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclGet(pb); - return run(algorithm, () -> storageRpc.getDefaultAcl(bucket, pb), codecs.objectAcl()); + return run( + "GetDefaultObjectACL", + algorithm, + () -> storageRpc.getDefaultAcl(bucket, pb), + codecs.objectAcl()); } @Override public boolean deleteDefaultAcl(final String bucket, final Entity entity) { String pb = codecs.entity().encode(entity); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclDelete(pb); - return run(algorithm, () -> storageRpc.deleteDefaultAcl(bucket, pb), Decoder.identity()); + return run( + "DeleteDefaultObjectACL", + algorithm, + () -> storageRpc.deleteDefaultAcl(bucket, pb), + Decoder.identity()); } @Override public Acl createDefaultAcl(String bucket, Acl acl) { final ObjectAccessControl aclPb = codecs.objectAcl().encode(acl).setBucket(bucket); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclCreate(aclPb); - return run(algorithm, () -> storageRpc.createDefaultAcl(aclPb), codecs.objectAcl()); + return run( + "CreateDefaultObjectACL", + algorithm, + () -> storageRpc.createDefaultAcl(aclPb), + codecs.objectAcl()); } @Override public Acl updateDefaultAcl(String bucket, Acl acl) { final ObjectAccessControl aclPb = codecs.objectAcl().encode(acl).setBucket(bucket); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclUpdate(aclPb); - return run(algorithm, () -> storageRpc.patchDefaultAcl(aclPb), codecs.objectAcl()); + return run( + "UpdateDefaultObjectACL", + algorithm, + () -> storageRpc.patchDefaultAcl(aclPb), + codecs.objectAcl()); } @Override public List listDefaultAcls(final String bucket) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclList(bucket); return run( + "ListDefaultObjectACLs", algorithm, () -> storageRpc.listDefaultAcls(bucket), (answer) -> @@ -1401,7 +1456,10 @@ public Acl getAcl(final BlobId blob, final Entity entity) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclGet(bucket, name, generation, pb); return run( - algorithm, () -> storageRpc.getAcl(bucket, name, generation, pb), codecs.objectAcl()); + "GetObjectACL", + algorithm, + () -> storageRpc.getAcl(bucket, name, generation, pb), + codecs.objectAcl()); } @Override @@ -1413,7 +1471,10 @@ public boolean deleteAcl(final BlobId blob, final Entity entity) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclDelete(bucket, name, generation, pb); return run( - algorithm, () -> storageRpc.deleteAcl(bucket, name, generation, pb), Decoder.identity()); + "DeleteObjectACL", + algorithm, + () -> storageRpc.deleteAcl(bucket, name, generation, pb), + Decoder.identity()); } @Override @@ -1426,7 +1487,7 @@ public Acl createAcl(final BlobId blob, final Acl acl) { .setObject(blob.getName()) .setGeneration(blob.getGeneration()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclCreate(aclPb); - return run(algorithm, () -> storageRpc.createAcl(aclPb), codecs.objectAcl()); + return run("CreateObjectACL", algorithm, () -> storageRpc.createAcl(aclPb), codecs.objectAcl()); } @Override @@ -1439,7 +1500,7 @@ public Acl updateAcl(BlobId blob, Acl acl) { .setObject(blob.getName()) .setGeneration(blob.getGeneration()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclUpdate(aclPb); - return run(algorithm, () -> storageRpc.patchAcl(aclPb), codecs.objectAcl()); + return run("UpdateObjectACL", algorithm, () -> storageRpc.patchAcl(aclPb), codecs.objectAcl()); } @Override @@ -1450,6 +1511,7 @@ public List listAcls(final BlobId blob) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclList(bucket, name, generation); return run( + "ListObjectACLs", algorithm, () -> storageRpc.listAcls(bucket, name, generation), (answer) -> @@ -1466,7 +1528,11 @@ public HmacKey createHmacKey( String pb = serviceAccount.getEmail(); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyCreate(pb, optionsMap); - return run(algorithm, () -> storageRpc.createHmacKey(pb, optionsMap), codecs.hmacKey()); + return run( + "CreateHMACKey", + algorithm, + () -> storageRpc.createHmacKey(pb, optionsMap), + codecs.hmacKey()); } @Override @@ -1481,7 +1547,10 @@ public HmacKeyMetadata getHmacKey(final String accessId, final GetHmacKeyOption. ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyGet(accessId, optionsMap); return run( - algorithm, () -> storageRpc.getHmacKey(accessId, optionsMap), codecs.hmacKeyMetadata()); + "GetHMACKey", + algorithm, + () -> storageRpc.getHmacKey(accessId, optionsMap), + codecs.hmacKeyMetadata()); } private HmacKeyMetadata updateHmacKey( @@ -1490,7 +1559,11 @@ private HmacKeyMetadata updateHmacKey( codecs.hmacKeyMetadata().encode(hmacKeyMetadata); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyUpdate(pb, optionsMap); - return run(algorithm, () -> storageRpc.updateHmacKey(pb, optionsMap), codecs.hmacKeyMetadata()); + return run( + "UpdateHMACKey", + algorithm, + () -> storageRpc.updateHmacKey(pb, optionsMap), + codecs.hmacKeyMetadata()); } @Override @@ -1514,6 +1587,7 @@ public void deleteHmacKey(final HmacKeyMetadata metadata, final DeleteHmacKeyOpt ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyDelete(pb, optionsMap); run( + "DeleteHMACKey", algorithm, (Callable) () -> { @@ -1543,7 +1617,8 @@ private static Page listHmacKeys( serviceOptions, retryAlgorithmManager, options, retrier), cursor, metadata); - }); + }, + serviceOptions.createMetricsContext("ListHMACKeys")); } @Override @@ -1552,6 +1627,7 @@ public Policy getIamPolicy(final String bucket, BucketSourceOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsGetIamPolicy(bucket, optionsMap); return run( + "GetIamPolicy", algorithm, () -> storageRpc.getIamPolicy(bucket, optionsMap), apiPolicy -> Conversions.json().policyCodec().decode(apiPolicy)); @@ -1566,6 +1642,7 @@ public Policy setIamPolicy( ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsSetIamPolicy(bucket, pb, optionsMap); return run( + "SetIamPolicy", algorithm, () -> storageRpc.setIamPolicy(bucket, pb, optionsMap), apiPolicy -> Conversions.json().policyCodec().decode(apiPolicy)); @@ -1578,6 +1655,7 @@ public List testIamPermissions( ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsTestIamPermissions(bucket, permissions, optionsMap); return run( + "TestIamPermissions", algorithm, () -> storageRpc.testIamPermissions(bucket, permissions, optionsMap), (response) -> { @@ -1600,6 +1678,7 @@ public Bucket lockRetentionPolicy(BucketInfo bucketInfo, BucketTargetOption... o ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsLockRetentionPolicy(bucketPb, optionsMap); return run( + "LockBucketRetentionPolicy", algorithm, () -> storageRpc.lockRetentionPolicy(bucketPb, optionsMap), (x) -> Conversions.json().bucketInfo().decode(x).asBucket(this)); @@ -1608,11 +1687,16 @@ public Bucket lockRetentionPolicy(BucketInfo bucketInfo, BucketTargetOption... o @Override public ServiceAccount getServiceAccount(final String projectId) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForServiceAccountGet(projectId); - return run(algorithm, () -> storageRpc.getServiceAccount(projectId), codecs.serviceAccount()); + return run( + "GetServiceAccount", + algorithm, + () -> storageRpc.getServiceAccount(projectId), + codecs.serviceAccount()); } - private U run(ResultRetryAlgorithm algorithm, Callable c, Decoder f) { - return retrier.run(algorithm, c, f); + private U run( + String operationName, ResultRetryAlgorithm algorithm, Callable c, Decoder f) { + return retrier.run(algorithm, c, f, getOptions().createMetricsContext(operationName)); } @Override @@ -1623,6 +1707,7 @@ public Notification createNotification( ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationCreate(bucket, notificationPb); return run( + "CreateNotification", algorithm, () -> storageRpc.createNotification(bucket, notificationPb), n -> codecs.notificationInfo().decode(n).asNotification(this)); @@ -1633,6 +1718,7 @@ public Notification getNotification(final String bucket, final String notificati ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationGet(bucket, notificationId); return run( + "GetNotification", algorithm, () -> storageRpc.getNotification(bucket, notificationId), n -> codecs.notificationInfo().decode(n).asNotification(this)); @@ -1643,6 +1729,7 @@ public List listNotifications(final String bucket) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationList(bucket); List result = run( + "ListNotifications", algorithm, () -> storageRpc.listNotifications(bucket), (answer) -> @@ -1657,7 +1744,10 @@ public boolean deleteNotification(final String bucket, final String notification ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationDelete(bucket, notificationId); return run( - algorithm, () -> storageRpc.deleteNotification(bucket, notificationId), Decoder.identity()); + "DeleteNotification", + algorithm, + () -> storageRpc.deleteNotification(bucket, notificationId), + Decoder.identity()); } @Override @@ -1670,6 +1760,7 @@ private Blob internalGetBlob(BlobId blob, Map optionsMap) ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storedObject, optionsMap); return run( + "GetObject", algorithm, () -> storageRpc.get(storedObject, optionsMap), (x) -> { @@ -1684,6 +1775,7 @@ private Bucket internalBucketGet(String bucket, Map option ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsGet(bucketPb, optionsMap); return run( + "GetBucket", algorithm, () -> storageRpc.get(bucketPb, optionsMap), (b) -> Conversions.json().bucketInfo().decode(b).asBucket(this)); @@ -1708,6 +1800,7 @@ public Blob moveBlob(MoveBlobRequest request) { ImmutableMap targetOptions = dstOpts.getRpcOptions(); return run( + "MoveObject", retryAlgorithmManager.getForObjectsMove(sourceOptions, targetOptions), () -> storageRpc.moveObject( @@ -1783,6 +1876,7 @@ public BlobInfo internalDirectUpload(BlobInfo info, Opts opts, retryAlgorithmManager.getForObjectsCreate(encoded, optionsMap); RewindableContent content = RewindableContent.of(buf); return run( + "WriteObject", algorithm, () -> { content.rewindTo(0); @@ -1802,6 +1896,7 @@ public Void internalObjectDelete(BlobId id, Opts opts) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsDelete(storageObject, optionsMap); return run( + "DeleteObject", algorithm, () -> { boolean deleted = storageRpc.delete(storageObject, optionsMap); @@ -1821,6 +1916,7 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storedObject, optionsMap); return run( + "GetObject", algorithm, () -> { StorageObject storageObject = storageRpc.get(storedObject, optionsMap); From a1b4efaa6484f84a6b3494ad55dfe7188eb90620 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Tue, 20 Oct 2026 14:23:45 +0000 Subject: [PATCH 2/3] feat(storage): wire RewriteObject metrics in HttpCopyWriter - Pass serviceOptions.createMetricsContext("RewriteObject") in copyChunk() - Use serviceOptions.asRetryDependencies() in StateImpl.restore() - Add test in CopyWriterTest to verify RewriteObject metrics context is passed [Generated-by: AI] --- .../google/cloud/storage/HttpCopyWriter.java | 9 +++------ .../google/cloud/storage/CopyWriterTest.java | 20 +++++++++++++++++++ 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java index 6d8a24fc8eda..b28578d41441 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java @@ -21,7 +21,6 @@ import com.google.cloud.storage.Retrying.DefaultRetrier; import com.google.cloud.storage.Retrying.HttpRetrier; import com.google.cloud.storage.Retrying.Retrier; -import com.google.cloud.storage.Retrying.RetryingDependencies; import com.google.cloud.storage.spi.v1.StorageRpc; import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest; import com.google.cloud.storage.spi.v1.StorageRpc.RewriteResponse; @@ -78,7 +77,8 @@ public void copyChunk() { retrier.run( serviceOptions.getRetryAlgorithmManager().getForObjectsRewrite(rewriteRequest), () -> storageRpc.continueRewrite(rewriteResponse), - Decoder.identity()); + Decoder.identity(), + serviceOptions.createMetricsContext("RewriteObject")); } } @@ -230,10 +230,7 @@ public CopyWriter restore() { totalBytesCopied); HttpRetrier httpRetrier = new HttpRetrier( - new DefaultRetrier( - UnaryOperator.identity(), - RetryingDependencies.simple( - serviceOptions.getClock(), serviceOptions.getRetrySettings()))); + new DefaultRetrier(UnaryOperator.identity(), serviceOptions.asRetryDependencies())); return new HttpCopyWriter(serviceOptions, rewriteResponse, httpRetrier); } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java index dfcd8851ecfe..1e410d9d3655 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java @@ -17,6 +17,7 @@ package com.google.cloud.storage; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.times; @@ -34,6 +35,7 @@ import java.util.Map; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; public class CopyWriterTest { @@ -210,4 +212,22 @@ public void testSaveAndRestoreWithResult() { assertEquals(42L, restoredRewriter.getBlobSize()); verify(storageRpcMock).continueRewrite(RESPONSE_WITH_OBJECT); } + + @Test + public void testCopyChunkPassesMetricsContext() { + HttpStorageOptions metricsOptions = + options.toBuilder() + .setEnableOtelMetrics(true) + .setMeterProvider(io.opentelemetry.sdk.metrics.SdkMeterProvider.builder().build()) + .build(); + Retrier retrierMock = Mockito.mock(Retrier.class); + copyWriter = new HttpCopyWriter(metricsOptions, RESPONSE_WITH_OBJECT, retrierMock); + copyWriter.copyChunk(); + ArgumentCaptor captor = + ArgumentCaptor.forClass(StorageMetricsContext.class); + verify(retrierMock).run(any(), any(), any(), captor.capture()); + StorageMetricsContext context = captor.getValue(); + assertNotNull(context); + assertEquals("RewriteObject", context.getOperationName()); + } } From 522bc8b5c30c9360197644851d91158de0ee8508 Mon Sep 17 00:00:00 2001 From: Nidhi Nandwani Date: Wed, 21 Oct 2026 09:56:06 +0000 Subject: [PATCH 3/3] feat(storage): address review comments on veneer operation metrics - Wire multi-step metrics context across operations with sub-calls: - ACL mutations in GrpcStorageImpl: use createMultiStepMetricsContext for DeleteBucketACL, CreateBucketACL, UpdateBucketACL, DeleteDefaultObjectACL, CreateDefaultObjectACL, UpdateDefaultObjectACL, DeleteObjectACL, CreateObjectACL, UpdateObjectACL, pass asSubAttemptContext() to constituent get and update RPCs, and record overall operation in finally. - Read operations: wire BaseStorageReadChannel and implementations (GrpcBlobReadChannel, BlobReadChannelV2) to record ReadObject on channel close(). Pass createMultiStepMetricsContext("ReadObject") from reader(), and wrap readAllBytes and downloadTo in GrpcStorageImpl. - Write operations: wire BaseStorageWriteChannel and implementations (GrpcBlobWriteChannel, BlobWriteChannelV2) to record WriteObject on channel close(). Pass createMultiStepMetricsContext("WriteObject") from writer(), and wrap createFrom / internalCreateFrom. - Multi-step copy: pass multi-step context in HttpCopyWriter and GapicCopyWriter so chunks increment sub-attempts while overall RewriteObject operation is recorded upon completion. - Pagination: propagate parent metrics context as sub-attempts across page fetchers and decorators. - Close owned StorageClientMetrics in StorageImpl.close() and GrpcStorageImpl.close(). [Generated-by: AI] --- .../cloud/storage/BaseStorageReadChannel.java | 33 +- .../storage/BaseStorageWriteChannel.java | 23 +- .../cloud/storage/BlobReadChannelV2.java | 16 +- .../cloud/storage/BlobWriteChannelV2.java | 15 +- .../google/cloud/storage/GapicCopyWriter.java | 34 +- .../cloud/storage/GrpcBlobReadChannel.java | 14 +- .../cloud/storage/GrpcBlobWriteChannel.java | 16 +- .../google/cloud/storage/GrpcStorageImpl.java | 468 ++++++++++++++---- .../google/cloud/storage/HttpCopyWriter.java | 28 +- .../com/google/cloud/storage/StorageImpl.java | 228 +++++++-- 10 files changed, 697 insertions(+), 178 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java index 465d3d6ae2f8..7ab63c19b7e7 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java @@ -43,13 +43,24 @@ abstract class BaseStorageReadChannel implements StorageReadChannel { private BufferHandle bufferHandle; private LazyReadChannel lazyReadChannel; protected final ReentrantLock lock; + @Nullable private final StorageMetricsContext metricsContext; + @Nullable private final StorageClientMetrics storageClientMetrics; protected BaseStorageReadChannel(Decoder objectDecoder) { + this(objectDecoder, null, null); + } + + protected BaseStorageReadChannel( + Decoder objectDecoder, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { this.objectDecoder = objectDecoder; this.result = SettableApiFuture.create(); this.open = true; this.byteRangeSpec = ByteRangeSpec.nullRange(); this.lock = new ReentrantLock(); + this.metricsContext = metricsContext; + this.storageClientMetrics = storageClientMetrics; } @Override @@ -76,14 +87,26 @@ public final boolean isOpen() { @Override public final void close() { lock.lock(); + Throwable error = null; try { - open = false; - if (internalGetLazyChannel().isOpen()) { - ReadableByteChannel channel = internalGetLazyChannel().getChannel(); - StorageException.wrapIOException(channel::close); + if (open) { + open = false; + if (internalGetLazyChannel().isOpen()) { + ReadableByteChannel channel = internalGetLazyChannel().getChannel(); + StorageException.wrapIOException(channel::close); + } } + } catch (Throwable t) { + error = t; + throw t; } finally { - lock.unlock(); + try { + if (metricsContext != null && storageClientMetrics != null) { + metricsContext.recordOperation(storageClientMetrics, error); + } + } finally { + lock.unlock(); + } } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java index 10f79c8dffdf..cd8d6597dc32 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java @@ -44,6 +44,8 @@ abstract class BaseStorageWriteChannel implements StorageWriteChannel { private int chunkSize; private LazyWriteChannel lazyWriteChannel; private BufferHandle bufferHandle; + @Nullable private final StorageMetricsContext metricsContext; + @Nullable private final StorageClientMetrics storageClientMetrics; /** * This is tracked for compatibility with BlobWriteChannel, such that simply creating a writer @@ -55,12 +57,21 @@ abstract class BaseStorageWriteChannel implements StorageWriteChannel { protected boolean writeCalledAtLeastOnce; protected BaseStorageWriteChannel(Decoder objectDecoder) { + this(objectDecoder, null, null); + } + + protected BaseStorageWriteChannel( + Decoder objectDecoder, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { this.objectDecoder = objectDecoder; this.result = SettableApiFuture.create(); this.lock = new ReentrantLock(); this.open = true; this.chunkSize = _16MiB; this.writeCalledAtLeastOnce = false; + this.metricsContext = metricsContext; + this.storageClientMetrics = storageClientMetrics; } @Override @@ -90,6 +101,7 @@ public final boolean isOpen() { @Override public final void close() throws IOException { lock.lock(); + Throwable error = null; try { if (open && !writeCalledAtLeastOnce) { this.write(ByteBuffer.allocate(0)); @@ -97,9 +109,18 @@ public final void close() throws IOException { if (internalGetLazyChannel().isOpen()) { StorageException.wrapIOException(internalGetLazyChannel().getChannel()::close); } + } catch (Throwable t) { + error = t; + throw t; } finally { open = false; - lock.unlock(); + try { + if (metricsContext != null && storageClientMetrics != null) { + metricsContext.recordOperation(storageClientMetrics, error); + } + } finally { + lock.unlock(); + } } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java index 8df640b7490a..b1266b9afc43 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java @@ -28,6 +28,7 @@ import java.io.Serializable; import java.util.Map; import java.util.Objects; +import org.checkerframework.checker.nullness.qual.Nullable; final class BlobReadChannelV2 extends BaseStorageReadChannel { @@ -40,7 +41,20 @@ final class BlobReadChannelV2 extends BaseStorageReadChannel { StorageObject storageObject, Map opts, BlobReadChannelContext blobReadChannelContext) { - super(Conversions.json().blobInfo()); + this(storageObject, opts, blobReadChannelContext, null); + } + + BlobReadChannelV2( + StorageObject storageObject, + Map opts, + BlobReadChannelContext blobReadChannelContext, + @Nullable StorageMetricsContext metricsContext) { + super( + Conversions.json().blobInfo(), + metricsContext, + blobReadChannelContext.getStorageOptions() != null + ? blobReadChannelContext.getStorageOptions().getStorageClientMetrics() + : null); this.storageObject = storageObject; this.opts = opts; this.blobReadChannelContext = blobReadChannelContext; diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java index edda3ce245e7..dc365d3eb012 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java @@ -25,6 +25,7 @@ import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Objects; +import org.checkerframework.checker.nullness.qual.Nullable; final class BlobWriteChannelV2 extends BaseStorageWriteChannel { @@ -32,7 +33,19 @@ final class BlobWriteChannelV2 extends BaseStorageWriteChannel { private final JsonResumableWrite start; BlobWriteChannelV2(BlobReadChannelContext blobChannelContext, JsonResumableWrite start) { - super(Conversions.json().blobInfo()); + this(blobChannelContext, start, null); + } + + BlobWriteChannelV2( + BlobReadChannelContext blobChannelContext, + JsonResumableWrite start, + @Nullable StorageMetricsContext metricsContext) { + super( + Conversions.json().blobInfo(), + metricsContext, + blobChannelContext.getStorageOptions() != null + ? blobChannelContext.getStorageOptions().getStorageClientMetrics() + : null); this.start = start; this.blobChannelContext = blobChannelContext; } 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 2e2736606c88..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,11 +94,19 @@ public void copyChunk() { if (!isDone()) { RewriteObjectRequest req = originalRequest.toBuilder().setRewriteToken(mostRecentResponse.getRewriteToken()).build(); - StorageMetricsContext metricsContext = - storage.getOptions().createMetricsContext("RewriteObject"); - GrpcCallContext retryContext = Retrying.newCallContext(metricsContext); - mostRecentResponse = - retrier.run(() -> callable.call(req, retryContext), Decoder.identity(), metricsContext); + 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 05c0ca964b84..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(); } } @@ -257,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)); @@ -266,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() @@ -284,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); + } } } @@ -309,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); } } @@ -355,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 @@ -484,7 +523,7 @@ public Page list(BucketListOption... options) { syntaxDecoders.bucket.andThen(opts.clearBucketFields()), retrier, retryAlgorithmManager.getFor(request), - () -> getOptions().createMetricsContext("ListBuckets")), + () -> metricsContext != null ? metricsContext.asSubAttemptContext() : null), metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); @@ -513,7 +552,9 @@ public Page list(String bucket, BlobListOption... options) { 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); @@ -731,14 +772,26 @@ public CopyWriter copy(CopyRequest copyRequest) { GrpcCallContext grpcCallContext = grpcCallContextMapper.apply(GrpcCallContext.createDefault()); UnaryCallable callable = storageClient.rewriteObjectCallable().withDefaultCallContext(grpcCallContext); - StorageMetricsContext metricsContext = getOptions().createMetricsContext("RewriteObject"); - GrpcCallContext retryContext = Retrying.newCallContext(metricsContext); + 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), - metricsContext); + (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 @@ -748,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(); } @@ -774,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); + } } } @@ -817,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 @@ -829,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 @@ -977,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)); @@ -1007,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); @@ -1027,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(); @@ -1039,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); + } } } @@ -1051,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 @@ -1061,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(); @@ -1077,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() @@ -1088,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); + } } } @@ -1099,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()); @@ -1117,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)); @@ -1141,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); @@ -1161,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(); @@ -1173,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(); @@ -1201,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() @@ -1212,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()); @@ -1230,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)); @@ -1255,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); @@ -1275,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(); @@ -1287,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(); @@ -1315,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() @@ -1326,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()); @@ -1597,12 +1791,22 @@ private final class ListObjectsWithSyntheticDirectoriesPage implements Page getNextPage() { ListObjectsRequest nextPageReq = req.toBuilder().setPageToken(resp.getNextPageToken()).build(); try { - StorageMetricsContext metricsContext = getOptions().createMetricsContext("ListObjects"); + StorageMetricsContext metricsContext = + parentMetricsContext != null + ? parentMetricsContext.asSubAttemptContext() + : getOptions().createMetricsContext("ListObjects"); GrpcCallContext merge = Utils.merge(ctx, Retrying.newCallContext(metricsContext)); ListObjectsResponse nextPageResp = retrier.run( @@ -1628,7 +1835,8 @@ public Page getNextPage() { () -> storageClient.listObjectsCallable().call(nextPageReq, merge), Decoder.identity(), metricsContext); - return new ListObjectsWithSyntheticDirectoriesPage(ctx, nextPageReq, nextPageResp); + return new ListObjectsWithSyntheticDirectoriesPage( + ctx, nextPageReq, nextPageResp, parentMetricsContext); } catch (Exception e) { throw StorageException.coalesce(e); } @@ -1943,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( @@ -2003,6 +2220,11 @@ private com.google.storage.v2.ListBucketsResponse listBuckets( } 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( @@ -2017,17 +2239,25 @@ private com.google.storage.v2.Bucket getBucketWithDefaultAcls(String bucketName) .setName(bucketNameCodec.encode(bucketName)) .build(); - StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetBucket"); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + 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(), - metricsContext); + 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(); @@ -2038,24 +2268,31 @@ private com.google.storage.v2.Bucket getBucketWithAcls( .setName(bucketNameCodec.encode(bucketName)) .build(); - StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetBucket"); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + 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(), - metricsContext); + 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(); - StorageMetricsContext metricsContext = getOptions().createMetricsContext("UpdateBucket"); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + 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(), - metricsContext); + activeContext); } private static UpdateBucketRequest createUpdateDefaultAclRequest( @@ -2093,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(); @@ -2104,13 +2345,14 @@ private Object getObjectWithAcls(Object obj) { .setObject(obj.getName()) .build(); - StorageMetricsContext metricsContext = getOptions().createMetricsContext("GetObject"); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + 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(), - metricsContext); + activeContext); } private static UpdateObjectRequest createUpdateObjectAclRequest( @@ -2132,14 +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(); - StorageMetricsContext metricsContext = getOptions().createMetricsContext("UpdateObject"); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + 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(), - metricsContext); + activeContext); } @NonNull diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java index b28578d41441..95063cc79f66 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java @@ -29,6 +29,7 @@ import java.util.Map; import java.util.Objects; import java.util.function.UnaryOperator; +import org.checkerframework.checker.nullness.qual.Nullable; public class HttpCopyWriter extends CopyWriter { @@ -36,13 +37,23 @@ public class HttpCopyWriter extends CopyWriter { private final StorageRpc storageRpc; private RewriteResponse rewriteResponse; private final Retrier retrier; + @Nullable private final StorageMetricsContext metricsContext; HttpCopyWriter( HttpStorageOptions serviceOptions, RewriteResponse rewriteResponse, Retrier retrier) { + this(serviceOptions, rewriteResponse, retrier, null); + } + + HttpCopyWriter( + HttpStorageOptions serviceOptions, + RewriteResponse rewriteResponse, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { this.serviceOptions = serviceOptions; this.rewriteResponse = rewriteResponse; this.storageRpc = serviceOptions.getStorageRpcV1(); this.retrier = retrier; + this.metricsContext = metricsContext; } @Override @@ -50,6 +61,9 @@ public Blob getResult() { while (!isDone()) { copyChunk(); } + if (metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), null); + } BlobInfo info = Conversions.json().blobInfo().decode(rewriteResponse.result); return info.asBlob(serviceOptions.getService()); } @@ -73,12 +87,22 @@ public long getTotalBytesCopied() { public void copyChunk() { if (!isDone()) { RewriteRequest rewriteRequest = rewriteResponse.rewriteRequest; - this.rewriteResponse = + StorageMetricsContext chunkContext = + metricsContext != null + ? metricsContext.asSubAttemptContext() + : serviceOptions.createMetricsContext("RewriteObject"); + RewriteResponse resp = retrier.run( serviceOptions.getRetryAlgorithmManager().getForObjectsRewrite(rewriteRequest), () -> storageRpc.continueRewrite(rewriteResponse), Decoder.identity(), - serviceOptions.createMetricsContext("RewriteObject")); + chunkContext); + if (resp != null) { + this.rewriteResponse = resp; + } + if (isDone() && metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), null); + } } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java index 21efe541bfe3..2db4707d6d58 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java @@ -134,6 +134,11 @@ final class StorageImpl extends BaseService implements Storage, this.retrier = retrier; } + @Override + public void close() throws Exception { + getOptions().closeStorageClientMetrics(); + } + @Override public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) { final com.google.api.services.storage.model.Bucket bucketPb = @@ -204,43 +209,55 @@ public Blob createFrom(BlobInfo blobInfo, Path path, int bufferSize, BlobWriteOp if (size == 0L) { return create(blobInfo, null, options); } - Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - final Map optionsMap = opts.getRpcOptions(); - BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null); - BlobInfo updated = opts.blobInfoMapper().apply(builder).build(); - StorageObject encode = codecs.blobInfo().encode(updated); - - Supplier uploadIdSupplier = - ResumableMedia.startUploadForBlobInfo( - getOptions(), - updated, - optionsMap, - retrier.withAlg(retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap))); - JsonResumableWrite jsonResumableWrite = - JsonResumableWrite.of( - encode, - optionsMap, - uploadIdSupplier.get(), - 0, - opts.getHasher(), - opts.getHasher().initialValue()); - - JsonResumableSession session = - ResumableSession.json( - HttpClientContext.from(storageRpc), - retrier.withAlg(retryAlgorithmManager.idempotent()), - jsonResumableWrite); - HttpContentRange contentRange = HttpContentRange.of(ByteRangeSpec.explicit(0L, size), size); - ResumableOperationResult put = - session.put(RewindableContent.of(path), contentRange); - // all exception translation is taken care of down in the JsonResumableSession - StorageObject object = put.getObject(); - if (object == null) { - // if by some odd chance the put didn't get the StorageObject, query for it - ResumableOperationResult<@Nullable StorageObject> query = session.query(); - object = query.getObject(); + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + Throwable error = null; + try { + Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); + final Map optionsMap = opts.getRpcOptions(); + BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null); + BlobInfo updated = opts.blobInfoMapper().apply(builder).build(); + StorageObject encode = codecs.blobInfo().encode(updated); + + Supplier uploadIdSupplier = + ResumableMedia.startUploadForBlobInfo( + getOptions(), + updated, + optionsMap, + retrier.withAlg( + retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap))); + JsonResumableWrite jsonResumableWrite = + JsonResumableWrite.of( + encode, + optionsMap, + uploadIdSupplier.get(), + 0, + opts.getHasher(), + opts.getHasher().initialValue()); + + JsonResumableSession session = + ResumableSession.json( + HttpClientContext.from(storageRpc), + retrier.withAlg(retryAlgorithmManager.idempotent()), + jsonResumableWrite); + HttpContentRange contentRange = HttpContentRange.of(ByteRangeSpec.explicit(0L, size), size); + ResumableOperationResult put = + session.put(RewindableContent.of(path), contentRange); + // all exception translation is taken care of down in the JsonResumableSession + StorageObject object = put.getObject(); + if (object == null) { + // if by some odd chance the put didn't get the StorageObject, query for it + ResumableOperationResult<@Nullable StorageObject> query = session.query(); + object = query.getObject(); + } + return codecs.blobInfo().decode(object).asBlob(this); + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } - return codecs.blobInfo().decode(object).asBlob(this); } @Override @@ -335,21 +352,36 @@ private static class BucketPageFetcher implements NextPageFetcher { private final Map requestOptions; private final HttpStorageOptions serviceOptions; private final Retrier retrier; + @Nullable private final transient StorageMetricsContext parentMetricsContext; BucketPageFetcher( HttpStorageOptions serviceOptions, String cursor, Map optionMap, Retrier retrier) { + this(serviceOptions, cursor, optionMap, retrier, null); + } + + BucketPageFetcher( + HttpStorageOptions serviceOptions, + String cursor, + Map optionMap, + Retrier retrier, + @Nullable StorageMetricsContext parentMetricsContext) { this.requestOptions = PageImpl.nextRequestOptions(StorageRpc.Option.PAGE_TOKEN, cursor, optionMap); this.serviceOptions = serviceOptions; this.retrier = retrier; + this.parentMetricsContext = parentMetricsContext; } @Override public Page getNextPage() { - return listBuckets(serviceOptions, requestOptions, retrier); + return listBuckets( + serviceOptions, + requestOptions, + retrier, + parentMetricsContext != null ? parentMetricsContext.asSubAttemptContext() : null); } } @@ -360,6 +392,7 @@ private static class BlobPageFetcher implements NextPageFetcher { private final HttpStorageOptions serviceOptions; private final String bucket; private final Retrier retrier; + @Nullable private final transient StorageMetricsContext parentMetricsContext; BlobPageFetcher( String bucket, @@ -367,16 +400,32 @@ private static class BlobPageFetcher implements NextPageFetcher { String cursor, Map optionMap, Retrier retrier) { + this(bucket, serviceOptions, cursor, optionMap, retrier, null); + } + + BlobPageFetcher( + String bucket, + HttpStorageOptions serviceOptions, + String cursor, + Map optionMap, + Retrier retrier, + @Nullable StorageMetricsContext parentMetricsContext) { this.retrier = retrier; this.requestOptions = PageImpl.nextRequestOptions(StorageRpc.Option.PAGE_TOKEN, cursor, optionMap); this.serviceOptions = serviceOptions; this.bucket = bucket; + this.parentMetricsContext = parentMetricsContext; } @Override public Page getNextPage() { - return listBlobs(bucket, serviceOptions, requestOptions, retrier); + return listBlobs( + bucket, + serviceOptions, + requestOptions, + retrier, + parentMetricsContext != null ? parentMetricsContext.asSubAttemptContext() : null); } } @@ -387,21 +436,37 @@ private static class HmacKeyMetadataPageFetcher implements NextPageFetcher options; private final Retrier retrier; + @Nullable private final transient StorageMetricsContext parentMetricsContext; HmacKeyMetadataPageFetcher( HttpStorageOptions serviceOptions, HttpRetryAlgorithmManager retryAlgorithmManager, Map options, Retrier retrier) { + this(serviceOptions, retryAlgorithmManager, options, retrier, null); + } + + HmacKeyMetadataPageFetcher( + HttpStorageOptions serviceOptions, + HttpRetryAlgorithmManager retryAlgorithmManager, + Map options, + Retrier retrier, + @Nullable StorageMetricsContext parentMetricsContext) { this.serviceOptions = serviceOptions; this.retryAlgorithmManager = retryAlgorithmManager; this.options = options; this.retrier = retrier; + this.parentMetricsContext = parentMetricsContext; } @Override public Page getNextPage() { - return listHmacKeys(serviceOptions, retryAlgorithmManager, options, retrier); + return listHmacKeys( + serviceOptions, + retryAlgorithmManager, + options, + retrier, + parentMetricsContext != null ? parentMetricsContext.asSubAttemptContext() : null); } } @@ -421,8 +486,20 @@ private static Page listBuckets( final HttpStorageOptions serviceOptions, final Map optionsMap, Retrier retrier) { + return listBuckets(serviceOptions, optionsMap, retrier, null); + } + + private static Page listBuckets( + final HttpStorageOptions serviceOptions, + final Map optionsMap, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = serviceOptions.getRetryAlgorithmManager().getForBucketsList(optionsMap); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : serviceOptions.createMetricsContext("ListBuckets"); return retrier.run( algorithm, () -> serviceOptions.getStorageRpcV1().list(optionsMap), @@ -439,9 +516,11 @@ private static Page listBuckets( .decode(bucketPb) .asBucket(serviceOptions.getService())); return new PageImpl<>( - new BucketPageFetcher(serviceOptions, cursor, optionsMap, retrier), cursor, buckets); + new BucketPageFetcher(serviceOptions, cursor, optionsMap, retrier, activeContext), + cursor, + buckets); }, - serviceOptions.createMetricsContext("ListBuckets")); + activeContext); } private static Page listBlobs( @@ -449,8 +528,21 @@ private static Page listBlobs( final HttpStorageOptions serviceOptions, final Map optionsMap, Retrier retrier) { + return listBlobs(bucket, serviceOptions, optionsMap, retrier, null); + } + + private static Page listBlobs( + final String bucket, + final HttpStorageOptions serviceOptions, + final Map optionsMap, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = serviceOptions.getRetryAlgorithmManager().getForObjectsList(bucket, optionsMap); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : serviceOptions.createMetricsContext("ListObjects"); return retrier.run( algorithm, () -> serviceOptions.getStorageRpcV1().list(bucket, optionsMap), @@ -466,11 +558,12 @@ private static Page listBlobs( return info.asBlob(serviceOptions.getService()); }); return new PageImpl<>( - new BlobPageFetcher(bucket, serviceOptions, cursor, optionsMap, retrier), + new BlobPageFetcher( + bucket, serviceOptions, cursor, optionsMap, retrier, activeContext), cursor, blobs); }, - serviceOptions.createMetricsContext("ListObjects")); + activeContext); } @Override @@ -705,11 +798,23 @@ public CopyWriter copy(final CopyRequest copyRequest) { targetOptions, copyRequest.getMegabytesCopiedPerChunk()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsRewrite(rewriteRequest); - return run( - "RewriteObject", + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("RewriteObject"); + StorageMetricsContext initialContext = + multiContext != null + ? multiContext.asSubAttemptContext() + : getOptions().createMetricsContext("RewriteObject"); + return retrier.run( algorithm, () -> storageRpc.openRewrite(rewriteRequest), - (r) -> new HttpCopyWriter(getOptions(), r, retrier)); + (r) -> { + HttpCopyWriter writer = new HttpCopyWriter(getOptions(), r, retrier, multiContext); + if (r.isDone && multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), null); + } + return writer; + }, + initialContext); } @Override @@ -747,7 +852,9 @@ public StorageReadChannel reader(BlobId blob, BlobSourceOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blob); StorageObject storageObject = Conversions.json().blobId().encode(blob); ImmutableMap optionsMap = opts.getRpcOptions(); - return new BlobReadChannelV2(storageObject, optionsMap, BlobReadChannelContext.from(this)); + StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); + return new BlobReadChannelV2( + storageObject, optionsMap, BlobReadChannelContext.from(this), metricsContext); } @Override @@ -801,7 +908,10 @@ public StorageWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) 0, opts.getHasher(), opts.getHasher().initialValue()); - return new BlobWriteChannelV2(BlobReadChannelContext.from(this), jsonResumableWrite); + StorageMetricsContext metricsContext = + getOptions().createMultiStepMetricsContext("WriteObject"); + return new BlobWriteChannelV2( + BlobReadChannelContext.from(this), jsonResumableWrite, metricsContext); } @Override @@ -818,7 +928,10 @@ public StorageWriteChannel writer(URL signedURL) { getOptions(), signedURL, retrier.withAlg(forResumableUploadSessionCreate)); JsonResumableWrite jsonResumableWrite = JsonResumableWrite.of(signedUrlString, uploadIdSupplier.get(), 0); - return new BlobWriteChannelV2(BlobReadChannelContext.from(this), jsonResumableWrite); + StorageMetricsContext metricsContext = + getOptions().createMultiStepMetricsContext("WriteObject"); + return new BlobWriteChannelV2( + BlobReadChannelContext.from(this), jsonResumableWrite, metricsContext); } @Override @@ -1602,7 +1715,20 @@ private static Page listHmacKeys( final HttpRetryAlgorithmManager retryAlgorithmManager, final Map options, Retrier retrier) { + return listHmacKeys(serviceOptions, retryAlgorithmManager, options, retrier, null); + } + + private static Page listHmacKeys( + final HttpStorageOptions serviceOptions, + final HttpRetryAlgorithmManager retryAlgorithmManager, + final Map options, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyList(options); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : serviceOptions.createMetricsContext("ListHMACKeys"); return retrier.run( algorithm, () -> serviceOptions.getStorageRpcV1().listHmacKeys(options), @@ -1614,11 +1740,11 @@ private static Page listHmacKeys( : Iterables.transform(result.y(), codecs.hmacKeyMetadata()::decode); return new PageImpl<>( new HmacKeyMetadataPageFetcher( - serviceOptions, retryAlgorithmManager, options, retrier), + serviceOptions, retryAlgorithmManager, options, retrier, activeContext), cursor, metadata); }, - serviceOptions.createMetricsContext("ListHMACKeys")); + activeContext); } @Override