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..96eef8a1d635 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,25 @@ 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; + @Nullable private Throwable lastOperationError; 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 +88,27 @@ 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 != null ? error : lastOperationError); + } + } finally { + lock.unlock(); + } } } @@ -140,12 +165,16 @@ public final int read(ByteBuffer dst) throws IOException { // be removed in the next major version. return -1; } else { + lastOperationError = e; throw new IOException(e); } } catch (IOException e) { + lastOperationError = e; throw e; } catch (Exception e) { - throw new IOException(StorageException.coalesce(e)); + Throwable coalesced = StorageException.coalesce(e); + lastOperationError = coalesced; + throw new IOException(coalesced); } } finally { lock.unlock(); @@ -177,6 +206,10 @@ protected final T getResolvedObject() { } } + protected final @Nullable StorageMetricsContext getMetricsContext() { + return metricsContext; + } + protected abstract LazyReadChannel newLazyReadChannel(); private void maybeResetChannel(boolean freeBuffer) throws IOException { 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..f3c82cbde0e3 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,9 @@ abstract class BaseStorageWriteChannel implements StorageWriteChannel { private int chunkSize; private LazyWriteChannel lazyWriteChannel; private BufferHandle bufferHandle; + @Nullable private final StorageMetricsContext metricsContext; + @Nullable private final StorageClientMetrics storageClientMetrics; + @Nullable private Throwable lastOperationError; /** * This is tracked for compatibility with BlobWriteChannel, such that simply creating a writer @@ -55,12 +58,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 +102,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 +110,19 @@ 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 != null ? error : lastOperationError); + } + } finally { + lock.unlock(); + } } } @@ -119,11 +142,15 @@ public final int write(ByteBuffer src) throws IOException { int write = tmp.write(src); return write; } catch (StorageException e) { + lastOperationError = e; throw new IOException(e); } catch (IOException e) { + lastOperationError = e; throw e; } catch (Exception e) { - throw new IOException(StorageException.coalesce(e)); + Throwable coalesced = StorageException.coalesce(e); + lastOperationError = coalesced; + throw new IOException(coalesced); } } finally { lock.unlock(); @@ -167,6 +194,10 @@ protected final void setOpen(boolean isOpen) { this.open = isOpen; } + protected final @Nullable StorageMetricsContext getMetricsContext() { + return metricsContext; + } + protected abstract LazyWriteChannel newLazyWriteChannel(); private LazyWriteChannel internalGetLazyChannel() { 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..d0ce12c99a70 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; @@ -69,7 +83,11 @@ protected LazyReadChannel newLazyReadChannel() { ReadableByteChannelSessionBuilder b = ResumableMedia.http() .read() - .byteChannel(blobReadChannelContext) + .byteChannel( + blobReadChannelContext.withRetrier( + blobReadChannelContext + .getRetrier() + .withMetricsContext(getMetricsContext()))) .setAutoGzipDecompression(autoGzipDecompression); BufferHandle bufferHandle = getBufferHandle(); // because we're erasing the specific type of channel, we need to declare it here. @@ -109,7 +127,11 @@ private BlobReadChannelV2State( public ReadChannel restore() { BlobReadChannelV2 channel = new BlobReadChannelV2( - request.getObject(), request.getOptions(), BlobReadChannelContext.from(options)); + request.getObject(), + request.getOptions(), + BlobReadChannelContext.from(options), + StorageClientMetrics.createMultiStepContext( + options.getStorageClientMetrics(), "ReadObject")); channel.setByteRangeSpec(request.getByteRangeSpec()); if (chunkSize != null) { channel.setChunkSize(chunkSize); @@ -186,6 +208,11 @@ public Retrier getRetrier() { return retrier; } + BlobReadChannelContext withRetrier(Retrier retrier) { + return new BlobReadChannelContext( + storageOptions, retryAlgorithmManager, httpClientContext, apiaryClient, retrier); + } + /** * This method is pretty unsafe, but so is all of the Capture/Restore API, and it leaks its * sludge all over everything. In general, prefer {@link 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..f88977772026 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; } @@ -78,7 +91,9 @@ protected LazyWriteChannel newLazyWriteChannel() { .withRetryConfig( blobChannelContext .getRetrier() - .withAlg(blobChannelContext.getRetryAlgorithmManager().idempotent())) + .withAlg( + blobChannelContext.getRetryAlgorithmManager().idempotent(), + getMetricsContext())) .setHasher(start.getHasher()) .buffered(getBufferHandle()) .setStartAsync(ApiFutures.immediateFuture(start)) @@ -120,7 +135,11 @@ public WriteChannel restore() { resumableWrite = resumableWrite.withBeginOffset(position); } BlobWriteChannelV2 channel = - new BlobWriteChannelV2(BlobReadChannelContext.from(options), resumableWrite); + new BlobWriteChannelV2( + BlobReadChannelContext.from(options), + resumableWrite, + StorageClientMetrics.createMultiStepContext( + options.getStorageClientMetrics(), "WriteObject")); if (chunkSize != null) { channel.setChunkSize(chunkSize); } 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..7786acfda2d6 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,22 +37,47 @@ 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, + StorageClientMetrics.createMultiStepContext( + serviceOptions.getStorageClientMetrics(), "RewriteObject")); + } + + 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 public Blob getResult() { - while (!isDone()) { - copyChunk(); + try { + while (!isDone()) { + copyChunk(); + } + if (metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), null); + } + BlobInfo info = Conversions.json().blobInfo().decode(rewriteResponse.result); + return info.asBlob(serviceOptions.getService()); + } catch (Throwable t) { + if (metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), t); + } + throw t; } - BlobInfo info = Conversions.json().blobInfo().decode(rewriteResponse.result); - return info.asBlob(serviceOptions.getService()); } @Override @@ -72,13 +97,32 @@ public long getTotalBytesCopied() { @Override public void copyChunk() { - if (!isDone()) { - RewriteRequest rewriteRequest = rewriteResponse.rewriteRequest; - this.rewriteResponse = - retrier.run( - serviceOptions.getRetryAlgorithmManager().getForObjectsRewrite(rewriteRequest), - () -> storageRpc.continueRewrite(rewriteResponse), - Decoder.identity()); + try { + if (!isDone()) { + RewriteRequest rewriteRequest = rewriteResponse.rewriteRequest; + StorageMetricsContext chunkContext = + metricsContext != null + ? metricsContext + : StorageClientMetrics.createContext( + serviceOptions.getStorageClientMetrics(), "RewriteObject"); + RewriteResponse resp = + retrier.run( + serviceOptions.getRetryAlgorithmManager().getForObjectsRewrite(rewriteRequest), + () -> storageRpc.continueRewrite(rewriteResponse), + Decoder.identity(), + chunkContext); + if (resp != null) { + this.rewriteResponse = resp; + } + if (isDone() && metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), null); + } + } + } catch (Throwable t) { + if (metricsContext != null) { + metricsContext.recordOperation(serviceOptions.getStorageClientMetrics(), t); + } + throw t; } } @@ -232,9 +276,14 @@ public CopyWriter restore() { new HttpRetrier( new DefaultRetrier( UnaryOperator.identity(), - RetryingDependencies.simple( - serviceOptions.getClock(), serviceOptions.getRetrySettings()))); - return new HttpCopyWriter(serviceOptions, rewriteResponse, httpRetrier); + serviceOptions.asRetryDependencies(), + serviceOptions.getStorageClientMetrics())); + return new HttpCopyWriter( + serviceOptions, + rewriteResponse, + httpRetrier, + StorageClientMetrics.createMultiStepContext( + serviceOptions.getStorageClientMetrics(), "RewriteObject")); } @Override 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..a59e23665727 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)); @@ -151,21 +157,38 @@ public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) { @Override public Blob create(BlobInfo blobInfo, BlobTargetOption... options) { Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - return internalDirectUpload(blobInfo, opts, Buffers.allocate(0)).asBlob(this); + return internalDirectUpload( + blobInfo, + opts, + Buffers.allocate(0), + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "WriteObject")) + .asBlob(this); } @Override public Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... options) { content = firstNonNull(content, EMPTY_BYTE_ARRAY); Opts opts = Opts.unwrap(options).resolveFrom(blobInfo); - return internalDirectUpload(blobInfo, opts, ByteBuffer.wrap(content)).asBlob(this); + return internalDirectUpload( + blobInfo, + opts, + ByteBuffer.wrap(content), + StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "WriteObject")) + .asBlob(this); } @Override 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); } @@ -180,11 +203,11 @@ public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... op InputStream inputStreamParam = firstNonNull(content, new ByteArrayInputStream(EMPTY_BYTE_ARRAY)); // retries are not safe when the input is an InputStream, so we can't retry. - BlobInfo info = - Conversions.json() - .blobInfo() - .decode(storageRpc.create(blobPb, inputStreamParam, optionsMap)); - return info.asBlob(this); + return run( + "WriteObject", + Retrying.neverRetry(), + () -> storageRpc.create(blobPb, inputStreamParam, optionsMap), + (x) -> Conversions.json().blobInfo().decode(x).asBlob(this)); } @Override @@ -203,43 +226,58 @@ 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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "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), + multiContext)); + 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(), multiContext), + 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 +357,7 @@ public Blob restore(BlobId blob, BlobRestoreOption... options) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsRestore(obj, optionsMap); return run( + "RestoreObject", algorithm, () -> storageRpc.restore(obj, optionsMap), (x) -> { @@ -347,7 +386,12 @@ private static class BucketPageFetcher implements NextPageFetcher { @Override public Page getNextPage() { - return listBuckets(serviceOptions, requestOptions, retrier); + return listBuckets( + serviceOptions, + requestOptions, + retrier, + StorageClientMetrics.createMultiStepContext( + serviceOptions.getStorageClientMetrics(), "ListBuckets")); } } @@ -374,7 +418,13 @@ private static class BlobPageFetcher implements NextPageFetcher { @Override public Page getNextPage() { - return listBlobs(bucket, serviceOptions, requestOptions, retrier); + return listBlobs( + bucket, + serviceOptions, + requestOptions, + retrier, + StorageClientMetrics.createMultiStepContext( + serviceOptions.getStorageClientMetrics(), "ListObjects")); } } @@ -399,26 +449,42 @@ private static class HmacKeyMetadataPageFetcher implements NextPageFetcher getNextPage() { - return listHmacKeys(serviceOptions, retryAlgorithmManager, options, retrier); + return listHmacKeys( + serviceOptions, + retryAlgorithmManager, + options, + retrier, + StorageClientMetrics.createMultiStepContext( + serviceOptions.getStorageClientMetrics(), "ListHMACKeys")); } } @Override public Page list(BucketListOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); - return listBuckets(getOptions(), optionsMap, retrier); + return listBuckets( + getOptions(), + optionsMap, + retrier, + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "ListBuckets")); } @Override public Page list(final String bucket, BlobListOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); - return listBlobs(bucket, getOptions(), optionsMap, retrier); + return listBlobs( + bucket, + getOptions(), + optionsMap, + retrier, + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "ListObjects")); } private static Page listBuckets( final HttpStorageOptions serviceOptions, final Map optionsMap, - Retrier retrier) { + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = serviceOptions.getRetryAlgorithmManager().getForBucketsList(optionsMap); return retrier.run( @@ -438,14 +504,16 @@ private static Page listBuckets( .asBucket(serviceOptions.getService())); return new PageImpl<>( new BucketPageFetcher(serviceOptions, cursor, optionsMap, retrier), cursor, buckets); - }); + }, + metricsContext); } private static Page listBlobs( final String bucket, final HttpStorageOptions serviceOptions, final Map optionsMap, - Retrier retrier) { + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = serviceOptions.getRetryAlgorithmManager().getForObjectsList(bucket, optionsMap); return retrier.run( @@ -466,7 +534,8 @@ private static Page listBlobs( new BlobPageFetcher(bucket, serviceOptions, cursor, optionsMap, retrier), cursor, blobs); - }); + }, + metricsContext); } @Override @@ -506,6 +575,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 +662,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 +684,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 +703,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 +739,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 +770,30 @@ public CopyWriter copy(final CopyRequest copyRequest) { targetOptions, copyRequest.getMegabytesCopiedPerChunk()); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsRewrite(rewriteRequest); - return run( - algorithm, - () -> storageRpc.openRewrite(rewriteRequest), - (r) -> new HttpCopyWriter(getOptions(), r, retrier)); + StorageMetricsContext multiContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "RewriteObject"); + try { + return retrier.run( + algorithm, + () -> storageRpc.openRewrite(rewriteRequest), + (r) -> { + HttpCopyWriter writer = new HttpCopyWriter(getOptions(), r, retrier, multiContext); + if (r.isDone && multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), null); + } + return writer; + }, + multiContext != null + ? multiContext + : StorageClientMetrics.createContext( + getOptions().getStorageClientMetrics(), "RewriteObject")); + } catch (Throwable t) { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } } @Override @@ -709,7 +809,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 +831,11 @@ 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 = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "ReadObject"); + return new BlobReadChannelV2( + storageObject, optionsMap, BlobReadChannelContext.from(this), metricsContext); } @Override @@ -747,6 +855,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( () -> { @@ -764,23 +873,36 @@ public StorageWriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) BlobInfo updated = opts.blobInfoMapper().apply(builder).build(); StorageObject encode = codecs.blobInfo().encode(updated); - // open the resumable session outside the write channel - // the exception behavior of open is different from #write(ByteBuffer) - 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()); - return new BlobWriteChannelV2(BlobReadChannelContext.from(this), jsonResumableWrite); + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "WriteObject"); + try { + // open the resumable session outside the write channel + // the exception behavior of open is different from #write(ByteBuffer) + Supplier uploadIdSupplier = + ResumableMedia.startUploadForBlobInfo( + getOptions(), + updated, + optionsMap, + retrier.withAlg( + retryAlgorithmManager.getForResumableUploadSessionCreate(optionsMap), + metricsContext)); + JsonResumableWrite jsonResumableWrite = + JsonResumableWrite.of( + encode, + optionsMap, + uploadIdSupplier.get(), + 0, + opts.getHasher(), + opts.getHasher().initialValue()); + return new BlobWriteChannelV2( + BlobReadChannelContext.from(this), jsonResumableWrite, metricsContext); + } catch (Throwable t) { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } } @Override @@ -792,12 +914,25 @@ public StorageWriteChannel writer(URL signedURL) { // open the resumable session outside the write channel // the exception behavior of open is different from #write(ByteBuffer) String signedUrlString = signedURL.toString(); + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "WriteObject"); Supplier uploadIdSupplier = ResumableMedia.startUploadForSignedUrl( - getOptions(), signedURL, retrier.withAlg(forResumableUploadSessionCreate)); - JsonResumableWrite jsonResumableWrite = - JsonResumableWrite.of(signedUrlString, uploadIdSupplier.get(), 0); - return new BlobWriteChannelV2(BlobReadChannelContext.from(this), jsonResumableWrite); + getOptions(), + signedURL, + retrier.withAlg(forResumableUploadSessionCreate, metricsContext)); + try { + JsonResumableWrite jsonResumableWrite = + JsonResumableWrite.of(signedUrlString, uploadIdSupplier.get(), 0); + return new BlobWriteChannelV2( + BlobReadChannelContext.from(this), jsonResumableWrite, metricsContext); + } catch (Throwable t) { + if (metricsContext != null) { + metricsContext.recordOperation(getOptions().getStorageClientMetrics(), t); + } + throw t; + } } @Override @@ -1279,7 +1414,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 +1432,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 +1450,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 +1468,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 +1486,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 +1506,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 +1570,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 +1585,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 +1601,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 +1614,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 +1625,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,13 +1642,22 @@ 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 public Page listHmacKeys(ListHmacKeysOption... options) { ImmutableMap optionsMap = Opts.unwrap(options).getRpcOptions(); - return listHmacKeys(getOptions(), retryAlgorithmManager, optionsMap, retrier); + return listHmacKeys( + getOptions(), + retryAlgorithmManager, + optionsMap, + retrier, + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), "ListHMACKeys")); } @Override @@ -1481,7 +1666,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 +1678,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 +1706,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) () -> { @@ -1527,7 +1720,8 @@ private static Page listHmacKeys( final HttpStorageOptions serviceOptions, final HttpRetryAlgorithmManager retryAlgorithmManager, final Map options, - Retrier retrier) { + Retrier retrier, + @Nullable StorageMetricsContext metricsContext) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForHmacKeyList(options); return retrier.run( algorithm, @@ -1543,7 +1737,8 @@ private static Page listHmacKeys( serviceOptions, retryAlgorithmManager, options, retrier), cursor, metadata); - }); + }, + metricsContext); } @Override @@ -1552,6 +1747,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 +1762,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 +1775,7 @@ public List testIamPermissions( ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForBucketsTestIamPermissions(bucket, permissions, optionsMap); return run( + "TestIamPermissions", algorithm, () -> storageRpc.testIamPermissions(bucket, permissions, optionsMap), (response) -> { @@ -1600,6 +1798,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 +1807,20 @@ 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, + StorageClientMetrics.createContext(getOptions().getStorageClientMetrics(), operationName)); } @Override @@ -1623,6 +1831,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 +1842,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 +1853,7 @@ public List listNotifications(final String bucket) { ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForNotificationList(bucket); List result = run( + "ListNotifications", algorithm, () -> storageRpc.listNotifications(bucket), (answer) -> @@ -1657,7 +1868,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 +1884,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 +1899,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 +1924,7 @@ public Blob moveBlob(MoveBlobRequest request) { ImmutableMap targetOptions = dstOpts.getRpcOptions(); return run( + "MoveObject", retryAlgorithmManager.getForObjectsMove(sourceOptions, targetOptions), () -> storageRpc.moveObject( @@ -1727,48 +1944,76 @@ public BlobInfo internalCreateFrom(Path path, BlobInfo info, Opts optionsMap = opts.getRpcOptions(); - BlobInfo.Builder builder = info.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 query = session.query(); - object = query.getObject(); + StorageMetricsContext multiContext = + StorageClientMetrics.createMultiStepContext( + getOptions().getStorageClientMetrics(), "WriteObject"); + Throwable error = null; + try { + final Map optionsMap = opts.getRpcOptions(); + BlobInfo.Builder builder = info.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), + multiContext)); + 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(), multiContext), + 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 query = session.query(); + object = query.getObject(); + } + return codecs.blobInfo().decode(object); + } catch (Throwable t) { + error = t; + throw t; + } finally { + if (multiContext != null) { + multiContext.recordOperation(getOptions().getStorageClientMetrics(), error); + } } - return codecs.blobInfo().decode(object); } @Override public BlobInfo internalDirectUpload(BlobInfo info, Opts opts, ByteBuffer buf) { + return internalDirectUpload(info, opts, buf, null); + } + + private BlobInfo internalDirectUpload( + BlobInfo info, + Opts opts, + ByteBuffer buf, + @Nullable StorageMetricsContext metricsContext) { BlobInfo.Builder builder = opts.blobInfoMapper().apply(info.toBuilder().clearMd5().clearCrc32c()); @Nullable Crc32cLengthKnown hash = opts.getHasher().hash(buf.duplicate()); @@ -1782,13 +2027,14 @@ public BlobInfo internalDirectUpload(BlobInfo info, Opts opts, ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsCreate(encoded, optionsMap); RewindableContent content = RewindableContent.of(buf); - return run( + return retrier.run( algorithm, () -> { content.rewindTo(0); return storageRpc.create(encoded, new RewindableContentInputStream(content), optionsMap); }, - Conversions.json().blobInfo()); + Conversions.json().blobInfo(), + metricsContext); } /** @@ -1801,7 +2047,7 @@ public Void internalObjectDelete(BlobId id, Opts opts) { ImmutableMap optionsMap = opts.getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsDelete(storageObject, optionsMap); - return run( + return retrier.run( algorithm, () -> { boolean deleted = storageRpc.delete(storageObject, optionsMap); @@ -1820,7 +2066,7 @@ public BlobInfo internalObjectGet(BlobId blobId, Opts opts) { ImmutableMap optionsMap = opts.getRpcOptions(); ResultRetryAlgorithm algorithm = retryAlgorithmManager.getForObjectsGet(storedObject, optionsMap); - return run( + return retrier.run( 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..c9a38219227b 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 @@ -16,24 +16,56 @@ 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_ERRORS; +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.cloud.storage.StorageErrorTypeMapper.STATUS_UNAVAILABLE; +import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.google.api.core.SettableApiFuture; +import com.google.api.gax.paging.Page; +import com.google.api.services.storage.model.StorageObject; +import com.google.cloud.ReadChannel; import com.google.cloud.RestorableState; import com.google.cloud.ServiceOptions; +import com.google.cloud.Tuple; +import com.google.cloud.WriteChannel; +import com.google.cloud.storage.BufferedWritableByteChannelSession.BufferedWritableByteChannel; import com.google.cloud.storage.Retrying.Retrier; +import com.google.cloud.storage.UnifiedOpts.Opts; import com.google.cloud.storage.spi.StorageRpcFactory; 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; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import io.opentelemetry.api.common.AttributeKey; +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.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.channels.ReadableByteChannel; +import java.util.Collection; 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 +242,350 @@ public void testSaveAndRestoreWithResult() { assertEquals(42L, restoredRewriter.getBlobSize()); verify(storageRpcMock).continueRewrite(RESPONSE_WITH_OBJECT); } + + @Test + public void testCopyChunkPassesMetricsContext() { + HttpStorageOptions metricsOptions = + options.toBuilder() + .setEnableOtelMetrics(true) + .setMeterProvider(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(); + assertThat(context).isNotNull(); + assertThat(context.getOperationName()).isEqualTo("RewriteObject"); + assertThat(context.shouldRecordOperationOnRetrierExit()).isFalse(); + } + + @Test + public void testMultiChunkCopyRecordsSingleOperationAndMultipleAttempts() { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + HttpStorageOptions metricsOptions = + options.toBuilder().setEnableOtelMetrics(true).setMeterProvider(meterProvider).build(); + when(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT)) + .thenReturn(RESPONSE_WITH_OBJECT, RESPONSE_WITH_OBJECT_DONE); + + copyWriter = + new HttpCopyWriter(metricsOptions, RESPONSE_WITH_OBJECT, metricsOptions.createRetrier()); + Blob copied = copyWriter.getResult(); + assertThat(copied.asBlobInfo()).isEqualTo(RESULT_INFO); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_OPERATIONS, + KEY_GCP_STORAGE_OPERATION, + "RewriteObject")) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_OK)) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, + KEY_GCP_STORAGE_OPERATION, + "RewriteObject")) + .isEqualTo(2L); + } + + @Test + public void testRestoredCopyWriterRecordsMultiStepMetrics() { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + HttpStorageOptions metricsOptions = + options.toBuilder().setEnableOtelMetrics(true).setMeterProvider(meterProvider).build(); + when(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT)) + .thenReturn(RESPONSE_WITH_OBJECT, RESPONSE_WITH_OBJECT_DONE); + + copyWriter = + new HttpCopyWriter(metricsOptions, RESPONSE_WITH_OBJECT, metricsOptions.createRetrier()); + RestorableState captured = copyWriter.capture(); + CopyWriter restored = captured.restore(); + Blob copied = restored.getResult(); + assertThat(copied.asBlobInfo()).isEqualTo(RESULT_INFO); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_OPERATIONS, + KEY_GCP_STORAGE_OPERATION, + "RewriteObject")) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, + KEY_GCP_STORAGE_OPERATION, + "RewriteObject")) + .isEqualTo(2L); + } + + @Test + public void testCopyChunkFailureRecordsErrorOperation() { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + HttpStorageOptions metricsOptions = + options.toBuilder().setEnableOtelMetrics(true).setMeterProvider(meterProvider).build(); + when(storageRpcMock.continueRewrite(RESPONSE_WITH_OBJECT)) + .thenThrow(new StorageException(503, "Service Unavailable")); + + copyWriter = + new HttpCopyWriter(metricsOptions, RESPONSE_WITH_OBJECT, metricsOptions.createRetrier()); + assertThrows(StorageException.class, () -> copyWriter.copyChunk()); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_UNAVAILABLE)) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_ERRORS, KEY_ERROR_TYPE, STATUS_UNAVAILABLE)) + .isEqualTo(1L); + } + + @Test + public void testReadChannelPreservesReadFailureOnClose() throws Exception { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + StorageClientMetrics clientMetrics = StorageClientMetrics.create(meterProvider, true, false); + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext(clientMetrics, "ReadObject"); + + @SuppressWarnings("unchecked") + ReadableByteChannelSession sessionMock = + Mockito.mock(ReadableByteChannelSession.class); + ReadableByteChannel channelMock = Mockito.mock(ReadableByteChannel.class); + when(sessionMock.open()).thenReturn(channelMock); + when(sessionMock.getResult()).thenReturn(SettableApiFuture.create()); + when(channelMock.isOpen()).thenReturn(true); + when(channelMock.read(any(ByteBuffer.class))).thenThrow(new StorageException(404, "Not Found")); + + BaseStorageReadChannel readChannel = + new BaseStorageReadChannel( + Conversions.json().blobInfo(), metricsContext, clientMetrics) { + @Override + public RestorableState capture() { + return null; + } + + @Override + protected LazyReadChannel newLazyReadChannel() { + return new LazyReadChannel<>(() -> sessionMock); + } + }; + + assertThrows(IOException.class, () -> readChannel.read(ByteBuffer.allocate(16))); + readChannel.close(); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_NOT_FOUND)) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_OK)) + .isEqualTo(0L); + } + + @Test + public void testReadChannel416TreatedAsNormalEofOnClose() throws Exception { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + StorageClientMetrics clientMetrics = StorageClientMetrics.create(meterProvider, true, false); + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext(clientMetrics, "ReadObject"); + + @SuppressWarnings("unchecked") + ReadableByteChannelSession sessionMock = + Mockito.mock(ReadableByteChannelSession.class); + ReadableByteChannel channelMock = Mockito.mock(ReadableByteChannel.class); + when(sessionMock.open()).thenReturn(channelMock); + when(sessionMock.getResult()).thenReturn(SettableApiFuture.create()); + when(channelMock.isOpen()).thenReturn(true); + when(channelMock.read(any(ByteBuffer.class))) + .thenThrow(new StorageException(416, "Range Not Satisfiable")); + + BaseStorageReadChannel readChannel = + new BaseStorageReadChannel( + Conversions.json().blobInfo(), metricsContext, clientMetrics) { + @Override + public RestorableState capture() { + return null; + } + + @Override + protected LazyReadChannel newLazyReadChannel() { + return new LazyReadChannel<>(() -> sessionMock); + } + }; + + assertThat(readChannel.read(ByteBuffer.allocate(16))).isEqualTo(-1); + readChannel.close(); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_OK)) + .isEqualTo(1L); + } + + @Test + public void testWriteChannelPreservesWriteFailureOnClose() throws Exception { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + StorageClientMetrics clientMetrics = StorageClientMetrics.create(meterProvider, true, false); + StorageMetricsContext metricsContext = + StorageClientMetrics.createMultiStepContext(clientMetrics, "WriteObject"); + + @SuppressWarnings("unchecked") + BufferedWritableByteChannelSession sessionMock = + Mockito.mock(BufferedWritableByteChannelSession.class); + BufferedWritableByteChannel channelMock = Mockito.mock(BufferedWritableByteChannel.class); + when(sessionMock.open()).thenReturn(channelMock); + when(sessionMock.getResult()).thenReturn(SettableApiFuture.create()); + when(channelMock.isOpen()).thenReturn(true); + when(channelMock.write(any(ByteBuffer.class))) + .thenThrow(new StorageException(503, "Service Unavailable")); + + BaseStorageWriteChannel writeChannel = + new BaseStorageWriteChannel( + Conversions.json().blobInfo(), metricsContext, clientMetrics) { + @Override + public RestorableState capture() { + return null; + } + + @Override + protected LazyWriteChannel newLazyWriteChannel() { + return new LazyWriteChannel<>(() -> sessionMock); + } + }; + + assertThrows(IOException.class, () -> writeChannel.write(ByteBuffer.wrap(new byte[] {1, 2}))); + writeChannel.close(); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_UNAVAILABLE)) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, METRIC_GCP_STORAGE_CLIENT_OPERATIONS, KEY_ERROR_TYPE, STATUS_OK)) + .isEqualTo(0L); + } + + @Test + public void testStorageImplPaginationRecordsSingleOperationAcrossPages() { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + HttpStorageOptions metricsOptions = + options.toBuilder().setEnableOtelMetrics(true).setMeterProvider(meterProvider).build(); + Storage storage = metricsOptions.getService(); + + StorageObject obj1 = Conversions.json().blobInfo().encode(BLOB_INFO); + StorageObject obj2 = Conversions.json().blobInfo().encode(RESULT_INFO); + when(storageRpcMock.list(eq(DESTINATION_BUCKET_NAME), anyMap())) + .thenReturn(Tuple.of("page-2", ImmutableList.of(obj1))) + .thenReturn(Tuple.of(null, ImmutableList.of(obj2))); + + Page firstPage = storage.list(DESTINATION_BUCKET_NAME); + Page secondPage = firstPage.getNextPage(); + assertThat(secondPage).isNotNull(); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_OPERATIONS, + KEY_GCP_STORAGE_OPERATION, + "ListObjects")) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_ATTEMPTS, + KEY_GCP_STORAGE_OPERATION, + "ListObjects")) + .isEqualTo(2L); + } + + @Test + public void testStorageImplPcuSubOperationsDoNotEmitTopLevelOperations() { + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + SdkMeterProvider meterProvider = + SdkMeterProvider.builder().registerMetricReader(metricReader).build(); + HttpStorageOptions metricsOptions = + options.toBuilder().setEnableOtelMetrics(true).setMeterProvider(meterProvider).build(); + StorageImpl storage = (StorageImpl) metricsOptions.getService(); + + StorageObject encodedResult = Conversions.json().blobInfo().encode(RESULT_INFO); + when(storageRpcMock.create(any(StorageObject.class), any(InputStream.class), anyMap())) + .thenReturn(encodedResult); + when(storageRpcMock.get(any(StorageObject.class), anyMap())).thenReturn(encodedResult); + when(storageRpcMock.delete(any(StorageObject.class), anyMap())).thenReturn(true); + + storage.create(BLOB_INFO, new byte[] {1, 2, 3}); + storage.internalDirectUpload(BLOB_INFO, Opts.empty(), ByteBuffer.wrap(new byte[] {4, 5})); + storage.internalObjectGet(BLOB_ID, Opts.empty()); + storage.internalObjectDelete(BLOB_ID, Opts.empty()); + + Collection metrics = metricReader.collectAllMetrics(); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_OPERATIONS, + KEY_GCP_STORAGE_OPERATION, + "WriteObject")) + .isEqualTo(1L); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_OPERATIONS, + KEY_GCP_STORAGE_OPERATION, + "GetObject")) + .isEqualTo(0L); + assertThat( + getCounterValueWithAttr( + metrics, + METRIC_GCP_STORAGE_CLIENT_OPERATIONS, + KEY_GCP_STORAGE_OPERATION, + "DeleteObject")) + .isEqualTo(0L); + } + + private static long getCounterValueWithAttr( + Collection metrics, + String metricName, + AttributeKey key, + String expectedValue) { + for (MetricData md : metrics) { + if (md.getName().equals(metricName) || md.getName().equals(metricName.replace(".", "/"))) { + return md.getLongSumData().getPoints().stream() + .filter(p -> expectedValue.equals(p.getAttributes().get(key))) + .mapToLong(LongPointData::getValue) + .sum(); + } + } + return 0L; + } }