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,25 @@ 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;
+  @Nullable private Throwable lastOperationError;
 
   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 +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<?, 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,9 @@ 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;
+  @Nullable private Throwable lastOperationError;
 
   /**
    * This is tracked for compatibility with BlobWriteChannel, such that simply creating a writer
@@ -55,12 +58,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 +102,27 @@ 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 != 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<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.
@@ -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

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))
@@ -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);
       }