diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java index 465d3d6ae2f8..7ab63c19b7e7 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java @@ -43,13 +43,24 @@ abstract class BaseStorageReadChannel implements StorageReadChannel { private BufferHandle bufferHandle; private LazyReadChannel lazyReadChannel; protected final ReentrantLock lock; + @Nullable private final StorageMetricsContext metricsContext; + @Nullable private final StorageClientMetrics storageClientMetrics; protected BaseStorageReadChannel(Decoder objectDecoder) { + this(objectDecoder, null, null); + } + + protected BaseStorageReadChannel( + Decoder objectDecoder, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { this.objectDecoder = objectDecoder; this.result = SettableApiFuture.create(); this.open = true; this.byteRangeSpec = ByteRangeSpec.nullRange(); this.lock = new ReentrantLock(); + this.metricsContext = metricsContext; + this.storageClientMetrics = storageClientMetrics; } @Override @@ -76,14 +87,26 @@ public final boolean isOpen() { @Override public final void close() { lock.lock(); + Throwable error = null; try { - open = false; - if (internalGetLazyChannel().isOpen()) { - ReadableByteChannel channel = internalGetLazyChannel().getChannel(); - StorageException.wrapIOException(channel::close); + if (open) { + open = false; + if (internalGetLazyChannel().isOpen()) { + ReadableByteChannel channel = internalGetLazyChannel().getChannel(); + StorageException.wrapIOException(channel::close); + } } + } catch (Throwable t) { + error = t; + throw t; } finally { - lock.unlock(); + try { + if (metricsContext != null && storageClientMetrics != null) { + metricsContext.recordOperation(storageClientMetrics, error); + } + } finally { + lock.unlock(); + } } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java index 10f79c8dffdf..cd8d6597dc32 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java @@ -44,6 +44,8 @@ abstract class BaseStorageWriteChannel implements StorageWriteChannel { private int chunkSize; private LazyWriteChannel lazyWriteChannel; private BufferHandle bufferHandle; + @Nullable private final StorageMetricsContext metricsContext; + @Nullable private final StorageClientMetrics storageClientMetrics; /** * This is tracked for compatibility with BlobWriteChannel, such that simply creating a writer @@ -55,12 +57,21 @@ abstract class BaseStorageWriteChannel implements StorageWriteChannel { protected boolean writeCalledAtLeastOnce; protected BaseStorageWriteChannel(Decoder objectDecoder) { + this(objectDecoder, null, null); + } + + protected BaseStorageWriteChannel( + Decoder objectDecoder, + @Nullable StorageMetricsContext metricsContext, + @Nullable StorageClientMetrics storageClientMetrics) { this.objectDecoder = objectDecoder; this.result = SettableApiFuture.create(); this.lock = new ReentrantLock(); this.open = true; this.chunkSize = _16MiB; this.writeCalledAtLeastOnce = false; + this.metricsContext = metricsContext; + this.storageClientMetrics = storageClientMetrics; } @Override @@ -90,6 +101,7 @@ public final boolean isOpen() { @Override public final void close() throws IOException { lock.lock(); + Throwable error = null; try { if (open && !writeCalledAtLeastOnce) { this.write(ByteBuffer.allocate(0)); @@ -97,9 +109,18 @@ public final void close() throws IOException { if (internalGetLazyChannel().isOpen()) { StorageException.wrapIOException(internalGetLazyChannel().getChannel()::close); } + } catch (Throwable t) { + error = t; + throw t; } finally { open = false; - lock.unlock(); + try { + if (metricsContext != null && storageClientMetrics != null) { + metricsContext.recordOperation(storageClientMetrics, error); + } + } finally { + lock.unlock(); + } } } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java index 8df640b7490a..b1266b9afc43 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobReadChannelV2.java @@ -28,6 +28,7 @@ import java.io.Serializable; import java.util.Map; import java.util.Objects; +import org.checkerframework.checker.nullness.qual.Nullable; final class BlobReadChannelV2 extends BaseStorageReadChannel { @@ -40,7 +41,20 @@ final class BlobReadChannelV2 extends BaseStorageReadChannel { StorageObject storageObject, Map opts, BlobReadChannelContext blobReadChannelContext) { - super(Conversions.json().blobInfo()); + this(storageObject, opts, blobReadChannelContext, null); + } + + BlobReadChannelV2( + StorageObject storageObject, + Map opts, + BlobReadChannelContext blobReadChannelContext, + @Nullable StorageMetricsContext metricsContext) { + super( + Conversions.json().blobInfo(), + metricsContext, + blobReadChannelContext.getStorageOptions() != null + ? blobReadChannelContext.getStorageOptions().getStorageClientMetrics() + : null); this.storageObject = storageObject; this.opts = opts; this.blobReadChannelContext = blobReadChannelContext; diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java index edda3ce245e7..dc365d3eb012 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java @@ -25,6 +25,7 @@ import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Objects; +import org.checkerframework.checker.nullness.qual.Nullable; final class BlobWriteChannelV2 extends BaseStorageWriteChannel { @@ -32,7 +33,19 @@ final class BlobWriteChannelV2 extends BaseStorageWriteChannel { private final JsonResumableWrite start; BlobWriteChannelV2(BlobReadChannelContext blobChannelContext, JsonResumableWrite start) { - super(Conversions.json().blobInfo()); + this(blobChannelContext, start, null); + } + + BlobWriteChannelV2( + BlobReadChannelContext blobChannelContext, + JsonResumableWrite start, + @Nullable StorageMetricsContext metricsContext) { + super( + Conversions.json().blobInfo(), + metricsContext, + blobChannelContext.getStorageOptions() != null + ? blobChannelContext.getStorageOptions().getStorageClientMetrics() + : null); this.start = start; this.blobChannelContext = blobChannelContext; } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java index 6d8a24fc8eda..95063cc79f66 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpCopyWriter.java @@ -21,7 +21,6 @@ import com.google.cloud.storage.Retrying.DefaultRetrier; import com.google.cloud.storage.Retrying.HttpRetrier; import com.google.cloud.storage.Retrying.Retrier; -import com.google.cloud.storage.Retrying.RetryingDependencies; import com.google.cloud.storage.spi.v1.StorageRpc; import com.google.cloud.storage.spi.v1.StorageRpc.RewriteRequest; import com.google.cloud.storage.spi.v1.StorageRpc.RewriteResponse; @@ -30,6 +29,7 @@ import java.util.Map; import java.util.Objects; import java.util.function.UnaryOperator; +import org.checkerframework.checker.nullness.qual.Nullable; public class HttpCopyWriter extends CopyWriter { @@ -37,13 +37,23 @@ public class HttpCopyWriter extends CopyWriter { private final StorageRpc storageRpc; private RewriteResponse rewriteResponse; private final Retrier retrier; + @Nullable private final StorageMetricsContext metricsContext; HttpCopyWriter( HttpStorageOptions serviceOptions, RewriteResponse rewriteResponse, Retrier retrier) { + this(serviceOptions, rewriteResponse, retrier, null); + } + + HttpCopyWriter( + HttpStorageOptions serviceOptions, + RewriteResponse rewriteResponse, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { this.serviceOptions = serviceOptions; this.rewriteResponse = rewriteResponse; this.storageRpc = serviceOptions.getStorageRpcV1(); this.retrier = retrier; + this.metricsContext = metricsContext; } @Override @@ -51,6 +61,9 @@ public Blob getResult() { while (!isDone()) { copyChunk(); } + if (metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), null); + } BlobInfo info = Conversions.json().blobInfo().decode(rewriteResponse.result); return info.asBlob(serviceOptions.getService()); } @@ -74,11 +87,22 @@ public long getTotalBytesCopied() { public void copyChunk() { if (!isDone()) { RewriteRequest rewriteRequest = rewriteResponse.rewriteRequest; - this.rewriteResponse = + StorageMetricsContext chunkContext = + metricsContext != null + ? metricsContext.asSubAttemptContext() + : serviceOptions.createMetricsContext("RewriteObject"); + RewriteResponse resp = retrier.run( serviceOptions.getRetryAlgorithmManager().getForObjectsRewrite(rewriteRequest), () -> storageRpc.continueRewrite(rewriteResponse), - Decoder.identity()); + Decoder.identity(), + chunkContext); + if (resp != null) { + this.rewriteResponse = resp; + } + if (isDone() && metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), null); + } } } @@ -230,10 +254,7 @@ public CopyWriter restore() { totalBytesCopied); HttpRetrier httpRetrier = new HttpRetrier( - new DefaultRetrier( - UnaryOperator.identity(), - RetryingDependencies.simple( - serviceOptions.getClock(), serviceOptions.getRetrySettings()))); + new DefaultRetrier(UnaryOperator.identity(), serviceOptions.asRetryDependencies())); return new HttpCopyWriter(serviceOptions, rewriteResponse, httpRetrier); } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java index 7060db7e4719..2db4707d6d58 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java @@ -134,6 +134,11 @@ final class StorageImpl extends BaseService implements Storage, this.retrier = retrier; } + @Override + public void close() throws Exception { + getOptions().closeStorageClientMetrics(); + } + @Override public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) { final com.google.api.services.storage.model.Bucket bucketPb = @@ -143,6 +148,7 @@ public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsCreate(bucketPb, optionsMap); return run( + "CreateBucket", algorithm, () -> storageRpc.create(bucketPb, optionsMap), (b) -> Conversions.json().bucketInfo().decode(b).asBucket(this)); @@ -203,43 +209,55 @@ public Blob createFrom(BlobInfo blobInfo, Path path, int bufferSize, BlobWriteOp if (size == 0L) { return create(blobInfo, null, options); } - Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - final Map optionsMap = opts.getRpcOptions(); - BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null); - BlobInfo updated = opts.blobInfoMapper().apply(builder).build(); - StorageObject encode = codecs.blobInfo().encode(updated); - - Supplier uploadIdSupplier = - ResumableMedia.startUploadForBlobInfo( - getOptions(), - updated, - optionsMap, - retrier.withAlg(retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap))); - JsonResumableWrite jsonResumableWrite = - JsonResumableWrite.of( - encode, - optionsMap, - uploadIdSupplier.get(), - 0, - opts.getHasher(), - opts.getHasher().initialValue()); - - JsonResumableSession session = - ResumableSession.json( - HttpClientContext.from(storageRpc), - retrier.withAlg(retryAlgorithmManager.idempotent()), - jsonResumableWrite); - HttpContentRange contentRange = HttpContentRange.of(ByteRangeSpec.explicit(0L, size), size); - ResumableOperationResult put = - session.put(RewindableContent.of(path), contentRange); - // all exception translation is taken care of down in the JsonResumableSession - StorageObject object = put.getObject(); - if (object == null) { - // if by some odd chance the put didn't get the StorageObject, query for it - ResumableOperationResult<@Nullable StorageObject> query = session.query(); - object = query.getObject(); + StorageMetricsContext multiContext = getOptions().createMultiStepMetricsContext("WriteObject"); + Throwable error = null; + try { + Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); + final Map optionsMap = opts.getRpcOptions(); + BlobInfo.Builder builder = blobInfo.toBuilder().setMd5(null).setCrc32c(null); + BlobInfo updated = opts.blobInfoMapper().apply(builder).build(); + StorageObject encode = codecs.blobInfo().encode(updated); + + Supplier uploadIdSupplier = + ResumableMedia.startUploadForBlobInfo( + getOptions(), + updated, + optionsMap, + retrier.withAlg( + retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap))); + JsonResumableWrite jsonResumableWrite = + JsonResumableWrite.of( + encode, + optionsMap, + uploadIdSupplier.get(), + 0, + opts.getHasher(), + opts.getHasher().initialValue()); + + JsonResumableSession session = + ResumableSession.json( + HttpClientContext.from(storageRpc), + retrier.withAlg(retryAlgorithmManager.idempotent()), + jsonResumableWrite); + HttpContentRange contentRange = HttpContentRange.of(ByteRangeSpec.explicit(0L, size), size); + ResumableOperationResult put = + session.put(RewindableContent.of(path), contentRange); + // all exception translation is taken care of down in the JsonResumableSession + StorageObject object = put.getObject(); + if (object == null) { + // if by some odd chance the put didn't get the StorageObject, query for it + ResumableOperationResult<@Nullable StorageObject> query = session.query(); + object = query.getObject(); + } + return codecs.blobInfo().decode(object).asBlob(this); + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } - return codecs.blobInfo().decode(object).asBlob(this); } @Override @@ -319,6 +337,7 @@ public Blob restore(BlobId blob, BlobRestoreOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsRestore(obj, optionsMap); return run( + "RestoreObject", algorithm, () -> storageRpc.restore(obj, optionsMap), (x) -> { @@ -333,21 +352,36 @@ private static class BucketPageFetcher implements NextPageFetcher { private final Map requestOptions; private final HttpStorageOptions serviceOptions; private final Retrier retrier; + @Nullable private final transient StorageMetricsContext parentMetricsContext; BucketPageFetcher( HttpStorageOptions serviceOptions, String cursor, Map optionMap, Retrier retrier) { + this(serviceOptions, cursor, optionMap, retrier, null); + } + + BucketPageFetcher( + HttpStorageOptions serviceOptions, + String cursor, + Map optionMap, + Retrier retrier, + @Nullable StorageMetricsContext parentMetricsContext) { this.requestOptions = PageImpl.nextRequestOptions(StorageRpc.Option.PAGE_TOKEN, cursor, optionMap); this.serviceOptions = serviceOptions; this.retrier = retrier; + this.parentMetricsContext = parentMetricsContext; } @Override public Page getNextPage() { - return listBuckets(serviceOptions, requestOptions, retrier); + return listBuckets( + serviceOptions, + requestOptions, + retrier, + parentMetricsContext != null ? parentMetricsContext.asSubAttemptContext() : null); } } @@ -358,6 +392,7 @@ private static class BlobPageFetcher implements NextPageFetcher { private final HttpStorageOptions serviceOptions; private final String bucket; private final Retrier retrier; + @Nullable private final transient StorageMetricsContext parentMetricsContext; BlobPageFetcher( String bucket, @@ -365,16 +400,32 @@ private static class BlobPageFetcher implements NextPageFetcher { String cursor, Map optionMap, Retrier retrier) { + this(bucket, serviceOptions, cursor, optionMap, retrier, null); + } + + BlobPageFetcher( + String bucket, + HttpStorageOptions serviceOptions, + String cursor, + Map optionMap, + Retrier retrier, + @Nullable StorageMetricsContext parentMetricsContext) { this.retrier = retrier; this.requestOptions = PageImpl.nextRequestOptions(StorageRpc.Option.PAGE_TOKEN, cursor, optionMap); this.serviceOptions = serviceOptions; this.bucket = bucket; + this.parentMetricsContext = parentMetricsContext; } @Override public Page getNextPage() { - return listBlobs(bucket, serviceOptions, requestOptions, retrier); + return listBlobs( + bucket, + serviceOptions, + requestOptions, + retrier, + parentMetricsContext != null ? parentMetricsContext.asSubAttemptContext() : null); } } @@ -385,21 +436,37 @@ private static class HmacKeyMetadataPageFetcher implements NextPageFetcher options; private final Retrier retrier; + @Nullable private final transient StorageMetricsContext parentMetricsContext; HmacKeyMetadataPageFetcher( HttpStorageOptions serviceOptions, HttpRetryAlgorithmManager retryAlgorithmManager, Map options, Retrier retrier) { + this(serviceOptions, retryAlgorithmManager, options, retrier, null); + } + + HmacKeyMetadataPageFetcher( + HttpStorageOptions serviceOptions, + HttpRetryAlgorithmManager retryAlgorithmManager, + Map options, + Retrier retrier, + @Nullable StorageMetricsContext parentMetricsContext) { this.serviceOptions = serviceOptions; this.retryAlgorithmManager = retryAlgorithmManager; this.options = options; this.retrier = retrier; + this.parentMetricsContext = parentMetricsContext; } @Override public Page getNextPage() { - return listHmacKeys(serviceOptions, retryAlgorithmManager, options, retrier); + return listHmacKeys( + serviceOptions, + retryAlgorithmManager, + options, + retrier, + parentMetricsContext != null ? parentMetricsContext.asSubAttemptContext() : null); } } @@ -419,8 +486,20 @@ private static Page listBuckets( final HttpStorageOptions serviceOptions, final Map optionsMap, Retrier retrier) { + return listBuckets(serviceOptions, optionsMap, retrier, null); + } + + private static Page listBuckets( + final HttpStorageOptions serviceOptions, + final Map optionsMap, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = serviceOptions.getRetryAlgorithmManager().getForBucketsList(optionsMap); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : serviceOptions.createMetricsContext("ListBuckets"); return retrier.run( algorithm, () -> serviceOptions.getStorageRpcV1().list(optionsMap), @@ -437,8 +516,11 @@ private static Page listBuckets( .decode(bucketPb) .asBucket(serviceOptions.getService())); return new PageImpl<>( - new BucketPageFetcher(serviceOptions, cursor, optionsMap, retrier), cursor, buckets); - }); + new BucketPageFetcher(serviceOptions, cursor, optionsMap, retrier, activeContext), + cursor, + buckets); + }, + activeContext); } private static Page listBlobs( @@ -446,8 +528,21 @@ private static Page listBlobs( final HttpStorageOptions serviceOptions, final Map optionsMap, Retrier retrier) { + return listBlobs(bucket, serviceOptions, optionsMap, retrier, null); + } + + private static Page listBlobs( + final String bucket, + final HttpStorageOptions serviceOptions, + final Map optionsMap, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = serviceOptions.getRetryAlgorithmManager().getForObjectsList(bucket, optionsMap); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : serviceOptions.createMetricsContext("ListObjects"); return retrier.run( algorithm, () -> serviceOptions.getStorageRpcV1().list(bucket, optionsMap), @@ -463,10 +558,12 @@ private static Page listBlobs( return info.asBlob(serviceOptions.getService()); }); return new PageImpl<>( - new BlobPageFetcher(bucket, serviceOptions, cursor, optionsMap, retrier), + new BlobPageFetcher( + bucket, serviceOptions, cursor, optionsMap, retrier, activeContext), cursor, blobs); - }); + }, + activeContext); } @Override @@ -506,6 +603,7 @@ public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsUpdate(bucketPb, optionsMap); return run( + "UpdateBucket", algorithm, () -> storageRpc.patch(bucketPb, optionsMap), (x) -> Conversions.json().bucketInfo().decode(x).asBucket(this)); @@ -592,6 +690,7 @@ public Blob update(BlobInfo blobInfo, BlobTargetOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsUpdate(pb, optionsMap); return run( + "UpdateObject", algorithm, () -> storageRpc.patch(pb, optionsMap), (x) -> { @@ -613,7 +712,11 @@ public boolean delete(String bucket, BucketSourceOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsDelete(bucketPb, optionsMap); - return run(algorithm, () -> storageRpc.delete(bucketPb, optionsMap), Decoder.identity()); + return run( + "DeleteBucket", + algorithm, + () -> storageRpc.delete(bucketPb, optionsMap), + Decoder.identity()); } @Override @@ -628,7 +731,11 @@ public boolean delete(BlobId blob, BlobSourceOption... options) { Opts.unwrap(options).resolveFrom(blob).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsDelete(storageObject, optionsMap); - return run(algorithm, () -> storageRpc.delete(storageObject, optionsMap), Decoder.identity()); + return run( + "DeleteObject", + algorithm, + () -> storageRpc.delete(storageObject, optionsMap), + Decoder.identity()); } @Override @@ -660,6 +767,7 @@ public Blob compose(final ComposeRequest composeRequest) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsCompose(sources, targetPb, targetOptions); return run( + "ComposeObject", algorithm, () -> storageRpc.compose(sources, targetPb, targetOptions), (x) -> { @@ -690,10 +798,23 @@ public CopyWriter copy(final CopyRequest copyRequest) { targetOptions, copyRequest.getMegabytesCopiedPerChunk()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsRewrite(rewriteRequest); - return run( + StorageMetricsContext multiContext = + getOptions().createMultiStepMetricsContext("RewriteObject"); + StorageMetricsContext initialContext = + multiContext != null + ? multiContext.asSubAttemptContext() + : getOptions().createMetricsContext("RewriteObject"); + return retrier.run( algorithm, () -> storageRpc.openRewrite(rewriteRequest), - (r) -> new HttpCopyWriter(getOptions(), r, retrier)); + (r) -> { + HttpCopyWriter writer = new HttpCopyWriter(getOptions(), r, retrier, multiContext); + if (r.isDone && multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), null); + } + return writer; + }, + initialContext); } @Override @@ -709,7 +830,11 @@ public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) { ImmutableMap optionsMap = resolve.getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storageObject, optionsMap); - return run(algorithm, () -> storageRpc.load(storageObject, optionsMap), Decoder.identity()); + return run( + "ReadObject", + algorithm, + () -> storageRpc.load(storageObject, optionsMap), + Decoder.identity()); } @Override @@ -727,7 +852,9 @@ public StorageReadChannel reader(BlobId blob, BlobSourceOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blob); StorageObject storageObject = Conversions.json().blobId().encode(blob); ImmutableMap optionsMap = opts.getRpcOptions(); - return new BlobReadChannelV2(storageObject, optionsMap, BlobReadChannelContext.from(this)); + StorageMetricsContext metricsContext = getOptions().createMultiStepMetricsContext("ReadObject"); + return new BlobReadChannelV2( + storageObject, optionsMap, BlobReadChannelContext.from(this), metricsContext); } @Override @@ -747,6 +874,7 @@ public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption. Opts.unwrap(options).resolveFrom(blob).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(pb, optionsMap); run( + "ReadObject", algorithm, callable( () -> { @@ -780,7 +908,10 @@ public StorageWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) 0, opts.getHasher(), opts.getHasher().initialValue()); - return new BlobWriteChannelV2(BlobReadChannelContext.from(this), jsonResumableWrite); + StorageMetricsContext metricsContext = + getOptions().createMultiStepMetricsContext("WriteObject"); + return new BlobWriteChannelV2( + BlobReadChannelContext.from(this), jsonResumableWrite, metricsContext); } @Override @@ -797,7 +928,10 @@ public StorageWriteChannel writer(URL signedURL) { getOptions(), signedURL, retrier.withAlg(forResumableUploadSessionCreate)); JsonResumableWrite jsonResumableWrite = JsonResumableWrite.of(signedUrlString, uploadIdSupplier.get(), 0); - return new BlobWriteChannelV2(BlobReadChannelContext.from(this), jsonResumableWrite); + StorageMetricsContext metricsContext = + getOptions().createMultiStepMetricsContext("WriteObject"); + return new BlobWriteChannelV2( + BlobReadChannelContext.from(this), jsonResumableWrite, metricsContext); } @Override @@ -1279,7 +1413,11 @@ public Acl getAcl(final String bucket, final Entity entity, BucketSourceOption.. String pb = codecs.entity().encode(entity); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclGet(pb, optionsMap); - return run(algorithm, () -> storageRpc.getAcl(bucket, pb, optionsMap), codecs.bucketAcl()); + return run( + "GetBucketACL", + algorithm, + () -> storageRpc.getAcl(bucket, pb, optionsMap), + codecs.bucketAcl()); } @Override @@ -1293,7 +1431,11 @@ public boolean deleteAcl( final String pb = codecs.entity().encode(entity); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclDelete(pb, optionsMap); - return run(algorithm, () -> storageRpc.deleteAcl(bucket, pb, optionsMap), Decoder.identity()); + return run( + "DeleteBucketACL", + algorithm, + () -> storageRpc.deleteAcl(bucket, pb, optionsMap), + Decoder.identity()); } @Override @@ -1307,7 +1449,11 @@ public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclCreate(aclPb, optionsMap); - return run(algorithm, () -> storageRpc.createAcl(aclPb, optionsMap), codecs.bucketAcl()); + return run( + "CreateBucketACL", + algorithm, + () -> storageRpc.createAcl(aclPb, optionsMap), + codecs.bucketAcl()); } @Override @@ -1321,7 +1467,11 @@ public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclUpdate(aclPb, optionsMap); - return run(algorithm, () -> storageRpc.patchAcl(aclPb, optionsMap), codecs.bucketAcl()); + return run( + "UpdateBucketACL", + algorithm, + () -> storageRpc.patchAcl(aclPb, optionsMap), + codecs.bucketAcl()); } @Override @@ -1335,6 +1485,7 @@ public List listAcls(final String bucket, BucketSourceOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketAclList(bucket, optionsMap); return run( + "ListBucketACLs", algorithm, () -> storageRpc.listAcls(bucket, optionsMap), (answer) -> @@ -1354,34 +1505,51 @@ public List listAcls(final String bucket) { public Acl getDefaultAcl(final String bucket, final Entity entity) { String pb = codecs.entity().encode(entity); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclGet(pb); - return run(algorithm, () -> storageRpc.getDefaultAcl(bucket, pb), codecs.objectAcl()); + return run( + "GetDefaultObjectACL", + algorithm, + () -> storageRpc.getDefaultAcl(bucket, pb), + codecs.objectAcl()); } @Override public boolean deleteDefaultAcl(final String bucket, final Entity entity) { String pb = codecs.entity().encode(entity); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclDelete(pb); - return run(algorithm, () -> storageRpc.deleteDefaultAcl(bucket, pb), Decoder.identity()); + return run( + "DeleteDefaultObjectACL", + algorithm, + () -> storageRpc.deleteDefaultAcl(bucket, pb), + Decoder.identity()); } @Override public Acl createDefaultAcl(String bucket, Acl acl) { final ObjectAccessControl aclPb = codecs.objectAcl().encode(acl).setBucket(bucket); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclCreate(aclPb); - return run(algorithm, () -> storageRpc.createDefaultAcl(aclPb), codecs.objectAcl()); + return run( + "CreateDefaultObjectACL", + algorithm, + () -> storageRpc.createDefaultAcl(aclPb), + codecs.objectAcl()); } @Override public Acl updateDefaultAcl(String bucket, Acl acl) { final ObjectAccessControl aclPb = codecs.objectAcl().encode(acl).setBucket(bucket); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclUpdate(aclPb); - return run(algorithm, () -> storageRpc.patchDefaultAcl(aclPb), codecs.objectAcl()); + return run( + "UpdateDefaultObjectACL", + algorithm, + () -> storageRpc.patchDefaultAcl(aclPb), + codecs.objectAcl()); } @Override public List listDefaultAcls(final String bucket) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForDefaultObjectAclList(bucket); return run( + "ListDefaultObjectACLs", algorithm, () -> storageRpc.listDefaultAcls(bucket), (answer) -> @@ -1401,7 +1569,10 @@ public Acl getAcl(final BlobId blob, final Entity entity) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclGet(bucket, name, generation, pb); return run( - algorithm, () -> storageRpc.getAcl(bucket, name, generation, pb), codecs.objectAcl()); + "GetObjectACL", + algorithm, + () -> storageRpc.getAcl(bucket, name, generation, pb), + codecs.objectAcl()); } @Override @@ -1413,7 +1584,10 @@ public boolean deleteAcl(final BlobId blob, final Entity entity) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclDelete(bucket, name, generation, pb); return run( - algorithm, () -> storageRpc.deleteAcl(bucket, name, generation, pb), Decoder.identity()); + "DeleteObjectACL", + algorithm, + () -> storageRpc.deleteAcl(bucket, name, generation, pb), + Decoder.identity()); } @Override @@ -1426,7 +1600,7 @@ public Acl createAcl(final BlobId blob, final Acl acl) { .setObject(blob.getName()) .setGeneration(blob.getGeneration()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclCreate(aclPb); - return run(algorithm, () -> storageRpc.createAcl(aclPb), codecs.objectAcl()); + return run("CreateObjectACL", algorithm, () -> storageRpc.createAcl(aclPb), codecs.objectAcl()); } @Override @@ -1439,7 +1613,7 @@ public Acl updateAcl(BlobId blob, Acl acl) { .setObject(blob.getName()) .setGeneration(blob.getGeneration()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclUpdate(aclPb); - return run(algorithm, () -> storageRpc.patchAcl(aclPb), codecs.objectAcl()); + return run("UpdateObjectACL", algorithm, () -> storageRpc.patchAcl(aclPb), codecs.objectAcl()); } @Override @@ -1450,6 +1624,7 @@ public List listAcls(final BlobId blob) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectAclList(bucket, name, generation); return run( + "ListObjectACLs", algorithm, () -> storageRpc.listAcls(bucket, name, generation), (answer) -> @@ -1466,7 +1641,11 @@ public HmacKey createHmacKey( String pb = serviceAccount.getEmail(); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyCreate(pb, optionsMap); - return run(algorithm, () -> storageRpc.createHmacKey(pb, optionsMap), codecs.hmacKey()); + return run( + "CreateHMACKey", + algorithm, + () -> storageRpc.createHmacKey(pb, optionsMap), + codecs.hmacKey()); } @Override @@ -1481,7 +1660,10 @@ public HmacKeyMetadata getHmacKey(final String accessId, final GetHmacKeyOption. ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyGet(accessId, optionsMap); return run( - algorithm, () -> storageRpc.getHmacKey(accessId, optionsMap), codecs.hmacKeyMetadata()); + "GetHMACKey", + algorithm, + () -> storageRpc.getHmacKey(accessId, optionsMap), + codecs.hmacKeyMetadata()); } private HmacKeyMetadata updateHmacKey( @@ -1490,7 +1672,11 @@ private HmacKeyMetadata updateHmacKey( codecs.hmacKeyMetadata().encode(hmacKeyMetadata); ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyUpdate(pb, optionsMap); - return run(algorithm, () -> storageRpc.updateHmacKey(pb, optionsMap), codecs.hmacKeyMetadata()); + return run( + "UpdateHMACKey", + algorithm, + () -> storageRpc.updateHmacKey(pb, optionsMap), + codecs.hmacKeyMetadata()); } @Override @@ -1514,6 +1700,7 @@ public void deleteHmacKey(final HmacKeyMetadata metadata, final DeleteHmacKeyOpt ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyDelete(pb, optionsMap); run( + "DeleteHMACKey", algorithm, (Callable) () -> { @@ -1528,7 +1715,20 @@ private static Page listHmacKeys( final HttpRetryAlgorithmManager retryAlgorithmManager, final Map options, Retrier retrier) { + return listHmacKeys(serviceOptions, retryAlgorithmManager, options, retrier, null); + } + + private static Page listHmacKeys( + final HttpStorageOptions serviceOptions, + final HttpRetryAlgorithmManager retryAlgorithmManager, + final Map options, + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyList(options); + StorageMetricsContext activeContext = + metricsContext != null + ? metricsContext + : serviceOptions.createMetricsContext("ListHMACKeys"); return retrier.run( algorithm, () -> serviceOptions.getStorageRpcV1().listHmacKeys(options), @@ -1540,10 +1740,11 @@ private static Page listHmacKeys( : Iterables.transform(result.y(), codecs.hmacKeyMetadata()::decode); return new PageImpl<>( new HmacKeyMetadataPageFetcher( - serviceOptions, retryAlgorithmManager, options, retrier), + serviceOptions, retryAlgorithmManager, options, retrier, activeContext), cursor, metadata); - }); + }, + activeContext); } @Override @@ -1552,6 +1753,7 @@ public Policy getIamPolicy(final String bucket, BucketSourceOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsGetIamPolicy(bucket, optionsMap); return run( + "GetIamPolicy", algorithm, () -> storageRpc.getIamPolicy(bucket, optionsMap), apiPolicy -> Conversions.json().policyCodec().decode(apiPolicy)); @@ -1566,6 +1768,7 @@ public Policy setIamPolicy( ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsSetIamPolicy(bucket, pb, optionsMap); return run( + "SetIamPolicy", algorithm, () -> storageRpc.setIamPolicy(bucket, pb, optionsMap), apiPolicy -> Conversions.json().policyCodec().decode(apiPolicy)); @@ -1578,6 +1781,7 @@ public List testIamPermissions( ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsTestIamPermissions(bucket, permissions, optionsMap); return run( + "TestIamPermissions", algorithm, () -> storageRpc.testIamPermissions(bucket, permissions, optionsMap), (response) -> { @@ -1600,6 +1804,7 @@ public Bucket lockRetentionPolicy(BucketInfo bucketInfo, BucketTargetOption... o ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsLockRetentionPolicy(bucketPb, optionsMap); return run( + "LockBucketRetentionPolicy", algorithm, () -> storageRpc.lockRetentionPolicy(bucketPb, optionsMap), (x) -> Conversions.json().bucketInfo().decode(x).asBucket(this)); @@ -1608,11 +1813,16 @@ public Bucket lockRetentionPolicy(BucketInfo bucketInfo, BucketTargetOption... o @Override public ServiceAccount getServiceAccount(final String projectId) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForServiceAccountGet(projectId); - return run(algorithm, () -> storageRpc.getServiceAccount(projectId), codecs.serviceAccount()); + return run( + "GetServiceAccount", + algorithm, + () -> storageRpc.getServiceAccount(projectId), + codecs.serviceAccount()); } - private U run(ResultRetryAlgorithm algorithm, Callable c, Decoder f) { - return retrier.run(algorithm, c, f); + private U run( + String operationName, ResultRetryAlgorithm algorithm, Callable c, Decoder f) { + return retrier.run(algorithm, c, f, getOptions().createMetricsContext(operationName)); } @Override @@ -1623,6 +1833,7 @@ public Notification createNotification( ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationCreate(bucket, notificationPb); return run( + "CreateNotification", algorithm, () -> storageRpc.createNotification(bucket, notificationPb), n -> codecs.notificationInfo().decode(n).asNotification(this)); @@ -1633,6 +1844,7 @@ public Notification getNotification(final String bucket, final String notificati ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationGet(bucket, notificationId); return run( + "GetNotification", algorithm, () -> storageRpc.getNotification(bucket, notificationId), n -> codecs.notificationInfo().decode(n).asNotification(this)); @@ -1643,6 +1855,7 @@ public List listNotifications(final String bucket) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationList(bucket); List result = run( + "ListNotifications", algorithm, () -> storageRpc.listNotifications(bucket), (answer) -> @@ -1657,7 +1870,10 @@ public boolean deleteNotification(final String bucket, final String notification ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationDelete(bucket, notificationId); return run( - algorithm, () -> storageRpc.deleteNotification(bucket, notificationId), Decoder.identity()); + "DeleteNotification", + algorithm, + () -> storageRpc.deleteNotification(bucket, notificationId), + Decoder.identity()); } @Override @@ -1670,6 +1886,7 @@ private Blob internalGetBlob(BlobId blob, Map optionsMap) ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storedObject, optionsMap); return run( + "GetObject", algorithm, () -> storageRpc.get(storedObject, optionsMap), (x) -> { @@ -1684,6 +1901,7 @@ private Bucket internalBucketGet(String bucket, Map option ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsGet(bucketPb, optionsMap); return run( + "GetBucket", algorithm, () -> storageRpc.get(bucketPb, optionsMap), (b) -> Conversions.json().bucketInfo().decode(b).asBucket(this)); @@ -1708,6 +1926,7 @@ public Blob moveBlob(MoveBlobRequest request) { ImmutableMap targetOptions = dstOpts.getRpcOptions(); return run( + "MoveObject", retryAlgorithmManager.getForObjectsMove(sourceOptions, targetOptions), () -> storageRpc.moveObject( @@ -1783,6 +2002,7 @@ public BlobInfo internalDirectUpload(BlobInfo info, Opts opts, retryAlgorithmManager.getForObjectsCreate(encoded, optionsMap); RewindableContent content = RewindableContent.of(buf); return run( + "WriteObject", algorithm, () -> { content.rewindTo(0); @@ -1802,6 +2022,7 @@ public Void internalObjectDelete(BlobId id, Opts opts) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsDelete(storageObject, optionsMap); return run( + "DeleteObject", algorithm, () -> { boolean deleted = storageRpc.delete(storageObject, optionsMap); @@ -1821,6 +2042,7 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storedObject, optionsMap); return run( + "GetObject", algorithm, () -> { StorageObject storageObject = storageRpc.get(storedObject, optionsMap); diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java index dfcd8851ecfe..1e410d9d3655 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/CopyWriterTest.java @@ -17,6 +17,7 @@ package com.google.cloud.storage; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.times; @@ -34,6 +35,7 @@ import java.util.Map; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mockito; public class CopyWriterTest { @@ -210,4 +212,22 @@ public void testSaveAndRestoreWithResult() { assertEquals(42L, restoredRewriter.getBlobSize()); verify(storageRpcMock).continueRewrite(RESPONSE_WITH_OBJECT); } + + @Test + public void testCopyChunkPassesMetricsContext() { + HttpStorageOptions metricsOptions = + options.toBuilder() + .setEnableOtelMetrics(true) + .setMeterProvider(io.opentelemetry.sdk.metrics.SdkMeterProvider.builder().build()) + .build(); + Retrier retrierMock = Mockito.mock(Retrier.class); + copyWriter = new HttpCopyWriter(metricsOptions, RESPONSE_WITH_OBJECT, retrierMock); + copyWriter.copyChunk(); + ArgumentCaptor captor = + ArgumentCaptor.forClass(StorageMetricsContext.class); + verify(retrierMock).run(any(), any(), any(), captor.capture()); + StorageMetricsContext context = captor.getValue(); + assertNotNull(context); + assertEquals("RewriteObject", context.getOperationName()); + } }