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..7fbd90a5370f 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,7 +23,10 @@ import com.google.cloud.storage.Retrying.RetrierWithAlg; import com.google.storage.v2.RewriteObjectRequest; import com.google.storage.v2.RewriteResponse; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; +@NullMarked final class GapicCopyWriter extends CopyWriter { // needed for #getResult @@ -32,27 +35,45 @@ 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; + private boolean operationRecorded; GapicCopyWriter( GrpcStorageImpl storage, UnaryCallable callable, RetrierWithAlg retrier, RewriteObjectRequest originalRequest, - RewriteResponse initialResponse) { + 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; + this.operationRecorded = initialResponse.getDone(); + } + + private void recordOperation(@Nullable Throwable error) { + if (metricsContext != null && !operationRecorded) { + metricsContext.recordOperation(storage.getOptions().getStorageClientMetrics(), error); + operationRecorded = true; + } } @Override public Blob getResult() { - while (!isDone()) { - copyChunk(); + try { + while (!isDone()) { + copyChunk(); + } + recordOperation(null); + } catch (Throwable t) { + recordOperation(t); + throw t; } BlobInfo info = Conversions.grpc().blobInfo().decode(mostRecentResponse.getResource()); return info.asBlob(storage); @@ -76,10 +97,25 @@ public long getTotalBytesCopied() { @Override public void copyChunk() { if (!isDone()) { - RewriteObjectRequest req = - originalRequest.toBuilder().setRewriteToken(mostRecentResponse.getRewriteToken()).build(); - GrpcCallContext retryContext = Retrying.newCallContext(); - mostRecentResponse = retrier.run(() -> callable.call(req, retryContext), Decoder.identity()); + try { + RewriteObjectRequest req = + originalRequest.toBuilder() + .setRewriteToken(mostRecentResponse.getRewriteToken()) + .build(); + StorageMetricsContext chunkContext = metricsContext; + GrpcCallContext retryContext = Retrying.newCallContext(chunkContext); + RewriteResponse resp = + retrier.run(() -> callable.call(req, retryContext), Decoder.identity(), chunkContext); + if (resp != null) { + mostRecentResponse = resp; + } + if (isDone()) { + recordOperation(null); + } + } catch (Throwable t) { + recordOperation(t); + throw t; + } } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobReadChannel.java index 9dd39dfe47ea..98a0a6faca72 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.jspecify.annotations.Nullable; final class GrpcBlobReadChannel extends BaseStorageReadChannel { @@ -41,7 +42,18 @@ final class GrpcBlobReadChannel extends BaseStorageReadChannel { ResultRetryAlgorithm resultRetryAlgorithm, ReadObjectRequest request, boolean autoGzipDecompression) { - super(Conversions.grpc().blobInfo()); + this(read, retrier, resultRetryAlgorithm, request, autoGzipDecompression, null, null); + } + + GrpcBlobReadChannel( + ZeroCopyServerStreamingCallable read, + Retrier retrier, + ResultRetryAlgorithm resultRetryAlgorithm, + ReadObjectRequest request, + boolean autoGzipDecompression, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { + super(Conversions.grpc().blobInfo(), metricsContext, storageClientMetrics); this.read = read; this.retrier = retrier; this.resultRetryAlgorithm = resultRetryAlgorithm; @@ -61,7 +73,8 @@ protected LazyReadChannel newLazyReadChannel() { ReadableByteChannelSessionBuilder b = ResumableMedia.gapic() .read() - .byteChannel(read, retrier, resultRetryAlgorithm) + .byteChannel( + read, retrier.withMetricsContext(getMetricsContext()), resultRetryAlgorithm) .setHasher(Hasher.readHasher()) .setAutoGzipDecompression(autoGzipDecompression); BufferHandle bufferHandle = getBufferHandle(); diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcBlobWriteChannel.java index 3a7bc2a790c1..d8fa127b5fd4 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.jspecify.annotations.Nullable; final class GrpcBlobWriteChannel extends BaseStorageWriteChannel { @@ -37,7 +38,20 @@ final class GrpcBlobWriteChannel extends BaseStorageWriteChannel> start, Hasher hasher) { - super(Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource)); + this(write, retrier, start, hasher, null, null); + } + + GrpcBlobWriteChannel( + ClientStreamingCallable write, + RetrierWithAlg retrier, + Supplier> start, + Hasher hasher, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { + super( + Conversions.grpc().blobInfo().compose(WriteObjectResponse::getResource), + metricsContext, + storageClientMetrics); this.write = write; this.retrier = retrier; this.start = start; diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java index 120b7a269724..9103b847c80a 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 @@ -144,7 +144,7 @@ import java.util.stream.Stream; import java.util.stream.StreamSupport; import org.checkerframework.checker.nullness.qual.NonNull; -import org.checkerframework.checker.nullness.qual.Nullable; +import org.jspecify.annotations.Nullable; final class GrpcStorageImpl extends BaseService implements Storage, StorageInternal { @@ -209,6 +209,8 @@ public void close() throws Exception { s.shutdownNow(); java.time.Duration terminationAwaitDuration = getOptions().getTerminationAwaitDuration(); s.awaitTermination(terminationAwaitDuration.toMillis(), TimeUnit.MILLISECONDS); + } finally { + getOptions().closeStorageClientMetrics(); } } @@ -227,11 +229,14 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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 @@ -249,41 +254,50 @@ public Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... option public Blob create( BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - return internalDirectUpload(blobInfo, opts, ByteBuffer.wrap(content, offset, length)) + return internalDirectUpload( + blobInfo, + opts, + ByteBuffer.wrap(content, offset, length), + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "WriteObject")) .asBlob(this); } @Override public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) { - try { - requireNonNull(blobInfo, "blobInfo must be non null"); - InputStream inputStreamParam = firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)); + requireNonNull(blobInfo, "blobInfo must be non null"); + InputStream inputStreamParam = firstNonNull(content, new ByteArrayInputStream(ZERO_BYTES)); - Opts optsWithDefaults = Opts.unwrap(options).prepend(defaultOpts); - GrpcCallContext grpcCallContext = - optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); - WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); - Hasher hasher = optsWithDefaults.getHasher(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); - UnbufferedWritableByteChannelSession session = - ResumableMedia.gapic() - .write() - .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge)) - .setByteStringStrategy(ByteStringStrategy.noCopy()) - .setHasher(hasher) - .direct() - .unbuffered() - .setRequest(req) - .build(); + Opts optsWithDefaults = Opts.unwrap(options).prepend(defaultOpts); + GrpcCallContext grpcCallContext = + optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); + WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); + Hasher hasher = optsWithDefaults.getHasher(); + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "WriteObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + return retrier.run( + Retrying.neverRetry(), + () -> { + UnbufferedWritableByteChannelSession session = + ResumableMedia.gapic() + .write() + .byteChannel(storageClient.writeObjectCallable().withDefaultCallContext(merge)) + .setByteStringStrategy(ByteStringStrategy.noCopy()) + .setHasher(hasher) + .direct() + .unbuffered() + .setRequest(req) + .build(); - try (UnbufferedWritableByteChannel c = session.open()) { - ByteStreams.copy(Channels.newChannel(inputStreamParam), c); - } - ApiFuture responseApiFuture = session.getResult(); - return this.getBlob(responseApiFuture); - } catch (IOException | ApiException e) { - throw StorageException.coalesce(e); - } + try (UnbufferedWritableByteChannel c = session.open()) { + ByteStreams.copy(Channels.newChannel(inputStreamParam), c); + } + ApiFuture responseApiFuture = session.getResult(); + return this.getBlob(responseApiFuture); + }, + Decoder.identity(), + metricsContext); } @Override @@ -307,37 +321,51 @@ 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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "WriteObject"); + Throwable error = null; try { - GrpcResumableSession got = session2.get(); - ResumableOperationResult<@Nullable Object> put = got.put(RewindableContent.of(path)); - Object object = put.getObject(); - if (object == null) { - // if by some odd chance the put didn't get the Object, query for it - ResumableOperationResult<@Nullable Object> query = got.query(); - object = query.getObject(); + GrpcCallContext grpcCallContext = + opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); + WriteObjectRequest req = getWriteObjectRequest(info, opts); + + ClientStreamingCallable write = + storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext); + + ApiFuture start = + startResumableWrite(grpcCallContext, req, opts, multiContext); + ApiFuture session2 = + ApiFutures.transform( + start, + rw -> + ResumableSession.grpc( + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext), + write, + storageClient.queryWriteStatusCallable(), + rw, + opts.getHasher()), + MoreExecutors.directExecutor()); + try { + GrpcResumableSession got = session2.get(); + ResumableOperationResult<@Nullable Object> put = got.put(RewindableContent.of(path)); + Object object = put.getObject(); + if (object == null) { + // if by some odd chance the put didn't get the Object, query for it + ResumableOperationResult<@Nullable Object> query = got.query(); + object = query.getObject(); + } + return codecs.blobInfo().decode(object).asBlob(this); + } catch (InterruptedException | ExecutionException e) { + throw StorageException.coalesce(e); + } + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); } - return codecs.blobInfo().decode(object).asBlob(this); - } catch (InterruptedException | ExecutionException e) { - throw StorageException.coalesce(e); } } @@ -353,35 +381,49 @@ 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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "WriteObject"); + Throwable error = null; + try { + Opts opts = Opts.unwrap(options).resolveFrom(blobInfo).prepend(defaultOpts); + GrpcCallContext grpcCallContext = + opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); + WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); - ApiFuture start = startResumableWrite(grpcCallContext, req, opts); - - BufferedWritableByteChannelSession session = - ResumableMedia.gapic() - .write() - .byteChannel( - storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext)) - .setHasher(opts.getHasher()) - .setByteStringStrategy(ByteStringStrategy.noCopy()) - .resumable() - .withRetryConfig(retrier.withAlg(retryAlgorithmManager.idempotent())) - .buffered(Buffers.allocateAligned(bufferSize, _256KiB)) - .setStartAsync(start) - .build(); + ApiFuture start = + startResumableWrite(grpcCallContext, req, opts, multiContext); - // Specifically not in the try-with, so we don't close the provided stream - ReadableByteChannel src = - Channels.newChannel(firstNonNull(in, new ByteArrayInputStream(ZERO_BYTES))); - try (BufferedWritableByteChannel dst = session.open()) { - ByteStreams.copy(src, dst); - } catch (Exception e) { - throw StorageException.coalesce(e); + BufferedWritableByteChannelSession session = + ResumableMedia.gapic() + .write() + .byteChannel( + storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext)) + .setHasher(opts.getHasher()) + .setByteStringStrategy(ByteStringStrategy.noCopy()) + .resumable() + .withRetryConfig(retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext)) + .buffered(Buffers.allocateAligned(bufferSize, _256KiB)) + .setStartAsync(start) + .build(); + + // Specifically not in the try-with, so we don't close the provided stream + ReadableByteChannel src = + Channels.newChannel(firstNonNull(in, new ByteArrayInputStream(ZERO_BYTES))); + try (BufferedWritableByteChannel dst = session.open()) { + ByteStreams.copy(src, dst); + } catch (Exception e) { + throw StorageException.coalesce(e); + } + return getBlob(session.getResult()); + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } - return getBlob(session.getResult()); } @Override @@ -400,11 +442,15 @@ 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 = + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "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 +485,17 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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 +514,11 @@ public Page list(BucketListOption... options) { if (!request.getReturnPartialSuccess()) { try { - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "ListBuckets"); + GrpcCallContext merge = + Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(request), () -> storageClient.listBucketsPagedCallable().call(request, merge), @@ -473,14 +526,22 @@ public Page list(BucketListOption... options) { new TransformingPageDecorator<>( resp.getPage(), syntaxDecoders.bucket.andThen(opts.clearBucketFields()), - retrier, - retryAlgorithmManager.getFor(request))); + retrier.withMetricsContext( + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "ListBuckets")), + retryAlgorithmManager.getFor(request)), + metricsContext); } catch (Exception e) { throw StorageException.coalesce(e); } } else { try { - com.google.storage.v2.ListBucketsResponse response = listBuckets(grpcCallContext, request); + com.google.storage.v2.ListBucketsResponse response = + listBuckets( + grpcCallContext, + request, + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "ListBuckets")); return new ListBucketsWithPartialSuccessPage(grpcCallContext, request, response, opts); } catch (Exception e) { throw StorageException.coalesce(e); @@ -497,11 +558,14 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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); } @@ -511,7 +575,8 @@ public Page list(String bucket, BlobListOption... options) { public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) { Opts unwrap = Opts.unwrap(options); if (bucketInfo.getModifiedFields().isEmpty()) { - return internalBucketGet(bucketInfo.getName(), unwrap.constrainTo(BucketSourceOpt.class)); + return internalBucketGet( + "UpdateBucket", bucketInfo.getName(), unwrap.constrainTo(BucketSourceOpt.class)); } Opts opts = unwrap.resolveFrom(bucketInfo).prepend(defaultOpts); GrpcCallContext grpcCallContext = @@ -526,18 +591,22 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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 public Blob update(BlobInfo blobInfo, BlobTargetOption... options) { Opts unwrap = Opts.unwrap(options); if (blobInfo.getModifiedFields().isEmpty()) { - return internalBlobGet(blobInfo.getBlobId(), unwrap.constrainTo(ObjectSourceOpt.class)); + return internalBlobGet( + "UpdateObject", blobInfo.getBlobId(), unwrap.constrainTo(ObjectSourceOpt.class)); } Opts opts = unwrap.resolveFrom(blobInfo).prepend(defaultOpts); GrpcCallContext grpcCallContext = @@ -552,11 +621,14 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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,19 +644,27 @@ 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()); - return Boolean.TRUE.equals( - retrier.run( - retryAlgorithmManager.getFor(req), - () -> { - try { + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "DeleteBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + try { + return Boolean.TRUE.equals( + retrier.run( + retryAlgorithmManager.getFor(req), + () -> { storageClient.deleteBucketCallable().call(req, merge); return true; - } catch (NotFoundException e) { - return false; - } - }, - Decoder.identity())); + }, + Decoder.identity(), + metricsContext)); + } catch (NotFoundException e) { + return false; + } catch (StorageException e) { + if (e.getCode() == 404 || e.getCause() instanceof NotFoundException) { + return false; + } + throw e; + } } @Override @@ -596,12 +676,16 @@ public boolean delete(String bucket, String blob, BlobSourceOption... options) { public boolean delete(BlobId blob, BlobSourceOption... options) { Opts opts = Opts.unwrap(options); try { - internalObjectDelete(blob, opts); + internalObjectDelete( + blob, + opts, + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "DeleteObject")); return true; } catch (NotFoundException e) { return false; } catch (StorageException e) { - if (e.getCode() == 404) { + if (e.getCode() == 404 || e.getCause() instanceof NotFoundException) { return false; } throw e; @@ -615,6 +699,11 @@ public boolean delete(BlobId blob) { @Override public Void internalObjectDelete(BlobId id, Opts opts) { + return internalObjectDelete(id, opts, null); + } + + private Void internalObjectDelete( + BlobId id, Opts opts, @Nullable StorageMetricsContext metricsContext) { Opts finalOpts = opts.resolveFrom(id).prepend(defaultOpts); GrpcCallContext grpcCallContext = finalOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); @@ -624,14 +713,15 @@ 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()); + 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 +737,14 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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 +801,36 @@ public CopyWriter copy(CopyRequest copyRequest) { GrpcCallContext grpcCallContext = grpcCallContextMapper.apply(GrpcCallContext.createDefault()); UnaryCallable callable = storageClient.rewriteObjectCallable().withDefaultCallContext(grpcCallContext); - GrpcCallContext retryContext = Retrying.newCallContext(); - RetrierWithAlg retrierWithAlg = retrier.withAlg(retryAlgorithmManager.idempotent()); - return retrier.run( - retryAlgorithmManager.getFor(req), - () -> callable.call(req, retryContext), - (resp) -> new GapicCopyWriter(this, callable, retrierWithAlg, req, resp)); + StorageMetricsContext multiContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "RewriteObject"); + StorageMetricsContext initialContext = + multiContext != null + ? multiContext + : StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "RewriteObject"); + GrpcCallContext retryContext = Retrying.newCallContext(initialContext); + RetrierWithAlg retrierWithAlg = + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext); + try { + return retrier.run( + retryAlgorithmManager.getFor(req), + () -> callable.call(req, retryContext), + (resp) -> { + GapicCopyWriter writer = + new GapicCopyWriter(this, callable, retrierWithAlg, req, resp, multiContext); + if (resp.getDone() && multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), null); + } + return writer; + }, + initialContext); + } catch (Throwable t) { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } } @Override @@ -723,14 +840,27 @@ public byte[] readAllBytes(String bucket, String blob, BlobSourceOption... optio @Override public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) { - UnbufferedReadableByteChannelSession session = unbufferedReadSession(blob, options); + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "ReadObject"); + UnbufferedReadableByteChannelSession session = + unbufferedReadSession(blob, options, metricsContext); ByteArrayOutputStream baos = new ByteArrayOutputStream(); + Throwable error = null; try (UnbufferedReadableByteChannel r = session.open(); WritableByteChannel w = Channels.newChannel(baos)) { ByteStreams.copy(r, w); } catch (ApiException | IOException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } return baos.toByteArray(); } @@ -749,39 +879,72 @@ 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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "ReadObject"); + GrpcCallContext subAttemptContext = + metricsContext != null + ? Retrying.newCallContext(metricsContext) + : Retrying.newCallContext(); + GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(subAttemptContext); return new GrpcBlobReadChannel( readObjectCallable(grpcCallContext), retrier, retryAlgorithmManager.getFor(request), request, - !opts.autoGzipDecompression()); + !opts.autoGzipDecompression(), + metricsContext, + getOptions().getStorageClientMetrics()); } @Override public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) { + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "ReadObject"); + UnbufferedReadableByteChannelSession session = + unbufferedReadSession(blob, options, metricsContext); - UnbufferedReadableByteChannelSession session = unbufferedReadSession(blob, options); - + Throwable error = null; try (UnbufferedReadableByteChannel r = session.open(); WritableByteChannel w = Channels.newChannel(outputStream)) { ByteStreams.copy(r, w); } catch (ApiException | IOException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @@ -792,24 +955,46 @@ public GrpcBlobWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, opts); Hasher hasher = opts.getHasher(); - // in JSON, the starting of the resumable session happens before the invocation of write can - // happen. Emulate the same thing here. - // 1. create the future - ApiFuture startResumableWrite = startResumableWrite(grpcCallContext, req, opts); - // 2. await the result of the future - ResumableWrite resumableWrite = ApiFutureUtils.await(startResumableWrite); + StorageMetricsContext multiContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "WriteObject"); + ResumableWrite resumableWrite; + try { + // in JSON, the starting of the resumable session happens before the invocation of write can + // happen. Emulate the same thing here. + // 1. create the future + ApiFuture startResumableWrite = + startResumableWrite(grpcCallContext, req, opts, multiContext); + // 2. await the result of the future + resumableWrite = ApiFutureUtils.await(startResumableWrite); + } catch (Throwable t) { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } // 3. wrap the result in another future container before constructing the BlobWriteChannel ApiFuture wrapped = ApiFutures.immediateFuture(resumableWrite); return new GrpcBlobWriteChannel( storageClient.writeObjectCallable().withDefaultCallContext(grpcCallContext), - retrier.withAlg(retryAlgorithmManager.idempotent()), + retrier.withAlg(retryAlgorithmManager.idempotent(), multiContext), () -> wrapped, - hasher); + hasher, + multiContext, + getOptions().getStorageClientMetrics()); } @Override public BlobInfo internalDirectUpload( BlobInfo blobInfo, Opts opts, ByteBuffer buf) { + return internalDirectUpload(blobInfo, opts, buf, null); + } + + private BlobInfo internalDirectUpload( + BlobInfo blobInfo, + Opts opts, + ByteBuffer buf, + @Nullable StorageMetricsContext metricsContext) { requireNonNull(blobInfo, "blobInfo must be non null"); requireNonNull(buf, "content must be non null"); Opts optsWithDefaults = opts.prepend(defaultOpts); @@ -817,7 +1002,7 @@ public BlobInfo internalDirectUpload( optsWithDefaults.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); WriteObjectRequest req = getWriteObjectRequest(blobInfo, optsWithDefaults); Hasher hasher = opts.getHasher(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); RewindableContent content = RewindableContent.of(buf); return retrier.run( retryAlgorithmManager.getFor(req), @@ -839,7 +1024,8 @@ public BlobInfo internalDirectUpload( } return session.getResult(); }, - this::getBlob); + this::getBlob, + metricsContext); } @Override @@ -950,9 +1136,11 @@ public List delete(Iterable blobIds) { @Override public Acl getAcl(String bucket, Entity entity, BucketSourceOption... options) { + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "GetBucketACL"); try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, metricsContext); Predicate entityPredicate = bucketAclEntityOrAltEq(codecs.entity().encode(entity)); @@ -980,9 +1168,13 @@ public Acl getAcl(String bucket, Entity entity) { @Override public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options) { + StorageMetricsContext multiContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "DeleteBucketACL"); + Throwable error = null; try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = bucketAclEntityOrAltEq(encode); @@ -1000,20 +1192,29 @@ public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... opt UpdateBucketRequest req = createUpdateBucketAclRequest(bucket, newAcls, metageneration); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getAclList().stream().filter(entityPredicate).findFirst(); return !first.isPresent(); } catch (NotFoundException e) { // HttpStorageRpc returns false if the bucket doesn't exist :( + error = e; return false; } catch (StorageException se) { + error = se; if (se.getCode() == 404) { return false; } else { throw se; } + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @@ -1024,7 +1225,7 @@ public boolean deleteAcl(String bucket, Entity entity) { @Override public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) { - return updateAcl(bucket, acl, options); + return internalUpdateAcl("CreateBucketACL", bucket, acl, options); } @Override @@ -1034,9 +1235,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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), operationName); + Throwable error = null; try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, multiContext); BucketAccessControl encode = codecs.bucketAcl().encode(acl); String entity = encode.getEntity(); @@ -1050,7 +1260,7 @@ public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { UpdateBucketRequest req = createUpdateBucketAclRequest(bucket, newDefaultAcls, resp.getMetageneration()); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); Optional first = updateResult.getAclList().stream() @@ -1061,7 +1271,15 @@ public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { return first.orElseThrow( () -> new StorageException(0, "Acl update call success, but not in response")); } catch (NotFoundException e) { + error = e; throw StorageException.coalesce(e); + } catch (Exception e) { + error = e; + throw e; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } } @@ -1072,9 +1290,12 @@ public Acl updateAcl(String bucket, Acl acl) { @Override public List listAcls(String bucket, BucketSourceOption... options) { + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "ListBucketACLs"); try { Opts opts = Opts.unwrap(options).prepend(defaultOpts); - com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts); + com.google.storage.v2.Bucket resp = getBucketWithAcls(bucket, opts, metricsContext); return resp.getAclList().stream() .map(codecs.bucketAcl()::decode) .collect(ImmutableList.toImmutableList()); @@ -1090,8 +1311,11 @@ public List listAcls(String bucket) { @Override public Acl getDefaultAcl(String bucket, Entity entity) { + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "GetDefaultObjectACL"); try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, metricsContext); Predicate entityPredicate = objectAclEntityOrAltEq(codecs.entity().encode(entity)); @@ -1114,8 +1338,12 @@ public Acl getDefaultAcl(String bucket, Entity entity) { @Override public boolean deleteDefaultAcl(String bucket, Entity entity) { + StorageMetricsContext multiContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "DeleteDefaultObjectACL"); + Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1134,32 +1362,49 @@ public boolean deleteDefaultAcl(String bucket, Entity entity) { UpdateBucketRequest req = createUpdateDefaultAclRequest(bucket, newDefaultAcls, metageneration); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getDefaultObjectAclList().stream().filter(entityPredicate).findFirst(); return !first.isPresent(); } catch (NotFoundException e) { // HttpStorageRpc returns false if the bucket doesn't exist :( + error = e; return false; } catch (StorageException se) { + error = se; if (se.getCode() == 404) { return false; } else { 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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), operationName); + Throwable error = null; try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, multiContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1174,7 +1419,7 @@ public Acl updateDefaultAcl(String bucket, Acl acl) { UpdateBucketRequest req = createUpdateDefaultAclRequest(bucket, newDefaultAcls, resp.getMetageneration()); - com.google.storage.v2.Bucket updateResult = updateBucket(req); + com.google.storage.v2.Bucket updateResult = updateBucket(req, multiContext); Optional first = updateResult.getDefaultObjectAclList().stream() @@ -1185,14 +1430,25 @@ 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 = + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "ListDefaultObjectACLs"); try { - com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket); + com.google.storage.v2.Bucket resp = getBucketWithDefaultAcls(bucket, metricsContext); return resp.getDefaultObjectAclList().stream() .map(codecs.objectAcl()::decode) .collect(ImmutableList.toImmutableList()); @@ -1203,9 +1459,11 @@ public List listDefaultAcls(String bucket) { @Override public Acl getAcl(BlobId blob, Entity entity) { + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "GetObjectACL"); try { Object req = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(req); + Object resp = getObjectWithAcls(req, metricsContext); Predicate entityPredicate = objectAclEntityOrAltEq(codecs.entity().encode(entity)); @@ -1228,9 +1486,13 @@ public Acl getAcl(BlobId blob, Entity entity) { @Override public boolean deleteAcl(BlobId blob, Entity entity) { + StorageMetricsContext multiContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "DeleteObjectACL"); + Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj); + Object resp = getObjectWithAcls(obj, multiContext); String encode = codecs.entity().encode(entity); Predicate entityPredicate = objectAclEntityOrAltEq(encode); @@ -1248,33 +1510,50 @@ public boolean deleteAcl(BlobId blob, Entity entity) { UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, metageneration); - Object updateResult = updateObject(req); + Object updateResult = updateObject(req, multiContext); // read the response to ensure there is no longer an acl for the specified entity Optional first = updateResult.getAclList().stream().filter(entityPredicate).findFirst(); return !first.isPresent(); } catch (NotFoundException e) { // HttpStorageRpc returns false if the bucket doesn't exist :( + error = e; return false; } catch (StorageException se) { + error = se; if (se.getCode() == 404) { return false; } else { 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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), operationName); + Throwable error = null; try { Object obj = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(obj); + Object resp = getObjectWithAcls(obj, multiContext); ObjectAccessControl encode = codecs.objectAcl().encode(acl); String entity = encode.getEntity(); @@ -1288,7 +1567,7 @@ public Acl updateAcl(BlobId blob, Acl acl) { UpdateObjectRequest req = createUpdateObjectAclRequest(obj, newDefaultAcls, resp.getMetageneration()); - Object updateResult = updateObject(req); + Object updateResult = updateObject(req, multiContext); Optional first = updateResult.getAclList().stream() @@ -1299,15 +1578,26 @@ 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 = + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "ListObjectACLs"); try { Object req = codecs.blobId().encode(blob); - Object resp = getObjectWithAcls(req); + Object resp = getObjectWithAcls(req, metricsContext); return resp.getAclList().stream() .map(codecs.objectAcl()::decode) .collect(ImmutableList.toImmutableList()); @@ -1350,11 +1640,14 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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 +1660,14 @@ 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 = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "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 +1681,10 @@ public List testIamPermissions( .setResource(bucketNameCodec.encode(bucket)) .addAllPermissions(permissions) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "TestIamPermissions"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.testIamPermissionsCallable().call(req, merge), @@ -1394,7 +1693,8 @@ public List testIamPermissions( return permissions.stream() .map(heldPermissions::contains) .collect(ImmutableList.toImmutableList()); - }); + }, + metricsContext); } @Override @@ -1458,10 +1758,16 @@ public Blob moveBlob(MoveBlobRequest request) { dstOpts.moveObjectsRequest().apply(b); MoveObjectRequest req = b.build(); + GrpcCallContext grpcCallContext = + dstOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "MoveObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(req), - () -> storageClient.moveObjectCallable().call(req), - syntaxDecoders.blob); + () -> storageClient.moveObjectCallable().call(req, merge), + syntaxDecoders.blob, + metricsContext); } @Override @@ -1585,12 +1891,16 @@ public Page getNextPage() { ListObjectsRequest nextPageReq = req.toBuilder().setPageToken(resp.getNextPageToken()).build(); try { - GrpcCallContext merge = Utils.merge(ctx, Retrying.newCallContext()); + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "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); @@ -1667,7 +1977,12 @@ public Page getNextPage() { ListBucketsRequest nextPageReq = req.toBuilder().setPageToken(resp.getNextPageToken()).build(); try { - com.google.storage.v2.ListBucketsResponse nextPageResp = listBuckets(ctx, nextPageReq); + com.google.storage.v2.ListBucketsResponse nextPageResp = + listBuckets( + ctx, + nextPageReq, + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "ListBuckets")); return new ListBucketsWithPartialSuccessPage(ctx, nextPageReq, nextPageResp, opts); } catch (Exception e) { throw StorageException.coalesce(e); @@ -1737,10 +2052,17 @@ public String getNextPageToken() { return page.getNextPageToken(); } + @SuppressWarnings({"Convert2MethodRef"}) @Override public Page getNextPage() { - return new TransformingPageDecorator<>( - page.getNextPage(), translator, retrier, resultRetryAlgorithm); + if (!hasNextPage()) { + return null; + } + Callable c = () -> page.getNextPage(); + PageT nextPage = retrier.run(resultRetryAlgorithm, c, Decoder.identity()); + return nextPage == null + ? null + : new TransformingPageDecorator<>(nextPage, translator, retrier, resultRetryAlgorithm); } @SuppressWarnings({"Convert2MethodRef"}) @@ -1893,16 +2215,20 @@ BidiWriteObjectRequest getBidiWriteObjectRequestForTakeover( } private UnbufferedReadableByteChannelSession unbufferedReadSession( - BlobId blob, BlobSourceOption[] options) { + BlobId blob, BlobSourceOption[] options, @Nullable StorageMetricsContext metricsContext) { Opts opts = Opts.unwrap(options).resolveFrom(blob).prepend(defaultOpts); ReadObjectRequest readObjectRequest = getReadObjectRequest(blob, opts); - GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(Retrying.newCallContext()); + GrpcCallContext subAttemptContext = + metricsContext != null + ? Retrying.newCallContext(metricsContext) + : Retrying.newCallContext(); + GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(subAttemptContext); return ResumableMedia.gapic() .read() .byteChannel( readObjectCallable(grpcCallContext), - retrier, + retrier.withMetricsContext(metricsContext), retryAlgorithmManager.getFor(readObjectRequest)) .setAutoGzipDecompression(!opts.autoGzipDecompression()) .unbuffered() @@ -1913,14 +2239,27 @@ private UnbufferedReadableByteChannelSession unbufferedReadSession( @VisibleForTesting ApiFuture startResumableWrite( GrpcCallContext grpcCallContext, WriteObjectRequest req, Opts opts) { - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + return startResumableWrite(grpcCallContext, req, opts, null); + } + + ApiFuture startResumableWrite( + GrpcCallContext grpcCallContext, + WriteObjectRequest req, + Opts opts, + @Nullable StorageMetricsContext metricsContext) { + GrpcCallContext merge = + Utils.merge( + grpcCallContext, + metricsContext != null + ? Retrying.newCallContext(metricsContext) + : Retrying.newCallContext()); return ResumableMedia.gapic() .write() .resumableWrite( storageClient.startResumableWriteCallable().withDefaultCallContext(merge), req, opts, - retrier.withAlg(retryAlgorithmManager.getFor(req))); + retrier.withAlg(retryAlgorithmManager.getFor(req), metricsContext)); } ApiFuture startResumableWrite( @@ -1943,15 +2282,19 @@ private SourceObject sourceObjectEncode(SourceBlob from) { } private com.google.storage.v2.ListBucketsResponse listBuckets( - GrpcCallContext grpcCallContext, ListBucketsRequest request) { - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + GrpcCallContext grpcCallContext, + ListBucketsRequest request, + @Nullable StorageMetricsContext metricsContext) { + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); return retrier.run( retryAlgorithmManager.getFor(request), () -> storageClient.listBucketsCallable().call(request, merge), - Decoder.identity()); + Decoder.identity(), + metricsContext); } - private com.google.storage.v2.Bucket getBucketWithDefaultAcls(String bucketName) { + private com.google.storage.v2.Bucket getBucketWithDefaultAcls( + String bucketName, @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields( ImmutableSet.of( @@ -1966,15 +2309,23 @@ private com.google.storage.v2.Bucket getBucketWithDefaultAcls(String bucketName) .setName(bucketNameCodec.encode(bucketName)) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "GetBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } private com.google.storage.v2.Bucket getBucketWithAcls( - String bucketName, Opts opts) { + String bucketName, + Opts opts, + @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields(ImmutableSet.of(BucketField.ACL, BucketField.METAGENERATION)); GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); @@ -1985,20 +2336,33 @@ private com.google.storage.v2.Bucket getBucketWithAcls( .setName(bucketNameCodec.encode(bucketName)) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "GetBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } - private com.google.storage.v2.Bucket updateBucket(UpdateBucketRequest req) { + private com.google.storage.v2.Bucket updateBucket( + UpdateBucketRequest req, @Nullable StorageMetricsContext metricsContext) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "UpdateBucket"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.updateBucketCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } private static UpdateBucketRequest createUpdateDefaultAclRequest( @@ -2035,7 +2399,7 @@ private static UpdateBucketRequest createUpdateBucketAclRequest( .build(); } - private Object getObjectWithAcls(Object obj) { + private Object getObjectWithAcls(Object obj, @Nullable StorageMetricsContext metricsContext) { Fields fields = UnifiedOpts.fields(ImmutableSet.of(BucketField.ACL, BucketField.METAGENERATION)); GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); @@ -2047,11 +2411,17 @@ private Object getObjectWithAcls(Object obj) { .setObject(obj.getName()) .build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "GetObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.getObjectCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } private static UpdateObjectRequest createUpdateObjectAclRequest( @@ -2072,18 +2442,31 @@ private static UpdateObjectRequest createUpdateObjectAclRequest( .build(); } - private Object updateObject(UpdateObjectRequest req) { + private Object updateObject( + UpdateObjectRequest req, @Nullable StorageMetricsContext metricsContext) { GrpcCallContext grpcCallContext = GrpcCallContext.createDefault(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "UpdateObject"); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(activeContext)); return retrier.run( retryAlgorithmManager.getFor(req), () -> storageClient.updateObjectCallable().call(req, merge), - Decoder.identity()); + Decoder.identity(), + activeContext); } @NonNull @Override public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { + return internalObjectGet(blobId, opts, null); + } + + @NonNull + private BlobInfo internalObjectGet( + BlobId blobId, Opts opts, @Nullable StorageMetricsContext metricsContext) { Opts finalOpts = opts.prepend(defaultOpts).prepend(ALL_BLOB_FIELDS); GrpcCallContext grpcCallContext = finalOpts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); @@ -2093,7 +2476,7 @@ 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()); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); //noinspection DataFlowIssue return retrier.run( retryAlgorithmManager.getFor(req), @@ -2101,16 +2484,27 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { resp -> { BlobInfo tmp = codecs.blobInfo().decode(resp); return finalOpts.clearBlobFields().decode(tmp); - }); + }, + metricsContext); } @Nullable private Blob internalBlobGet(BlobId blob, Opts unwrap) { + return internalBlobGet("GetObject", blob, unwrap); + } + + @Nullable + private Blob internalBlobGet(String operationName, BlobId blob, Opts unwrap) { Opts opts = unwrap.resolveFrom(blob); try { - return internalObjectGet(blob, opts).asBlob(this); + return internalObjectGet( + blob, + opts, + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), operationName)) + .asBlob(this); } catch (StorageException e) { - if (e.getCause() instanceof NotFoundException) { + if (e.getCode() == 404 || e.getCause() instanceof NotFoundException) { return null; } else { throw e; @@ -2122,23 +2516,35 @@ private Blob internalBlobGet(BlobId blob, Opts unwrap) { @Nullable private Bucket internalBucketGet(String bucket, Opts unwrap) { + return internalBucketGet("GetBucket", bucket, unwrap); + } + + @Nullable + private Bucket internalBucketGet( + String operationName, String bucket, Opts unwrap) { Opts opts = unwrap.prepend(defaultOpts).prepend(ALL_BUCKET_FIELDS); GrpcCallContext grpcCallContext = opts.grpcMetadataMapper().apply(GrpcCallContext.createDefault()); GetBucketRequest.Builder builder = GetBucketRequest.newBuilder().setName(bucketNameCodec.encode(bucket)); GetBucketRequest req = opts.getBucketsRequest().apply(builder).build(); - GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext()); - return retrier.run( - retryAlgorithmManager.getFor(req), - () -> { - try { - return storageClient.getBucketCallable().call(req, merge); - } catch (NotFoundException e) { - return null; - } - }, - syntaxDecoders.bucket.andThen(opts.clearBucketFields())); + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), operationName); + GrpcCallContext merge = Utils.merge(grpcCallContext, Retrying.newCallContext(metricsContext)); + try { + return retrier.run( + retryAlgorithmManager.getFor(req), + () -> storageClient.getBucketCallable().call(req, merge), + syntaxDecoders.bucket.andThen(opts.clearBucketFields()), + metricsContext); + } catch (StorageException e) { + if (e.getCode() == 404 || e.getCause() instanceof NotFoundException) { + return null; + } + throw e; + } catch (NotFoundException e) { + return null; + } } private ZeroCopyServerStreamingCallable readObjectCallable( diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicCopyWriterTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicCopyWriterTest.java new file mode 100644 index 000000000000..2d2198be7b1e --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GapicCopyWriterTest.java @@ -0,0 +1,210 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.StorageClientMetrics.KEY_ERROR_TYPE; +import static com.google.cloud.storage.StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS; +import static com.google.cloud.storage.StorageErrorTypeMapper.STATUS_OK; +import static com.google.cloud.storage.StorageErrorTypeMapper.STATUS_UNAVAILABLE; +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.when; + +import com.google.api.gax.grpc.GrpcCallContext; +import com.google.api.gax.rpc.UnaryCallable; +import com.google.cloud.NoCredentials; +import com.google.cloud.storage.Retrying.Retrier; +import com.google.cloud.storage.Retrying.RetrierWithAlg; +import com.google.cloud.storage.UnifiedOpts.Opts; +import com.google.storage.v2.Object; +import com.google.storage.v2.RewriteObjectRequest; +import com.google.storage.v2.RewriteResponse; +import com.google.storage.v2.StorageClient; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.LongPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.util.Collection; +import org.junit.Before; +import org.junit.Test; + +public final class GapicCopyWriterTest { + + private static final String BUCKET = "test-bucket"; + private static final String OBJECT = "test-object"; + private static final RewriteObjectRequest REQUEST = + RewriteObjectRequest.newBuilder() + .setSourceBucket("projects/_/buckets/" + BUCKET) + .setSourceObject(OBJECT) + .setDestinationBucket("projects/_/buckets/" + BUCKET) + .setDestinationName("test-dest") + .build(); + private static final Object RESOURCE = + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET) + .setName("test-dest") + .setGeneration(1L) + .build(); + private static final RewriteResponse NOT_DONE_RESPONSE = + RewriteResponse.newBuilder() + .setDone(false) + .setObjectSize(100L) + .setTotalBytesRewritten(50L) + .setRewriteToken("token-1") + .build(); + private static final RewriteResponse DONE_RESPONSE = + RewriteResponse.newBuilder() + .setDone(true) + .setObjectSize(100L) + .setTotalBytesRewritten(100L) + .setResource(RESOURCE) + .build(); + + private InMemoryMetricReader reader; + private StorageClientMetrics metrics; + private GrpcStorageImpl storage; + private UnaryCallable callable; + private RetrierWithAlg retrier; + + @Before + @SuppressWarnings("unchecked") + public void setUp() { + reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + GrpcStorageOptions options = + GrpcStorageOptions.newBuilder() + .setMeterProvider(provider) + .setEnableOtelMetrics(true) + .setProjectId("test-project") + .setCredentials(NoCredentials.getInstance()) + .build(); + metrics = options.getStorageClientMetrics(); + + TestStorageStub storageStub = new TestStorageStub(); + StorageClient storageClient = StorageClient.create(storageStub); + storage = + new GrpcStorageImpl( + options, + storageClient, + null, + ResponseContentLifecycleManager.noop(), + null, + options.createRetrier(), + Opts.empty()); + + callable = TestStorageStub.mockCallable(); + retrier = Retrier.attemptOnce().withAlg(Retrying.neverRetry()); + } + + @Test + public void copyChunk_doesNotDoubleRecordOperationCompletionWhenDone() { + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext(metrics, "RewriteObject"); + when(callable.call(any(RewriteObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn(DONE_RESPONSE); + + GapicCopyWriter writer = + new GapicCopyWriter(storage, callable, retrier, REQUEST, NOT_DONE_RESPONSE, metricsContext); + + writer.copyChunk(); + assertThat(writer.isDone()).isTrue(); + assertOperationCount(STATUS_OK, 1L); + + // Subsequent copyChunk should be a no-op and not record another operation + writer.copyChunk(); + assertOperationCount(STATUS_OK, 1L); + + // getResult should return blob and not record another operation + Blob blob = writer.getResult(); + assertThat(blob).isNotNull(); + assertOperationCount(STATUS_OK, 1L); + } + + @Test + public void getResult_doesNotDoubleRecordWhenInitialResponseIsAlreadyDone() { + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext(metrics, "RewriteObject"); + + GapicCopyWriter writer = + new GapicCopyWriter(storage, callable, retrier, REQUEST, DONE_RESPONSE, metricsContext); + + assertThat(writer.isDone()).isTrue(); + Blob blob = writer.getResult(); + assertThat(blob).isNotNull(); + assertNoOperationsRecorded(); + } + + @Test + public void whenMetricsContextIsNull_internalChunksDoNotEmitTopLevelOperationMetrics() { + when(callable.call(any(RewriteObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn(DONE_RESPONSE); + + GapicCopyWriter writer = + new GapicCopyWriter(storage, callable, retrier, REQUEST, NOT_DONE_RESPONSE, null); + + writer.copyChunk(); + assertThat(writer.isDone()).isTrue(); + Blob blob = writer.getResult(); + assertThat(blob).isNotNull(); + + assertNoOperationsRecorded(); + } + + @Test + public void errorsInCopyChunk_recordFailureOnMetricsContext() { + StorageMetricsContext metricsContext = + StorageClientMetrics.createContext(metrics, "RewriteObject"); + when(callable.call(any(RewriteObjectRequest.class), any(GrpcCallContext.class))) + .thenThrow(new StorageException(503, "Service Unavailable")); + + GapicCopyWriter writer = + new GapicCopyWriter(storage, callable, retrier, REQUEST, NOT_DONE_RESPONSE, metricsContext); + + assertThrows(StorageException.class, writer::copyChunk); + assertOperationCount(STATUS_UNAVAILABLE, 1L); + } + + private void assertOperationCount(String errorType, long expectedCount) { + Collection collected = reader.collectAllMetrics(); + long count = 0; + for (MetricData md : collected) { + if (md.getName().equals(METRIC_GCP_STORAGE_CLIENT_OPERATIONS)) { + for (LongPointData point : md.getLongSumData().getPoints()) { + String val = point.getAttributes().get(KEY_ERROR_TYPE); + if (errorType.equals(val)) { + count += point.getValue(); + } + } + } + } + assertThat(count).isEqualTo(expectedCount); + } + + private void assertNoOperationsRecorded() { + Collection collected = reader.collectAllMetrics(); + long count = 0; + for (MetricData md : collected) { + if (md.getName().equals(METRIC_GCP_STORAGE_CLIENT_OPERATIONS)) { + for (LongPointData point : md.getLongSumData().getPoints()) { + count += point.getValue(); + } + } + } + assertThat(count).isEqualTo(0L); + } +} diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcStorageImplTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcStorageImplTest.java new file mode 100644 index 000000000000..396361de1436 --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/GrpcStorageImplTest.java @@ -0,0 +1,525 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import static com.google.cloud.storage.StorageClientMetrics.KEY_ERROR_TYPE; +import static com.google.cloud.storage.StorageClientMetrics.KEY_GCP_STORAGE_OPERATION; +import static com.google.cloud.storage.StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_ATTEMPTS; +import static com.google.cloud.storage.StorageClientMetrics.METRIC_GCP_STORAGE_CLIENT_OPERATIONS; +import static com.google.cloud.storage.StorageErrorTypeMapper.STATUS_NOT_FOUND; +import static com.google.cloud.storage.StorageErrorTypeMapper.STATUS_OK; +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.when; + +import com.google.api.gax.grpc.GrpcCallContext; +import com.google.api.gax.grpc.GrpcStatusCode; +import com.google.api.gax.paging.Page; +import com.google.api.gax.rpc.NotFoundException; +import com.google.api.gax.rpc.UnaryCallable; +import com.google.cloud.NoCredentials; +import com.google.cloud.storage.Acl.User; +import com.google.cloud.storage.Retrying.Retrier; +import com.google.cloud.storage.Storage.BucketListOption; +import com.google.cloud.storage.Storage.CopyRequest; +import com.google.cloud.storage.Storage.MoveBlobRequest; +import com.google.cloud.storage.UnifiedOpts.Opts; +import com.google.common.collect.ImmutableList; +import com.google.protobuf.Empty; +import com.google.storage.v2.Bucket; +import com.google.storage.v2.CreateBucketRequest; +import com.google.storage.v2.DeleteBucketRequest; +import com.google.storage.v2.DeleteObjectRequest; +import com.google.storage.v2.GetBucketRequest; +import com.google.storage.v2.GetObjectRequest; +import com.google.storage.v2.ListBucketsRequest; +import com.google.storage.v2.ListBucketsResponse; +import com.google.storage.v2.ListObjectsRequest; +import com.google.storage.v2.ListObjectsResponse; +import com.google.storage.v2.MoveObjectRequest; +import com.google.storage.v2.Object; +import com.google.storage.v2.RestoreObjectRequest; +import com.google.storage.v2.RewriteObjectRequest; +import com.google.storage.v2.RewriteResponse; +import com.google.storage.v2.StorageClient; +import com.google.storage.v2.StorageClient.ListBucketsPage; +import com.google.storage.v2.StorageClient.ListBucketsPagedResponse; +import com.google.storage.v2.UpdateObjectRequest; +import io.grpc.Status; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.data.LongPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; +import java.util.Collection; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +public final class GrpcStorageImplTest { + + private static final String BUCKET_NAME = "test-bucket"; + private static final String BLOB_NAME = "test-blob"; + + private InMemoryMetricReader reader; + private GrpcStorageOptions options; + private TestStorageStub storageStub; + private GrpcStorageImpl grpcStorage; + + @Before + public void setUp() { + reader = InMemoryMetricReader.create(); + SdkMeterProvider provider = SdkMeterProvider.builder().registerMetricReader(reader).build(); + options = + GrpcStorageOptions.newBuilder() + .setMeterProvider(provider) + .setEnableOtelMetrics(true) + .setProjectId("test-project") + .setCredentials(NoCredentials.getInstance()) + .build(); + + storageStub = new TestStorageStub(); + StorageClient storageClient = StorageClient.create(storageStub); + Retrier retrier = options.createRetrier(); + + grpcStorage = + new GrpcStorageImpl( + options, + storageClient, + null, + ResponseContentLifecycleManager.noop(), + null, + retrier, + Opts.empty()); + } + + @Test + @SuppressWarnings("unchecked") + public void unaryCalls_recordOperationMetrics() { + // 1. Create (Bucket) + UnaryCallable createBucketCallable = + TestStorageStub.mockCallable(); + storageStub.createBucketCallable = createBucketCallable; + when(createBucketCallable.call(any(CreateBucketRequest.class), any(GrpcCallContext.class))) + .thenReturn(Bucket.newBuilder().setName("projects/_/buckets/" + BUCKET_NAME).build()); + + com.google.cloud.storage.Bucket createdBucket = grpcStorage.create(BucketInfo.of(BUCKET_NAME)); + assertThat(createdBucket).isNotNull(); + assertOperationCount("CreateBucket", STATUS_OK, 1L); + + // 2. Get (Blob) + UnaryCallable getObjectCallable = TestStorageStub.mockCallable(); + storageStub.getObjectCallable = getObjectCallable; + when(getObjectCallable.call(any(GetObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn( + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET_NAME) + .setName(BLOB_NAME) + .build()); + + Blob retrievedBlob = grpcStorage.get(BlobId.of(BUCKET_NAME, BLOB_NAME)); + assertThat(retrievedBlob).isNotNull(); + assertOperationCount("GetObject", STATUS_OK, 1L); + + // 3. Delete (Blob) + UnaryCallable deleteObjectCallable = TestStorageStub.mockCallable(); + storageStub.deleteObjectCallable = deleteObjectCallable; + when(deleteObjectCallable.call(any(DeleteObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn(Empty.getDefaultInstance()); + + boolean deleted = grpcStorage.delete(BlobId.of(BUCKET_NAME, BLOB_NAME)); + assertThat(deleted).isTrue(); + assertOperationCount("DeleteObject", STATUS_OK, 1L); + + // 4. Update (Blob) + UnaryCallable updateObjectCallable = + TestStorageStub.mockCallable(); + storageStub.updateObjectCallable = updateObjectCallable; + when(updateObjectCallable.call(any(UpdateObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn( + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET_NAME) + .setName(BLOB_NAME) + .setContentType("text/plain") + .build()); + + Blob updatedBlob = + grpcStorage.update( + BlobInfo.newBuilder(BUCKET_NAME, BLOB_NAME).setContentType("text/plain").build()); + assertThat(updatedBlob).isNotNull(); + assertOperationCount("UpdateObject", STATUS_OK, 1L); + + // 5. Restore (Blob) + UnaryCallable restoreObjectCallable = + TestStorageStub.mockCallable(); + storageStub.restoreObjectCallable = restoreObjectCallable; + when(restoreObjectCallable.call(any(RestoreObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn( + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET_NAME) + .setName(BLOB_NAME) + .setGeneration(1L) + .build()); + + Blob restoredBlob = grpcStorage.restore(BlobId.of(BUCKET_NAME, BLOB_NAME, 1L)); + assertThat(restoredBlob).isNotNull(); + assertOperationCount("RestoreObject", STATUS_OK, 1L); + + // 6. MoveBlob + UnaryCallable moveObjectCallable = TestStorageStub.mockCallable(); + storageStub.moveObjectCallable = moveObjectCallable; + when(moveObjectCallable.call(any(MoveObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn( + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET_NAME) + .setName("dst-blob") + .build()); + + Blob movedBlob = + grpcStorage.moveBlob( + MoveBlobRequest.newBuilder() + .setSource(BlobId.of(BUCKET_NAME, BLOB_NAME)) + .setTarget(BlobId.of(BUCKET_NAME, "dst-blob")) + .build()); + assertThat(movedBlob).isNotNull(); + assertOperationCount("MoveObject", STATUS_OK, 1L); + } + + @Test + @SuppressWarnings("unchecked") + public void deleteBlob_onNonExistentObject_returnsFalseWithErrorTypeNotFound() { + UnaryCallable deleteObjectCallable = TestStorageStub.mockCallable(); + storageStub.deleteObjectCallable = deleteObjectCallable; + when(deleteObjectCallable.call(any(DeleteObjectRequest.class), any(GrpcCallContext.class))) + .thenThrow( + new NotFoundException( + "Not Found", null, GrpcStatusCode.of(Status.Code.NOT_FOUND), false)); + + boolean deleted = grpcStorage.delete(BlobId.of(BUCKET_NAME, "non-existent")); + assertThat(deleted).isFalse(); + + // Verify operations metric records STATUS_NOT_FOUND and not STATUS_OK + assertOperationCount("DeleteObject", STATUS_NOT_FOUND, 1L); + assertOperationCount("DeleteObject", STATUS_OK, 0L); + } + + @Test + public void deleteBucket_onNonExistentBucket_returnsFalseWithErrorTypeNotFound() { + UnaryCallable deleteBucketCallable = TestStorageStub.mockCallable(); + storageStub.deleteBucketCallable = deleteBucketCallable; + when(deleteBucketCallable.call(any(DeleteBucketRequest.class), any(GrpcCallContext.class))) + .thenThrow( + new NotFoundException( + "Not Found", null, GrpcStatusCode.of(Status.Code.NOT_FOUND), false)); + + boolean deleted = grpcStorage.delete(BUCKET_NAME); + assertThat(deleted).isFalse(); + + assertOperationCount("DeleteBucket", STATUS_NOT_FOUND, 1L); + assertOperationCount("DeleteBucket", STATUS_OK, 0L); + } + + @Test + public void getBucketAndBlob_onNonExistent_returnsNullWithErrorTypeNotFound() { + UnaryCallable getBucketCallable = TestStorageStub.mockCallable(); + storageStub.getBucketCallable = getBucketCallable; + when(getBucketCallable.call(any(GetBucketRequest.class), any(GrpcCallContext.class))) + .thenThrow( + new NotFoundException( + "Not Found", null, GrpcStatusCode.of(Status.Code.NOT_FOUND), false)); + + assertThat(grpcStorage.get(BUCKET_NAME)).isNull(); + assertOperationCount("GetBucket", STATUS_NOT_FOUND, 1L); + assertOperationCount("GetBucket", STATUS_OK, 0L); + + UnaryCallable getObjectCallable = TestStorageStub.mockCallable(); + storageStub.getObjectCallable = getObjectCallable; + when(getObjectCallable.call(any(GetObjectRequest.class), any(GrpcCallContext.class))) + .thenThrow( + new NotFoundException( + "Not Found", null, GrpcStatusCode.of(Status.Code.NOT_FOUND), false)); + + assertThat(grpcStorage.get(BlobId.of(BUCKET_NAME, "non-existent"))).isNull(); + assertOperationCount("GetObject", STATUS_NOT_FOUND, 1L); + assertOperationCount("GetObject", STATUS_OK, 0L); + } + + @Test + public void deleteAclAndDefaultAcl_onNonExistent_returnsFalseWithErrorTypeNotFound() { + UnaryCallable getBucketCallable = TestStorageStub.mockCallable(); + storageStub.getBucketCallable = getBucketCallable; + when(getBucketCallable.call(any(GetBucketRequest.class), any(GrpcCallContext.class))) + .thenThrow( + new NotFoundException( + "Not Found", null, GrpcStatusCode.of(Status.Code.NOT_FOUND), false)); + + UnaryCallable getObjectCallable = TestStorageStub.mockCallable(); + storageStub.getObjectCallable = getObjectCallable; + when(getObjectCallable.call(any(GetObjectRequest.class), any(GrpcCallContext.class))) + .thenThrow( + new NotFoundException( + "Not Found", null, GrpcStatusCode.of(Status.Code.NOT_FOUND), false)); + + User entity = User.ofAllAuthenticatedUsers(); + assertThat(grpcStorage.deleteAcl(BUCKET_NAME, entity)).isFalse(); + assertOperationCount("DeleteBucketACL", STATUS_NOT_FOUND, 1L); + assertOperationCount("DeleteBucketACL", STATUS_OK, 0L); + + assertThat(grpcStorage.deleteDefaultAcl(BUCKET_NAME, entity)).isFalse(); + assertOperationCount("DeleteDefaultObjectACL", STATUS_NOT_FOUND, 1L); + assertOperationCount("DeleteDefaultObjectACL", STATUS_OK, 0L); + + assertThat(grpcStorage.deleteAcl(BlobId.of(BUCKET_NAME, BLOB_NAME), entity)).isFalse(); + assertOperationCount("DeleteObjectACL", STATUS_NOT_FOUND, 1L); + assertOperationCount("DeleteObjectACL", STATUS_OK, 0L); + } + + @Test + public void pcuSubOperations_doNotEmitTopLevelOperations_andInternalObjectGetThrowsOn404() { + UnaryCallable getObjectCallable = TestStorageStub.mockCallable(); + storageStub.getObjectCallable = getObjectCallable; + when(getObjectCallable.call(any(GetObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn( + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET_NAME) + .setName(BLOB_NAME) + .build()) + .thenThrow( + new NotFoundException( + "Not Found", null, GrpcStatusCode.of(Status.Code.NOT_FOUND), false)); + + UnaryCallable deleteObjectCallable = TestStorageStub.mockCallable(); + storageStub.deleteObjectCallable = deleteObjectCallable; + when(deleteObjectCallable.call(any(DeleteObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn(Empty.getDefaultInstance()); + + BlobId blobId = BlobId.of(BUCKET_NAME, BLOB_NAME); + BlobInfo info = grpcStorage.internalObjectGet(blobId, Opts.empty()); + assertThat(info).isNotNull(); + grpcStorage.internalObjectDelete(blobId, Opts.empty()); + + assertOperationCount("GetObject", STATUS_OK, 0L); + assertOperationCount("DeleteObject", STATUS_OK, 0L); + + // Verify internalObjectGet throws on 404 rather than returning null + assertThrows(StorageException.class, () -> grpcStorage.internalObjectGet(blobId, Opts.empty())); + assertOperationCount("GetObject", STATUS_NOT_FOUND, 0L); + } + + @Test + @SuppressWarnings("unchecked") + public void listBucketsAndObjects_paginationRecordsSingleOperationAndMultipleAttempts() { + // 1. list(BucketListOption...) via TransformingPageDecorator (getNextPage & iterateAll) + UnaryCallable listBucketsPagedCallable = + TestStorageStub.mockCallable(); + storageStub.listBucketsPagedCallable = listBucketsPagedCallable; + + ListBucketsPagedResponse pagedResp = + Mockito.mock(ListBucketsPagedResponse.class, Mockito.withSettings().withoutAnnotations()); + ListBucketsPage page1 = + Mockito.mock(ListBucketsPage.class, Mockito.withSettings().withoutAnnotations()); + ListBucketsPage page2 = + Mockito.mock(ListBucketsPage.class, Mockito.withSettings().withoutAnnotations()); + Bucket bucketProto1 = Bucket.newBuilder().setName("projects/_/buckets/b1").build(); + Bucket bucketProto2 = Bucket.newBuilder().setName("projects/_/buckets/b2").build(); + + when(pagedResp.getPage()).thenReturn(page1); + when(page1.hasNextPage()).thenReturn(true); + when(page1.getNextPage()).thenReturn(page2); + when(page1.getValues()).thenReturn(ImmutableList.of(bucketProto1)); + when(page2.hasNextPage()).thenReturn(false); + when(page2.getValues()).thenReturn(ImmutableList.of(bucketProto2)); + when(listBucketsPagedCallable.call(any(ListBucketsRequest.class), any(GrpcCallContext.class))) + .thenReturn(pagedResp); + + Page firstBucketPage = grpcStorage.list(); + Page secondBucketPage = firstBucketPage.getNextPage(); + assertThat(secondBucketPage).isNotNull(); + assertOperationCount("ListBuckets", STATUS_OK, 1L); + assertAttemptCount("ListBuckets", 2L); + + // IterateAll also records attempt via bound retrier without incrementing operations + ImmutableList allBuckets = + ImmutableList.copyOf(firstBucketPage.iterateAll()); + assertThat(allBuckets).hasSize(2); + assertOperationCount("ListBuckets", STATUS_OK, 1L); + assertAttemptCount("ListBuckets", 3L); + + // 2. list(BucketListOption.returnPartialSuccess(true)) via ListBucketsWithPartialSuccessPage + UnaryCallable listBucketsCallable = + TestStorageStub.mockCallable(); + storageStub.listBucketsCallable = listBucketsCallable; + when(listBucketsCallable.call(any(ListBucketsRequest.class), any(GrpcCallContext.class))) + .thenReturn( + ListBucketsResponse.newBuilder() + .addBuckets(bucketProto1) + .setNextPageToken("token-2") + .build(), + ListBucketsResponse.newBuilder().addBuckets(bucketProto2).build()); + + Page partialPage1 = + grpcStorage.list(BucketListOption.returnPartialSuccess(true)); + Page partialPage2 = partialPage1.getNextPage(); + assertThat(partialPage2).isNotNull(); + assertOperationCount("ListBuckets", STATUS_OK, 2L); + assertAttemptCount("ListBuckets", 5L); + + // 3. list(String bucket, BlobListOption...) via ListObjectsWithSyntheticDirectoriesPage + UnaryCallable listObjectsCallable = + TestStorageStub.mockCallable(); + storageStub.listObjectsCallable = listObjectsCallable; + Object obj1 = + Object.newBuilder().setBucket("projects/_/buckets/" + BUCKET_NAME).setName("o1").build(); + Object obj2 = + Object.newBuilder().setBucket("projects/_/buckets/" + BUCKET_NAME).setName("o2").build(); + when(listObjectsCallable.call(any(ListObjectsRequest.class), any(GrpcCallContext.class))) + .thenReturn( + ListObjectsResponse.newBuilder() + .addObjects(obj1) + .setNextPageToken("obj-token-2") + .build(), + ListObjectsResponse.newBuilder().addObjects(obj2).build()); + + Page firstObjectPage = grpcStorage.list(BUCKET_NAME); + Page secondObjectPage = firstObjectPage.getNextPage(); + assertThat(secondObjectPage).isNotNull(); + assertOperationCount("ListObjects", STATUS_OK, 1L); + assertAttemptCount("ListObjects", 2L); + } + + @Test + public void updateBucketAndBlob_withNoModifiedFields_recordsUpdateOperationMetrics() { + UnaryCallable getBucketCallable = TestStorageStub.mockCallable(); + storageStub.getBucketCallable = getBucketCallable; + when(getBucketCallable.call(any(GetBucketRequest.class), any(GrpcCallContext.class))) + .thenReturn(Bucket.newBuilder().setName("projects/_/buckets/" + BUCKET_NAME).build()); + + UnaryCallable getObjectCallable = TestStorageStub.mockCallable(); + storageStub.getObjectCallable = getObjectCallable; + when(getObjectCallable.call(any(GetObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn( + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET_NAME) + .setName(BLOB_NAME) + .build()); + + BucketInfo unmodifiedBucket = BucketInfo.of(BUCKET_NAME); + assertThat(unmodifiedBucket.getModifiedFields()).isEmpty(); + com.google.cloud.storage.Bucket updatedBucket = grpcStorage.update(unmodifiedBucket); + assertThat(updatedBucket).isNotNull(); + assertOperationCount("UpdateBucket", STATUS_OK, 1L); + assertOperationCount("GetBucket", STATUS_OK, 0L); + + BlobInfo unmodifiedBlob = BlobInfo.newBuilder(BUCKET_NAME, BLOB_NAME).build(); + assertThat(unmodifiedBlob.getModifiedFields()).isEmpty(); + Blob updatedBlob = grpcStorage.update(unmodifiedBlob); + assertThat(updatedBlob).isNotNull(); + assertOperationCount("UpdateObject", STATUS_OK, 1L); + assertOperationCount("GetObject", STATUS_OK, 0L); + } + + @Test + @SuppressWarnings("unchecked") + public void copy_propagatesMultiContextToRetrierWithAlg() { + UnaryCallable rewriteObjectCallable = + TestStorageStub.mockCallable(); + storageStub.rewriteObjectCallable = rewriteObjectCallable; + when(rewriteObjectCallable.withDefaultCallContext(any(GrpcCallContext.class))) + .thenReturn(rewriteObjectCallable); + + RewriteResponse initialResp = + RewriteResponse.newBuilder() + .setDone(false) + .setRewriteToken("token-step-1") + .setTotalBytesRewritten(50L) + .setObjectSize(100L) + .build(); + RewriteResponse doneResp = + RewriteResponse.newBuilder() + .setDone(true) + .setTotalBytesRewritten(100L) + .setObjectSize(100L) + .setResource( + Object.newBuilder() + .setBucket("projects/_/buckets/" + BUCKET_NAME) + .setName("dst-blob") + .build()) + .build(); + + when(rewriteObjectCallable.call(any(RewriteObjectRequest.class), any(GrpcCallContext.class))) + .thenReturn(initialResp, doneResp); + + CopyWriter writer = + grpcStorage.copy( + CopyRequest.newBuilder() + .setSource(BUCKET_NAME, BLOB_NAME) + .setTarget(BlobId.of(BUCKET_NAME, "dst-blob")) + .build()); + + // Initial step returned done=false, so top-level operation is not recorded yet + assertThat(writer.isDone()).isFalse(); + assertOperationCount("RewriteObject", STATUS_OK, 0L); + + // Copying chunk completes rewrite using bound multiContext + writer.copyChunk(); + assertThat(writer.isDone()).isTrue(); + + // Multi-step operation is completed once + assertOperationCount("RewriteObject", STATUS_OK, 1L); + + // Calling getResult returns blob without double-recording + Blob result = writer.getResult(); + assertThat(result).isNotNull(); + assertOperationCount("RewriteObject", STATUS_OK, 1L); + + // Verify 2 attempts were recorded under RewriteObject + assertAttemptCount("RewriteObject", 2L); + } + + private void assertOperationCount(String operation, String errorType, long expectedCount) { + Collection collected = reader.collectAllMetrics(); + long count = 0; + for (MetricData md : collected) { + if (md.getName().equals(METRIC_GCP_STORAGE_CLIENT_OPERATIONS)) { + for (LongPointData point : md.getLongSumData().getPoints()) { + String op = point.getAttributes().get(KEY_GCP_STORAGE_OPERATION); + String err = point.getAttributes().get(KEY_ERROR_TYPE); + if (operation.equals(op) && errorType.equals(err)) { + count += point.getValue(); + } + } + } + } + assertThat(count).isEqualTo(expectedCount); + } + + private void assertAttemptCount(String operation, long expectedCount) { + Collection collected = reader.collectAllMetrics(); + long count = 0; + for (MetricData md : collected) { + if (md.getName().equals(METRIC_GCP_STORAGE_CLIENT_ATTEMPTS)) { + for (LongPointData point : md.getLongSumData().getPoints()) { + String op = point.getAttributes().get(KEY_GCP_STORAGE_OPERATION); + if (operation.equals(op)) { + count += point.getValue(); + } + } + } + } + assertThat(count).isEqualTo(expectedCount); + } +} diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/TestStorageStub.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/TestStorageStub.java new file mode 100644 index 000000000000..d113e469f216 --- /dev/null +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/TestStorageStub.java @@ -0,0 +1,144 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.google.cloud.storage; + +import com.google.api.gax.rpc.UnaryCallable; +import com.google.protobuf.Empty; +import com.google.storage.v2.Bucket; +import com.google.storage.v2.CreateBucketRequest; +import com.google.storage.v2.DeleteBucketRequest; +import com.google.storage.v2.DeleteObjectRequest; +import com.google.storage.v2.GetBucketRequest; +import com.google.storage.v2.GetObjectRequest; +import com.google.storage.v2.ListBucketsRequest; +import com.google.storage.v2.ListBucketsResponse; +import com.google.storage.v2.ListObjectsRequest; +import com.google.storage.v2.ListObjectsResponse; +import com.google.storage.v2.MoveObjectRequest; +import com.google.storage.v2.Object; +import com.google.storage.v2.RestoreObjectRequest; +import com.google.storage.v2.RewriteObjectRequest; +import com.google.storage.v2.RewriteResponse; +import com.google.storage.v2.StorageClient.ListBucketsPagedResponse; +import com.google.storage.v2.UpdateObjectRequest; +import com.google.storage.v2.stub.StorageStub; +import java.util.concurrent.TimeUnit; +import org.mockito.Mockito; + +class TestStorageStub extends StorageStub { + UnaryCallable createBucketCallable; + UnaryCallable getBucketCallable; + UnaryCallable deleteBucketCallable; + UnaryCallable listBucketsCallable; + UnaryCallable listBucketsPagedCallable; + UnaryCallable getObjectCallable; + UnaryCallable deleteObjectCallable; + UnaryCallable updateObjectCallable; + UnaryCallable restoreObjectCallable; + UnaryCallable moveObjectCallable; + UnaryCallable rewriteObjectCallable; + UnaryCallable listObjectsCallable; + + @SuppressWarnings("unchecked") + static UnaryCallable mockCallable() { + return Mockito.mock(UnaryCallable.class, Mockito.withSettings().withoutAnnotations()); + } + + @Override + public UnaryCallable createBucketCallable() { + return createBucketCallable; + } + + @Override + public UnaryCallable getBucketCallable() { + return getBucketCallable; + } + + @Override + public UnaryCallable deleteBucketCallable() { + return deleteBucketCallable; + } + + @Override + public UnaryCallable listBucketsCallable() { + return listBucketsCallable; + } + + @Override + public UnaryCallable listBucketsPagedCallable() { + return listBucketsPagedCallable; + } + + @Override + public UnaryCallable getObjectCallable() { + return getObjectCallable; + } + + @Override + public UnaryCallable deleteObjectCallable() { + return deleteObjectCallable; + } + + @Override + public UnaryCallable updateObjectCallable() { + return updateObjectCallable; + } + + @Override + public UnaryCallable restoreObjectCallable() { + return restoreObjectCallable; + } + + @Override + public UnaryCallable moveObjectCallable() { + return moveObjectCallable; + } + + @Override + public UnaryCallable rewriteObjectCallable() { + return rewriteObjectCallable; + } + + @Override + public UnaryCallable listObjectsCallable() { + return listObjectsCallable; + } + + @Override + public void shutdown() {} + + @Override + public boolean isShutdown() { + return false; + } + + @Override + public boolean isTerminated() { + return false; + } + + @Override + public void shutdownNow() {} + + @Override + public boolean awaitTermination(long duration, TimeUnit unit) throws InterruptedException { + return false; + } + + @Override + public void close() {} +}