feat(storage): wire operation metrics in HTTP veneer and storage channels

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageReadChannel.java

@@ -43,13 +43,24 @@ abstract class BaseStorageReadChannel<T> implements StorageReadChannel {
   private BufferHandle bufferHandle;
   private LazyReadChannel<?, T> lazyReadChannel;
   protected final ReentrantLock lock;
+  @Nullable private final StorageMetricsContext metricsContext;
+  @Nullable private final StorageClientMetrics storageClientMetrics;
 
   protected BaseStorageReadChannel(Decoder<T, BlobInfo> objectDecoder) {
+    this(objectDecoder, null, null);
+  }
+
+  protected BaseStorageReadChannel(
+      Decoder<T, BlobInfo> 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();
+      }
     }
   }
 
@@ -177,6 +200,10 @@ protected final T getResolvedObject() {
     }
   }
 
+  protected final @Nullable StorageMetricsContext getMetricsContext() {
+    return metricsContext;
+  }
+
   protected abstract LazyReadChannel<?, T> newLazyReadChannel();
 
   private void maybeResetChannel(boolean freeBuffer) throws IOException {

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BaseStorageWriteChannel.java

@@ -44,6 +44,8 @@ abstract class BaseStorageWriteChannel<T> implements StorageWriteChannel {
   private int chunkSize;
   private LazyWriteChannel<T> 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<T> implements StorageWriteChannel {
   protected boolean writeCalledAtLeastOnce;
 
   protected BaseStorageWriteChannel(Decoder<T, BlobInfo> objectDecoder) {
+    this(objectDecoder, null, null);
+  }
+
+  protected BaseStorageWriteChannel(
+      Decoder<T, BlobInfo> 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,16 +101,26 @@ 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));
       }
       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();
+      }
     }
   }
 
@@ -167,6 +188,10 @@ protected final void setOpen(boolean isOpen) {
     this.open = isOpen;
   }
 
+  protected final @Nullable StorageMetricsContext getMetricsContext() {
+    return metricsContext;
+  }
+
   protected abstract LazyWriteChannel<T> newLazyWriteChannel();
 
   private LazyWriteChannel<T> internalGetLazyChannel() {

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<StorageObject> {
 
@@ -40,7 +41,20 @@ final class BlobReadChannelV2 extends BaseStorageReadChannel<StorageObject> {
       StorageObject storageObject,
       Map<StorageRpc.Option, ?> opts,
       BlobReadChannelContext blobReadChannelContext) {
-    super(Conversions.json().blobInfo());
+    this(storageObject, opts, blobReadChannelContext, null);
+  }
+
+  BlobReadChannelV2(
+      StorageObject storageObject,
+      Map<StorageRpc.Option, ?> 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<?, StorageObject> 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.
@@ -186,6 +204,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

java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BlobWriteChannelV2.java

@@ -25,14 +25,27 @@
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Objects;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 final class BlobWriteChannelV2 extends BaseStorageWriteChannel<StorageObject> {
 
   private final BlobReadChannelContext blobChannelContext;
   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<StorageObject> newLazyWriteChannel() {
                 .withRetryConfig(
                     blobChannelContext
                         .getRetrier()
-                        .withAlg(blobChannelContext.getRetryAlgorithmManager().idempotent()))
+                        .withAlg(
+                            blobChannelContext.getRetryAlgorithmManager().idempotent(),
+                            getMetricsContext()))
                 .setHasher(start.getHasher())
                 .buffered(getBufferHandle())
                 .setStartAsync(ApiFutures.immediateFuture(start))

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,29 +29,50 @@
 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 {
 
   private final HttpStorageOptions serviceOptions;
   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
   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 +92,31 @@ 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
+                : serviceOptions.createMetricsContext("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;
     }
   }
 
@@ -230,10 +268,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);
     }