diff --git a/google-cloud-storage/clirr-ignored-differences.xml b/google-cloud-storage/clirr-ignored-differences.xml
index d81c37bb2..0b19230fd 100644
--- a/google-cloud-storage/clirr-ignored-differences.xml
+++ b/google-cloud-storage/clirr-ignored-differences.xml
@@ -96,4 +96,17 @@
     <method>boolean equals(java.lang.Object)</method>
   </difference>
 
+  <difference>
+    <differenceType>7013</differenceType>
+    <className>com/google/cloud/storage/StorageOptions$Builder</className>
+    <method>com.google.cloud.storage.StorageOptions$Builder setOpenTelemetry(io.opentelemetry.api.OpenTelemetry)</method>
+  </difference>
+
+  <difference>
+    <differenceType>7013</differenceType>
+    <className>com/google/cloud/storage/StorageOptions</className>
+    <method>io.opentelemetry.api.OpenTelemetry getOpenTelemetry()</method>
+  </difference>
+
+
 </differences>
diff --git a/google-cloud-storage/pom.xml b/google-cloud-storage/pom.xml
index 665a854dc..a246beb31 100644
--- a/google-cloud-storage/pom.xml
+++ b/google-cloud-storage/pom.xml
@@ -83,6 +83,10 @@
       <groupId>io.opencensus</groupId>
       <artifactId>opencensus-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-context</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-iam-v1</artifactId>
@@ -205,6 +209,11 @@
       <artifactId>grpc-googleapis</artifactId>
       <scope>runtime</scope>
     </dependency>
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk-trace</artifactId>
+      <scope>test</scope>
+    </dependency>
 
     <!--
     We're not using this directly, however there appears to be some resolution
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 436be1e70..279418c5f 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -81,9 +81,11 @@
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
 import io.grpc.protobuf.ProtoUtils;
+import io.opentelemetry.api.OpenTelemetry;
 import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.ObjectInputStream;
 import java.io.Serializable;
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -119,6 +121,7 @@ public final class GrpcStorageOptions extends StorageOptions
   private final boolean grpcClientMetricsManuallyEnabled;
   private final GrpcInterceptorProvider grpcInterceptorProvider;
   private final BlobWriteSessionConfig blobWriteSessionConfig;
+  private transient OpenTelemetry openTelemetry;
 
   private GrpcStorageOptions(Builder builder, GrpcStorageDefaults serviceDefaults) {
     super(builder, serviceDefaults);
@@ -135,6 +138,7 @@ private GrpcStorageOptions(Builder builder, GrpcStorageDefaults serviceDefaults)
     this.grpcClientMetricsManuallyEnabled = builder.grpcMetricsManuallyEnabled;
     this.grpcInterceptorProvider = builder.grpcInterceptorProvider;
     this.blobWriteSessionConfig = builder.blobWriteSessionConfig;
+    this.openTelemetry = builder.openTelemetry;
   }
 
   @Override
@@ -157,6 +161,11 @@ StorageSettings getStorageSettings() throws IOException {
     return resolveSettingsAndOpts().x();
   }
 
+  private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+    in.defaultReadObject();
+    this.openTelemetry = HttpStorageOptions.getDefaultInstance().getOpenTelemetry();
+  }
+
   /**
    * We have to perform several introspections and detections to cross-wire/support several features
    * that are either gapic primitives, ServiceOption primitives or GCS semantic requirements.
@@ -349,6 +358,13 @@ private Tuple<StorageSettings, Opts<UserProject>> resolveSettingsAndOpts() throw
     return Tuple.of(builder.build(), defaultOpts);
   }
 
+  /** @since 2.47.0 This new api is in preview and is subject to breaking changes. */
+  @BetaApi
+  @Override
+  public OpenTelemetry getOpenTelemetry() {
+    return openTelemetry;
+  }
+
   /** @since 2.14.0 */
   @Override
   public GrpcStorageOptions.Builder toBuilder() {
@@ -364,6 +380,7 @@ public int hashCode() {
         enableGrpcClientMetrics,
         grpcInterceptorProvider,
         blobWriteSessionConfig,
+        openTelemetry,
         baseHashCode());
   }
 
@@ -382,6 +399,7 @@ public boolean equals(Object o) {
         && Objects.equals(terminationAwaitDuration, that.terminationAwaitDuration)
         && Objects.equals(grpcInterceptorProvider, that.grpcInterceptorProvider)
         && Objects.equals(blobWriteSessionConfig, that.blobWriteSessionConfig)
+        && Objects.equals(openTelemetry, that.openTelemetry)
         && this.baseEquals(that);
   }
 
@@ -423,6 +441,7 @@ public static final class Builder extends StorageOptions.Builder {
         GrpcStorageDefaults.INSTANCE.grpcInterceptorProvider();
     private BlobWriteSessionConfig blobWriteSessionConfig =
         GrpcStorageDefaults.INSTANCE.getDefaultStorageWriterConfig();
+    private OpenTelemetry openTelemetry = GrpcStorageDefaults.INSTANCE.getDefaultOpenTelemetry();
 
     private boolean grpcMetricsManuallyEnabled = false;
 
@@ -437,6 +456,7 @@ public static final class Builder extends StorageOptions.Builder {
       this.enableGrpcClientMetrics = gso.enableGrpcClientMetrics;
       this.grpcInterceptorProvider = gso.grpcInterceptorProvider;
       this.blobWriteSessionConfig = gso.blobWriteSessionConfig;
+      this.openTelemetry = gso.openTelemetry;
     }
 
     /**
@@ -619,6 +639,19 @@ public GrpcStorageOptions.Builder setBlobWriteSessionConfig(
       return this;
     }
 
+    /**
+     * Enable OpenTelemetry Tracing and provide an instance for the client to use.
+     *
+     * @param openTelemetry User defined instance of OpenTelemetry to be used by the library
+     * @since 2.47.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public GrpcStorageOptions.Builder setOpenTelemetry(OpenTelemetry openTelemetry) {
+      requireNonNull(openTelemetry, "openTelemetry must be non null");
+      this.openTelemetry = openTelemetry;
+      return this;
+    }
+
     /** @since 2.14.0 */
     @Override
     public GrpcStorageOptions build() {
@@ -695,6 +728,12 @@ public GrpcInterceptorProvider grpcInterceptorProvider() {
     public BlobWriteSessionConfig getDefaultStorageWriterConfig() {
       return BlobWriteSessionConfigs.getDefault();
     }
+
+    /** @since 2.47.0 This new api is in preview and is subject to breaking changes. */
+    @BetaApi
+    public OpenTelemetry getDefaultOpenTelemetry() {
+      return OpenTelemetry.noop();
+    }
   }
 
   /**
@@ -751,22 +790,27 @@ public Storage create(StorageOptions options) {
                 new InternalZeroCopyGrpcStorageStub(
                     stubSettings, clientContext, grpcStorageCallableFactory);
             StorageClient client = new InternalStorageClient(stub);
-            return new GrpcStorageImpl(
-                grpcStorageOptions,
-                client,
-                stub.getObjectMediaResponseMarshaller,
-                grpcStorageOptions.blobWriteSessionConfig.createFactory(Clock.systemUTC()),
-                defaultOpts);
+            GrpcStorageImpl grpcStorage =
+                new GrpcStorageImpl(
+                    grpcStorageOptions,
+                    client,
+                    stub.getObjectMediaResponseMarshaller,
+                    grpcStorageOptions.blobWriteSessionConfig.createFactory(Clock.systemUTC()),
+                    defaultOpts);
+            return OtelStorageDecorator.decorate(
+                grpcStorage, options.getOpenTelemetry(), Transport.GRPC);
           } else {
             StorageClient client = StorageClient.create(storageSettings);
-            return new GrpcStorageImpl(
-                grpcStorageOptions,
-                client,
-                ResponseContentLifecycleManager.noop(),
-                grpcStorageOptions.blobWriteSessionConfig.createFactory(Clock.systemUTC()),
-                defaultOpts);
+            GrpcStorageImpl grpcStorage =
+                new GrpcStorageImpl(
+                    grpcStorageOptions,
+                    client,
+                    ResponseContentLifecycleManager.noop(),
+                    grpcStorageOptions.blobWriteSessionConfig.createFactory(Clock.systemUTC()),
+                    defaultOpts);
+            return OtelStorageDecorator.decorate(
+                grpcStorage, options.getOpenTelemetry(), Transport.GRPC);
           }
-
         } catch (IOException e) {
           throw new IllegalStateException(
               "Unable to instantiate gRPC com.google.cloud.storage.Storage client.", e);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
index a14bad424..e440765fe 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/HttpStorageOptions.java
@@ -38,6 +38,7 @@
 import com.google.cloud.storage.spi.v1.StorageRpc;
 import com.google.common.base.MoreObjects;
 import com.google.common.collect.ImmutableSet;
+import io.opentelemetry.api.OpenTelemetry;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.Serializable;
@@ -59,6 +60,8 @@ public class HttpStorageOptions extends StorageOptions {
   private transient RetryDependenciesAdapter retryDepsAdapter;
   private final BlobWriteSessionConfig blobWriteSessionConfig;
 
+  private transient OpenTelemetry openTelemetry;
+
   private HttpStorageOptions(Builder builder, StorageDefaults serviceDefaults) {
     super(builder, serviceDefaults);
     this.retryAlgorithmManager =
@@ -67,6 +70,7 @@ private HttpStorageOptions(Builder builder, StorageDefaults serviceDefaults) {
                 builder.storageRetryStrategy, defaults().getStorageRetryStrategy()));
     retryDepsAdapter = new RetryDependenciesAdapter();
     blobWriteSessionConfig = builder.blobWriteSessionConfig;
+    openTelemetry = builder.openTelemetry;
   }
 
   @Override
@@ -84,6 +88,13 @@ StorageRpc getStorageRpcV1() {
     return (StorageRpc) getRpc();
   }
 
+  /** @since 2.47.0 This new api is in preview and is subject to breaking changes. */
+  @BetaApi
+  @Override
+  public OpenTelemetry getOpenTelemetry() {
+    return openTelemetry;
+  }
+
   @Override
   public HttpStorageOptions.Builder toBuilder() {
     return new HttpStorageOptions.Builder(this);
@@ -91,7 +102,8 @@ public HttpStorageOptions.Builder toBuilder() {
 
   @Override
   public int hashCode() {
-    return Objects.hash(retryAlgorithmManager, blobWriteSessionConfig, baseHashCode());
+    return Objects.hash(
+        retryAlgorithmManager, blobWriteSessionConfig, openTelemetry, baseHashCode());
   }
 
   @Override
@@ -105,12 +117,14 @@ public boolean equals(Object o) {
     HttpStorageOptions that = (HttpStorageOptions) o;
     return Objects.equals(retryAlgorithmManager, that.retryAlgorithmManager)
         && Objects.equals(blobWriteSessionConfig, that.blobWriteSessionConfig)
+        && Objects.equals(openTelemetry, that.openTelemetry)
         && this.baseEquals(that);
   }
 
   private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
     in.defaultReadObject();
     this.retryDepsAdapter = new RetryDependenciesAdapter();
+    this.openTelemetry = HttpStorageOptions.getDefaultInstance().getOpenTelemetry();
   }
 
   public static HttpStorageOptions.Builder newBuilder() {
@@ -135,6 +149,7 @@ public static class Builder extends StorageOptions.Builder {
     private StorageRetryStrategy storageRetryStrategy;
     private BlobWriteSessionConfig blobWriteSessionConfig =
         HttpStorageDefaults.INSTANCE.getDefaultStorageWriterConfig();
+    private OpenTelemetry openTelemetry = HttpStorageDefaults.INSTANCE.getDefaultOpenTelemetry();
 
     Builder() {}
 
@@ -143,6 +158,7 @@ public static class Builder extends StorageOptions.Builder {
       HttpStorageOptions hso = (HttpStorageOptions) options;
       this.storageRetryStrategy = hso.retryAlgorithmManager.retryStrategy;
       this.blobWriteSessionConfig = hso.blobWriteSessionConfig;
+      this.openTelemetry = hso.getOpenTelemetry();
     }
 
     @Override
@@ -268,6 +284,19 @@ public HttpStorageOptions build() {
       }
       return options;
     }
+
+    /**
+     * Enable OpenTelemetry Tracing and provide an instance for the client to use.
+     *
+     * @param openTelemetry User defined instance of OpenTelemetry to be used by the library
+     * @since 2.47.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public HttpStorageOptions.Builder setOpenTelemetry(OpenTelemetry openTelemetry) {
+      requireNonNull(openTelemetry, "openTelemetry must be non null");
+      this.openTelemetry = openTelemetry;
+      return this;
+    }
   }
 
   public static final class HttpStorageDefaults extends StorageDefaults {
@@ -301,6 +330,12 @@ public StorageRetryStrategy getStorageRetryStrategy() {
     public BlobWriteSessionConfig getDefaultStorageWriterConfig() {
       return BlobWriteSessionConfigs.getDefault();
     }
+
+    /** @since 2.47.0 This new api is in preview and is subject to breaking changes. */
+    @BetaApi
+    public OpenTelemetry getDefaultOpenTelemetry() {
+      return OpenTelemetry.noop();
+    }
   }
 
   /**
@@ -341,8 +376,12 @@ public Storage create(StorageOptions options) {
         HttpStorageOptions httpStorageOptions = (HttpStorageOptions) options;
         Clock clock = Clock.systemUTC();
         try {
-          return new StorageImpl(
-              httpStorageOptions, httpStorageOptions.blobWriteSessionConfig.createFactory(clock));
+          StorageImpl storage =
+              new StorageImpl(
+                  httpStorageOptions,
+                  httpStorageOptions.blobWriteSessionConfig.createFactory(clock));
+          return OtelStorageDecorator.decorate(
+              storage, httpStorageOptions.getOpenTelemetry(), Transport.HTTP);
         } catch (IOException e) {
           throw new IllegalStateException(
               "Unable to instantiate HTTP com.google.cloud.storage.Storage client.", e);
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java
new file mode 100644
index 000000000..a1813a98c
--- /dev/null
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java
@@ -0,0 +1,1733 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static java.util.Objects.requireNonNull;
+
+import com.google.api.core.ApiFuture;
+import com.google.api.core.BetaApi;
+import com.google.api.gax.paging.Page;
+import com.google.cloud.Policy;
+import com.google.cloud.ReadChannel;
+import com.google.cloud.RestorableState;
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.Acl.Entity;
+import com.google.cloud.storage.HmacKey.HmacKeyMetadata;
+import com.google.cloud.storage.HmacKey.HmacKeyState;
+import com.google.cloud.storage.PostPolicyV4.PostConditionsV4;
+import com.google.cloud.storage.PostPolicyV4.PostFieldsV4;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URL;
+import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.TimeUnit;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+@SuppressWarnings("DuplicatedCode")
+final class OtelStorageDecorator implements Storage {
+
+  /** Becomes the {@code otel.scope.name} attribute in a span */
+  private static final String OTEL_SCOPE_NAME = "cloud.google.com/java/storage";
+
+  private final Storage delegate;
+  private final OpenTelemetry otel;
+  private final Attributes baseAttributes;
+  private final Tracer tracer;
+
+  private OtelStorageDecorator(Storage delegate, OpenTelemetry otel, Attributes baseAttributes) {
+    this.delegate = delegate;
+    this.otel = otel;
+    this.baseAttributes = baseAttributes;
+    this.tracer =
+        TracerDecorator.decorate(null, otel, baseAttributes, Storage.class.getName() + "/");
+  }
+
+  @Override
+  public Bucket create(BucketInfo bucketInfo, BucketTargetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("create")
+            .setAttribute("gsutil.uri", fmtBucket(bucketInfo.getName()))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.create(bucketInfo, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob create(BlobInfo blobInfo, BlobTargetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("create")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.create(blobInfo, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob create(BlobInfo blobInfo, byte[] content, BlobTargetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("create")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.create(blobInfo, content, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob create(
+      BlobInfo blobInfo, byte[] content, int offset, int length, BlobTargetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("create")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.create(blobInfo, content, offset, length, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  @Deprecated
+  public Blob create(BlobInfo blobInfo, InputStream content, BlobWriteOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("create")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.create(blobInfo, content, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob createFrom(BlobInfo blobInfo, Path path, BlobWriteOption... options)
+      throws IOException {
+    Span span =
+        tracer
+            .spanBuilder("createFrom")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createFrom(blobInfo, path, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob createFrom(BlobInfo blobInfo, Path path, int bufferSize, BlobWriteOption... options)
+      throws IOException {
+    Span span =
+        tracer
+            .spanBuilder("createFrom")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createFrom(blobInfo, path, bufferSize, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob createFrom(BlobInfo blobInfo, InputStream content, BlobWriteOption... options)
+      throws IOException {
+    Span span =
+        tracer
+            .spanBuilder("createFrom")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createFrom(blobInfo, content, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob createFrom(
+      BlobInfo blobInfo, InputStream content, int bufferSize, BlobWriteOption... options)
+      throws IOException {
+    Span span =
+        tracer
+            .spanBuilder("createFrom")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createFrom(blobInfo, content, bufferSize, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Bucket get(String bucket, BucketGetOption... options) {
+    Span span = tracer.spanBuilder("get").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.get(bucket, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Bucket lockRetentionPolicy(BucketInfo bucket, BucketTargetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("lockRetentionPolicy")
+            .setAttribute("gsutil.uri", fmtBucket(bucket.getName()))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.lockRetentionPolicy(bucket, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob get(String bucket, String blob, BlobGetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("get")
+            .setAttribute("gsutil.uri", String.format("gs://%s/%s", bucket, blob))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.get(bucket, blob, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob get(BlobId blob, BlobGetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("get")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.get(blob, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob get(BlobId blob) {
+    Span span =
+        tracer
+            .spanBuilder("get")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.get(blob);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob restore(BlobId blob, BlobRestoreOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("restore")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.restore(blob, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Page<Bucket> list(BucketListOption... options) {
+    Span span = tracer.spanBuilder("list").setAttribute("gsutil.uri", "gs://").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.list(options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Page<Blob> list(String bucket, BlobListOption... options) {
+    Span span =
+        tracer.spanBuilder("list").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.list(bucket, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Bucket update(BucketInfo bucketInfo, BucketTargetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("update")
+            .setAttribute("gsutil.uri", fmtBucket(bucketInfo.getName()))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.update(bucketInfo, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob update(BlobInfo blobInfo, BlobTargetOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("update")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.update(blobInfo, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob update(BlobInfo blobInfo) {
+    Span span =
+        tracer
+            .spanBuilder("update")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.update(blobInfo);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean delete(String bucket, BucketSourceOption... options) {
+    Span span =
+        tracer.spanBuilder("delete").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.delete(bucket, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean delete(String bucket, String blob, BlobSourceOption... options) {
+    Span span =
+        tracer.spanBuilder("delete").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.delete(bucket, blob, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean delete(BlobId blob, BlobSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("delete")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.delete(blob, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean delete(BlobId blob) {
+    Span span =
+        tracer
+            .spanBuilder("delete")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.delete(blob);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Blob compose(ComposeRequest composeRequest) {
+    Span span =
+        tracer
+            .spanBuilder("compose")
+            .setAttribute("gsutil.uri", composeRequest.getTarget().getBlobId().toGsUtilUri())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.compose(composeRequest);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public CopyWriter copy(CopyRequest copyRequest) {
+    Span span =
+        tracer
+            .spanBuilder("copy")
+            .setAttribute("gsutil.uri", copyRequest.getTarget().getBlobId().toGsUtilUri())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      CopyWriter copyWriter = delegate.copy(copyRequest);
+      return new OtelDecoratedCopyWriter(copyWriter, span);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      span.end();
+      throw t;
+    }
+  }
+
+  @Override
+  public byte[] readAllBytes(String bucket, String blob, BlobSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("readAllBytes")
+            .setAttribute("gsutil.uri", BlobId.of(bucket, blob).toGsUtilUri())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.readAllBytes(bucket, blob, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public byte[] readAllBytes(BlobId blob, BlobSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("readAllBytes")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.readAllBytes(blob, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public StorageBatch batch() {
+    return delegate.batch();
+  }
+
+  @Override
+  public ReadChannel reader(String bucket, String blob, BlobSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("reader")
+            .setAttribute("gsutil.uri", BlobId.of(bucket, blob).toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      ReadChannel reader = delegate.reader(bucket, blob, options);
+      return new OtelDecoratedReadChannel(reader, span);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      span.end();
+      throw t;
+    }
+  }
+
+  @Override
+  public ReadChannel reader(BlobId blob, BlobSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("reader")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      ReadChannel reader = delegate.reader(blob, options);
+      return new OtelDecoratedReadChannel(reader, span);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      span.end();
+      throw t;
+    }
+  }
+
+  @Override
+  public void downloadTo(BlobId blob, Path path, BlobSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("downloadTo")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      delegate.downloadTo(blob, path, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public void downloadTo(BlobId blob, OutputStream outputStream, BlobSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("downloadTo")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      delegate.downloadTo(blob, outputStream, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public WriteChannel writer(BlobInfo blobInfo, BlobWriteOption... options) {
+    Span sessionSpan =
+        tracer
+            .spanBuilder("writer")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = sessionSpan.makeCurrent()) {
+      WriteChannel writer = delegate.writer(blobInfo, options);
+      return new OtelDecoratedWriteChannel(writer, sessionSpan);
+    } catch (Throwable t) {
+      sessionSpan.recordException(t);
+      sessionSpan.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      sessionSpan.end();
+      throw t;
+    }
+  }
+
+  @Override
+  public WriteChannel writer(URL signedURL) {
+    Span sessionSpan = tracer.spanBuilder("writer").startSpan();
+    try (Scope ignore = sessionSpan.makeCurrent()) {
+      WriteChannel writer = delegate.writer(signedURL);
+      return new OtelDecoratedWriteChannel(writer, sessionSpan);
+    } catch (Throwable t) {
+      sessionSpan.recordException(t);
+      sessionSpan.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      sessionSpan.end();
+      throw t;
+    }
+  }
+
+  @Override
+  public URL signUrl(BlobInfo blobInfo, long duration, TimeUnit unit, SignUrlOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("signUrl")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.signUrl(blobInfo, duration, unit, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public PostPolicyV4 generateSignedPostPolicyV4(
+      BlobInfo blobInfo,
+      long duration,
+      TimeUnit unit,
+      PostFieldsV4 fields,
+      PostConditionsV4 conditions,
+      PostPolicyV4Option... options) {
+    Span span =
+        tracer
+            .spanBuilder("generateSignedPostPolicyV4")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.generateSignedPostPolicyV4(
+          blobInfo, duration, unit, fields, conditions, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public PostPolicyV4 generateSignedPostPolicyV4(
+      BlobInfo blobInfo,
+      long duration,
+      TimeUnit unit,
+      PostFieldsV4 fields,
+      PostPolicyV4Option... options) {
+    Span span =
+        tracer
+            .spanBuilder("generateSignedPostPolicyV4")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.generateSignedPostPolicyV4(blobInfo, duration, unit, fields, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public PostPolicyV4 generateSignedPostPolicyV4(
+      BlobInfo blobInfo,
+      long duration,
+      TimeUnit unit,
+      PostConditionsV4 conditions,
+      PostPolicyV4Option... options) {
+    Span span =
+        tracer
+            .spanBuilder("generateSignedPostPolicyV4")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.generateSignedPostPolicyV4(blobInfo, duration, unit, conditions, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public PostPolicyV4 generateSignedPostPolicyV4(
+      BlobInfo blobInfo, long duration, TimeUnit unit, PostPolicyV4Option... options) {
+    Span span =
+        tracer
+            .spanBuilder("generateSignedPostPolicyV4")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.generateSignedPostPolicyV4(blobInfo, duration, unit, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Blob> get(BlobId... blobIds) {
+    Span span = tracer.spanBuilder("get").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.get(blobIds);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Blob> get(Iterable<BlobId> blobIds) {
+    Span span = tracer.spanBuilder("get").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.get(blobIds);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Blob> update(BlobInfo... blobInfos) {
+    Span span = tracer.spanBuilder("update").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.update(blobInfos);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Blob> update(Iterable<BlobInfo> blobInfos) {
+    Span span = tracer.spanBuilder("update").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.update(blobInfos);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Boolean> delete(BlobId... blobIds) {
+    Span span = tracer.spanBuilder("delete").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.delete(blobIds);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Boolean> delete(Iterable<BlobId> blobIds) {
+    Span span = tracer.spanBuilder("delete").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.delete(blobIds);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl getAcl(String bucket, Entity entity, BucketSourceOption... options) {
+    Span span =
+        tracer.spanBuilder("getAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getAcl(bucket, entity, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl getAcl(String bucket, Entity entity) {
+    Span span =
+        tracer.spanBuilder("getAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getAcl(bucket, entity);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean deleteAcl(String bucket, Entity entity, BucketSourceOption... options) {
+    Span span =
+        tracer.spanBuilder("deleteAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.deleteAcl(bucket, entity, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean deleteAcl(String bucket, Entity entity) {
+    Span span =
+        tracer.spanBuilder("deleteAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.deleteAcl(bucket, entity);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl createAcl(String bucket, Acl acl, BucketSourceOption... options) {
+    Span span =
+        tracer.spanBuilder("createAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createAcl(bucket, acl, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl createAcl(String bucket, Acl acl) {
+    Span span =
+        tracer.spanBuilder("createAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createAcl(bucket, acl);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl updateAcl(String bucket, Acl acl, BucketSourceOption... options) {
+    Span span =
+        tracer.spanBuilder("updateAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.updateAcl(bucket, acl, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl updateAcl(String bucket, Acl acl) {
+    Span span =
+        tracer.spanBuilder("updateAcl").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.updateAcl(bucket, acl);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Acl> listAcls(String bucket, BucketSourceOption... options) {
+    Span span =
+        tracer.spanBuilder("listAcls").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.listAcls(bucket, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Acl> listAcls(String bucket) {
+    Span span =
+        tracer.spanBuilder("listAcls").setAttribute("gsutil.uri", fmtBucket(bucket)).startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.listAcls(bucket);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl getDefaultAcl(String bucket, Entity entity) {
+    Span span =
+        tracer
+            .spanBuilder("getDefaultAcl")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getDefaultAcl(bucket, entity);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean deleteDefaultAcl(String bucket, Entity entity) {
+    Span span =
+        tracer
+            .spanBuilder("deleteDefaultAcl")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.deleteDefaultAcl(bucket, entity);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl createDefaultAcl(String bucket, Acl acl) {
+    Span span =
+        tracer
+            .spanBuilder("createDefaultAcl")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createDefaultAcl(bucket, acl);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl updateDefaultAcl(String bucket, Acl acl) {
+    Span span =
+        tracer
+            .spanBuilder("updateDefaultAcl")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.updateDefaultAcl(bucket, acl);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Acl> listDefaultAcls(String bucket) {
+    Span span =
+        tracer
+            .spanBuilder("listDefaultAcls")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.listDefaultAcls(bucket);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl getAcl(BlobId blob, Entity entity) {
+    Span span =
+        tracer
+            .spanBuilder("getAcl")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getAcl(blob, entity);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean deleteAcl(BlobId blob, Entity entity) {
+    Span span =
+        tracer
+            .spanBuilder("deleteAcl")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.deleteAcl(blob, entity);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl createAcl(BlobId blob, Acl acl) {
+    Span span =
+        tracer
+            .spanBuilder("createAcl")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createAcl(blob, acl);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Acl updateAcl(BlobId blob, Acl acl) {
+    Span span =
+        tracer
+            .spanBuilder("updateAcl")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.updateAcl(blob, acl);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Acl> listAcls(BlobId blob) {
+    Span span =
+        tracer
+            .spanBuilder("listAcls")
+            .setAttribute("gsutil.uri", blob.toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.listAcls(blob);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public HmacKey createHmacKey(ServiceAccount serviceAccount, CreateHmacKeyOption... options) {
+    Span span = tracer.spanBuilder("createHmacKey").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createHmacKey(serviceAccount, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Page<HmacKeyMetadata> listHmacKeys(ListHmacKeysOption... options) {
+    Span span = tracer.spanBuilder("listHmacKeys").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.listHmacKeys(options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public HmacKeyMetadata getHmacKey(String accessId, GetHmacKeyOption... options) {
+    Span span = tracer.spanBuilder("getHmacKey").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getHmacKey(accessId, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public void deleteHmacKey(HmacKeyMetadata hmacKeyMetadata, DeleteHmacKeyOption... options) {
+    Span span = tracer.spanBuilder("deleteHmacKey").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      delegate.deleteHmacKey(hmacKeyMetadata, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public HmacKeyMetadata updateHmacKeyState(
+      HmacKeyMetadata hmacKeyMetadata, HmacKeyState state, UpdateHmacKeyOption... options) {
+    Span span = tracer.spanBuilder("updateHmacKeyState").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.updateHmacKeyState(hmacKeyMetadata, state, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Policy getIamPolicy(String bucket, BucketSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("getIamPolicy")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getIamPolicy(bucket, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Policy setIamPolicy(String bucket, Policy policy, BucketSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("setIamPolicy")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.setIamPolicy(bucket, policy, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Boolean> testIamPermissions(
+      String bucket, List<String> permissions, BucketSourceOption... options) {
+    Span span =
+        tracer
+            .spanBuilder("testIamPermissions")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.testIamPermissions(bucket, permissions, options);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public ServiceAccount getServiceAccount(String projectId) {
+    Span span = tracer.spanBuilder("getServiceAccount").startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getServiceAccount(projectId);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Notification createNotification(String bucket, NotificationInfo notificationInfo) {
+    Span span =
+        tracer
+            .spanBuilder("createNotification")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.createNotification(bucket, notificationInfo);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public Notification getNotification(String bucket, String notificationId) {
+    Span span =
+        tracer
+            .spanBuilder("getNotification")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.getNotification(bucket, notificationId);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public List<Notification> listNotifications(String bucket) {
+    Span span =
+        tracer
+            .spanBuilder("listNotifications")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.listNotifications(bucket);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public boolean deleteNotification(String bucket, String notificationId) {
+    Span span =
+        tracer
+            .spanBuilder("deleteNotification")
+            .setAttribute("gsutil.uri", fmtBucket(bucket))
+            .startSpan();
+    try (Scope ignore = span.makeCurrent()) {
+      return delegate.deleteNotification(bucket, notificationId);
+    } catch (Throwable t) {
+      span.recordException(t);
+      span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      span.end();
+    }
+  }
+
+  @Override
+  public void close() throws Exception {
+    delegate.close();
+  }
+
+  @Override
+  @BetaApi
+  public BlobWriteSession blobWriteSession(BlobInfo blobInfo, BlobWriteOption... options) {
+    Span sessionSpan =
+        tracer
+            .spanBuilder("blobWriteSession")
+            .setAttribute("gsutil.uri", blobInfo.getBlobId().toGsUtilUriWithGeneration())
+            .startSpan();
+    try (Scope ignore = sessionSpan.makeCurrent()) {
+      BlobWriteSession session = delegate.blobWriteSession(blobInfo, options);
+      return new OtelDecoratedBlobWriteSession(session);
+    } catch (Throwable t) {
+      sessionSpan.recordException(t);
+      sessionSpan.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+      throw t;
+    } finally {
+      sessionSpan.end();
+    }
+  }
+
+  @Override
+  public StorageOptions getOptions() {
+    return delegate.getOptions();
+  }
+
+  static Storage decorate(Storage delegate, OpenTelemetry otel, Transport transport) {
+    requireNonNull(delegate, "delegate must be non null");
+    requireNonNull(otel, "otel must be non null");
+    if (otel == OpenTelemetry.noop()) {
+      return delegate;
+    }
+    Attributes baseAttributes =
+        Attributes.builder()
+            .put("gcp.client.service", "Storage")
+            .put("gcp.client.version", StorageOptions.getDefaultInstance().getLibraryVersion())
+            .put("gcp.client.repo", "googleapis/java-storage")
+            .put("gcp.client.artifact", "com.google.cloud:google-cloud-storage")
+            .put("rpc.system", transport.toString().toLowerCase(Locale.ROOT))
+            .put("service.name", "storage.googleapis.com")
+            .build();
+    return new OtelStorageDecorator(delegate, otel, baseAttributes);
+  }
+
+  private static @NonNull String fmtBucket(String bucket) {
+    return String.format("gs://%s/", bucket);
+  }
+
+  private static final class TracerDecorator implements Tracer {
+    @Nullable private final Context parentContextOverride;
+    private final Tracer delegate;
+    private final Attributes baseAttributes;
+    private final String spanNamePrefix;
+
+    private TracerDecorator(
+        @Nullable Context parentContextOverride,
+        Tracer delegate,
+        Attributes baseAttributes,
+        String spanNamePrefix) {
+      this.parentContextOverride = parentContextOverride;
+      this.delegate = delegate;
+      this.baseAttributes = baseAttributes;
+      this.spanNamePrefix = spanNamePrefix;
+    }
+
+    private static TracerDecorator decorate(
+        @Nullable Context parentContextOverride,
+        OpenTelemetry otel,
+        Attributes baseAttributes,
+        String spanNamePrefix) {
+      requireNonNull(otel, "otel must be non null");
+      requireNonNull(baseAttributes, "baseAttributes must be non null");
+      requireNonNull(spanNamePrefix, "spanNamePrefix must be non null");
+      Tracer tracer =
+          otel.getTracer(OTEL_SCOPE_NAME, StorageOptions.getDefaultInstance().getLibraryVersion());
+      return new TracerDecorator(parentContextOverride, tracer, baseAttributes, spanNamePrefix);
+    }
+
+    @Override
+    public SpanBuilder spanBuilder(String spanName) {
+      SpanBuilder spanBuilder =
+          delegate.spanBuilder(spanNamePrefix + spanName).setAllAttributes(baseAttributes);
+      if (parentContextOverride != null) {
+        spanBuilder.setParent(parentContextOverride);
+      }
+      return spanBuilder;
+    }
+  }
+
+  private static final class OtelDecoratedReadChannel implements ReadChannel {
+
+    private final ReadChannel reader;
+    private final Span span;
+
+    private OtelDecoratedReadChannel(ReadChannel reader, Span span) {
+      this.reader = reader;
+      this.span = span;
+    }
+
+    @Override
+    public void seek(long position) throws IOException {
+      reader.seek(position);
+    }
+
+    @Override
+    public void setChunkSize(int chunkSize) {
+      reader.setChunkSize(chunkSize);
+    }
+
+    @Override
+    public RestorableState<ReadChannel> capture() {
+      return reader.capture();
+    }
+
+    @Override
+    public ReadChannel limit(long limit) {
+      return reader.limit(limit);
+    }
+
+    @Override
+    public long limit() {
+      return reader.limit();
+    }
+
+    @Override
+    public int read(ByteBuffer dst) throws IOException {
+      return reader.read(dst);
+    }
+
+    @Override
+    public boolean isOpen() {
+      return reader.isOpen();
+    }
+
+    @Override
+    public void close() {
+      try {
+        reader.close();
+      } finally {
+        span.end();
+      }
+    }
+  }
+
+  private final class OtelDecoratedBlobWriteSession implements BlobWriteSession {
+
+    private final BlobWriteSession delegate;
+    private final Tracer tracer;
+
+    public OtelDecoratedBlobWriteSession(BlobWriteSession delegate) {
+      this.delegate = delegate;
+      this.tracer =
+          TracerDecorator.decorate(
+              Context.current(),
+              otel,
+              OtelStorageDecorator.this.baseAttributes,
+              BlobWriteSession.class.getName() + "/");
+    }
+
+    @Override
+    public WritableByteChannel open() throws IOException {
+      Span openSpan = tracer.spanBuilder("open").startSpan();
+      try (Scope ignore = openSpan.makeCurrent()) {
+        WritableByteChannel delegate = this.delegate.open();
+        return new OtelDecoratingWritableByteChannel(delegate, openSpan);
+      } catch (Throwable t) {
+        openSpan.recordException(t);
+        openSpan.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+        throw t;
+      }
+    }
+
+    @Override
+    public ApiFuture<BlobInfo> getResult() {
+      return delegate.getResult();
+    }
+
+    private class OtelDecoratingWritableByteChannel implements WritableByteChannel {
+
+      private final WritableByteChannel delegate;
+      private final Span openSpan;
+
+      private OtelDecoratingWritableByteChannel(WritableByteChannel delegate, Span openSpan) {
+        this.delegate = delegate;
+        this.openSpan = openSpan;
+      }
+
+      @Override
+      public int write(ByteBuffer src) throws IOException {
+        return delegate.write(src);
+      }
+
+      @Override
+      public boolean isOpen() {
+        return delegate.isOpen();
+      }
+
+      @Override
+      public void close() throws IOException {
+        try {
+          delegate.close();
+        } finally {
+          openSpan.end();
+        }
+      }
+    }
+  }
+
+  private static final class OtelDecoratedWriteChannel implements WriteChannel {
+    private final WriteChannel delegate;
+    private final Span openSpan;
+
+    private OtelDecoratedWriteChannel(WriteChannel delegate, Span openSpan) {
+      this.delegate = delegate;
+      this.openSpan = openSpan;
+    }
+
+    @Override
+    public void setChunkSize(int chunkSize) {
+      delegate.setChunkSize(chunkSize);
+    }
+
+    @Override
+    public RestorableState<WriteChannel> capture() {
+      return delegate.capture();
+    }
+
+    @Override
+    public int write(ByteBuffer src) throws IOException {
+      return delegate.write(src);
+    }
+
+    @Override
+    public boolean isOpen() {
+      return delegate.isOpen();
+    }
+
+    @Override
+    public void close() throws IOException {
+      try {
+        delegate.close();
+      } finally {
+        openSpan.end();
+      }
+    }
+  }
+
+  private final class OtelDecoratedCopyWriter extends CopyWriter {
+
+    private final CopyWriter copyWriter;
+    private final Span span;
+    private final Context parentContext;
+    private final Tracer tracer;
+
+    public OtelDecoratedCopyWriter(CopyWriter copyWriter, Span span) {
+      this.copyWriter = copyWriter;
+      this.span = span;
+      this.parentContext = Context.current();
+      this.tracer =
+          TracerDecorator.decorate(
+              Context.current(),
+              otel,
+              OtelStorageDecorator.this.baseAttributes,
+              CopyWriter.class.getName() + "/");
+    }
+
+    @Override
+    public Blob getResult() {
+      try {
+        return copyWriter.getResult();
+      } catch (Throwable t) {
+        span.recordException(t);
+        span.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+        throw t;
+      } finally {
+        span.end();
+      }
+    }
+
+    @Override
+    public long getBlobSize() {
+      return copyWriter.getBlobSize();
+    }
+
+    @Override
+    public boolean isDone() {
+      boolean done = copyWriter.isDone();
+      if (done) {
+        span.end();
+      }
+      return done;
+    }
+
+    @Override
+    public long getTotalBytesCopied() {
+      return copyWriter.getTotalBytesCopied();
+    }
+
+    @Override
+    public RestorableState<CopyWriter> capture() {
+      return copyWriter.capture();
+    }
+
+    @Override
+    public void copyChunk() {
+      Span copyChunkSpan = tracer.spanBuilder("copyChunk").setParent(parentContext).startSpan();
+      try (Scope ignore = copyChunkSpan.makeCurrent()) {
+        copyWriter.copyChunk();
+      } catch (Throwable t) {
+        copyChunkSpan.recordException(t);
+        copyChunkSpan.setStatus(StatusCode.ERROR, t.getClass().getSimpleName());
+        span.end();
+        throw t;
+      } finally {
+        copyChunkSpan.end();
+      }
+    }
+  }
+}
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
index db807ce6c..b824acf4d 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/ParallelCompositeUploadWritableByteChannel.java
@@ -54,6 +54,7 @@
 import com.google.common.hash.Hasher;
 import com.google.common.hash.Hashing;
 import io.grpc.Status.Code;
+import io.opentelemetry.context.Context;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousCloseException;
@@ -145,7 +146,7 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
       BlobInfo ultimateObject,
       Opts<ObjectTargetOpt> opts) {
     this.bufferPool = bufferPool;
-    this.exec = exec;
+    this.exec = Context.current().wrap(exec);
     this.partNamingStrategy = partNamingStrategy;
     this.partCleanupStrategy = partCleanupStrategy;
     this.maxElementsPerCompact = maxElementsPerCompact;
@@ -154,7 +155,7 @@ final class ParallelCompositeUploadWritableByteChannel implements BufferedWritab
     this.storage = storage;
     this.ultimateObject = ultimateObject;
     this.opts = opts;
-    this.queue = AsyncAppendingQueue.of(exec, maxElementsPerCompact, this::compose);
+    this.queue = AsyncAppendingQueue.of(this.exec, maxElementsPerCompact, this::compose);
     this.pendingParts = new ArrayList<>();
     // this can be modified by another thread
     this.successfulParts = Collections.synchronizedList(new ArrayList<>());
diff --git a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
index 81d787f76..f149db110 100644
--- a/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
+++ b/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageOptions.java
@@ -29,6 +29,7 @@
 import com.google.cloud.storage.Storage.BlobWriteOption;
 import com.google.cloud.storage.TransportCompatibility.Transport;
 import com.google.cloud.storage.spi.StorageRpcFactory;
+import io.opentelemetry.api.OpenTelemetry;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Properties;
@@ -110,6 +111,15 @@ public abstract static class Builder
     public abstract StorageOptions.Builder setBlobWriteSessionConfig(
         @NonNull BlobWriteSessionConfig blobWriteSessionConfig);
 
+    /**
+     * Enable OpenTelemetry Tracing and provide an instance for the client to use.
+     *
+     * @param openTelemetry User defined instance of OpenTelemetry to be used by the library
+     * @since 2.47.0 This new api is in preview and is subject to breaking changes.
+     */
+    @BetaApi
+    public abstract StorageOptions.Builder setOpenTelemetry(OpenTelemetry openTelemetry);
+
     @Override
     public abstract StorageOptions build();
   }
@@ -144,6 +154,10 @@ public static String version() {
     return VERSION;
   }
 
+  /** @since 2.47.0 This new api is in preview and is subject to breaking changes. */
+  @BetaApi
+  public abstract OpenTelemetry getOpenTelemetry();
+
   @SuppressWarnings("unchecked")
   @Override
   public abstract StorageOptions.Builder toBuilder();
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java
new file mode 100644
index 000000000..3b8957bba
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java
@@ -0,0 +1,91 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.cloud.storage.TestUtils.assertAll;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.cloud.storage.otel.TestExporter;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = Backend.PROD,
+    transports = {Transport.HTTP, Transport.GRPC})
+public final class ITOpenTelemetryTest {
+
+  @Inject public Storage storage;
+
+  @Inject public BucketInfo bucket;
+
+  @Inject public Generator generator;
+  @Inject public Transport transport;
+
+  @Test
+  public void checkInstrumentation() throws Exception {
+    TestExporter exporter = new TestExporter();
+
+    OpenTelemetrySdk openTelemetrySdk =
+        OpenTelemetrySdk.builder()
+            .setTracerProvider(
+                SdkTracerProvider.builder()
+                    .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+                    .build())
+            .build();
+    StorageOptions storageOptions =
+        storage.getOptions().toBuilder().setOpenTelemetry(openTelemetrySdk).build();
+    try (Storage storage = storageOptions.getService()) {
+      storage.create(BlobInfo.newBuilder(bucket, generator.randomObjectName()).build());
+    }
+
+    SpanData spanData = exporter.getExportedSpans().get(0);
+    assertAll(
+        () -> assertThat(getAttributeValue(spanData, "gcp.client.service")).isEqualTo("Storage"),
+        () ->
+            assertThat(getAttributeValue(spanData, "gcp.client.repo"))
+                .isEqualTo("googleapis/java-storage"),
+        () ->
+            assertThat(getAttributeValue(spanData, "gcp.client.artifact"))
+                .isEqualTo("com.google.cloud:google-cloud-storage"),
+        () ->
+            assertThat(getAttributeValue(spanData, "rpc.system"))
+                .isEqualTo(transport.name().toLowerCase()));
+  }
+
+  @Test
+  public void noOpDoesNothing() {
+    assertThat(storage.getOptions().getOpenTelemetry()).isSameInstanceAs(OpenTelemetry.noop());
+    storage.create(BlobInfo.newBuilder(bucket, generator.randomObjectName()).build());
+  }
+
+  private static String getAttributeValue(SpanData spanData, String key) {
+    return spanData.getAttributes().get(AttributeKey.stringKey(key));
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTestbenchTest.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTestbenchTest.java
new file mode 100644
index 000000000..f813fcf41
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTestbenchTest.java
@@ -0,0 +1,217 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static com.google.common.truth.Truth.assertThat;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.WriteChannel;
+import com.google.cloud.storage.Storage.BlobSourceOption;
+import com.google.cloud.storage.Storage.BlobTargetOption;
+import com.google.cloud.storage.Storage.CopyRequest;
+import com.google.cloud.storage.TransportCompatibility.Transport;
+import com.google.cloud.storage.it.runner.StorageITRunner;
+import com.google.cloud.storage.it.runner.annotations.Backend;
+import com.google.cloud.storage.it.runner.annotations.CrossRun;
+import com.google.cloud.storage.it.runner.annotations.Inject;
+import com.google.cloud.storage.it.runner.registry.Generator;
+import com.google.cloud.storage.otel.TestExporter;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.UUID;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(StorageITRunner.class)
+@CrossRun(
+    backends = Backend.TEST_BENCH,
+    transports = {Transport.HTTP, Transport.GRPC})
+public class ITOpenTelemetryTestbenchTest {
+  @Inject public Transport transport;
+  @Inject public Generator generator;
+  @Inject public BucketInfo testBucket;
+  @Inject public Storage storage;
+  private SpanExporter exporter;
+  private static final byte[] helloWorldTextBytes = "hello world".getBytes();
+  private BlobId blobId;
+  private static final Path tmpDir = Paths.get(System.getProperty("java.io.tmpdir"));
+
+  @Before
+  public void setUp() {
+    exporter = new TestExporter();
+    OpenTelemetrySdk openTelemetrySdk =
+        OpenTelemetrySdk.builder()
+            .setTracerProvider(
+                SdkTracerProvider.builder()
+                    .addSpanProcessor(SimpleSpanProcessor.create(exporter))
+                    .build())
+            .build();
+    StorageOptions options =
+        storage.getOptions().toBuilder().setOpenTelemetry(openTelemetrySdk).build();
+    storage = options.getService();
+    String objectString = generator.randomObjectName();
+    blobId = BlobId.of(testBucket.getName(), objectString);
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (storage != null) {
+      storage.close();
+    }
+  }
+
+  @Test
+  public void runCreateBucket() {
+    String bucket = "random-bucket" + UUID.randomUUID();
+    storage.create(BucketInfo.of(bucket));
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+  }
+
+  @Test
+  public void runCreateBlob() {
+    byte[] content = "Hello, World!".getBytes(UTF_8);
+    BlobId toCreate = BlobId.of(testBucket.getName(), generator.randomObjectName());
+    storage.create(BlobInfo.newBuilder(toCreate).build(), content);
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+    Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("create")));
+  }
+
+  @Test
+  public void runReadAllBytes() {
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/plain").build();
+    storage.create(blobInfo, helloWorldTextBytes);
+    byte[] read = storage.readAllBytes(blobId);
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+    Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("readAllBytes")));
+  }
+
+  @Test
+  public void runCreateFrom() throws IOException {
+    Path helloWorldTxtGz = File.createTempFile(blobId.getName(), ".txt.gz").toPath();
+    storage.createFrom(BlobInfo.newBuilder(blobId).build(), helloWorldTxtGz);
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+    Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("createFrom")));
+  }
+
+  @Test
+  public void runDownloadToPath() throws IOException {
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/plain").build();
+    storage.create(blobInfo, helloWorldTextBytes);
+    try (TmpFile file = TmpFile.of(tmpDir, "download-to", ".txt")) {
+      storage.downloadTo(blobId, file.getPath());
+      TestExporter testExported = (TestExporter) exporter;
+      List<SpanData> spanData = testExported.getExportedSpans();
+      checkCommonAttributes(spanData);
+      Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("downloadTo")));
+    }
+  }
+
+  @Test
+  public void runDownloadToOutputStream() {
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/plain").build();
+    storage.create(blobInfo, helloWorldTextBytes);
+    ByteArrayOutputStream baos = new ByteArrayOutputStream();
+    storage.downloadTo(blobId, baos);
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+    Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("downloadTo")));
+  }
+
+  @Test
+  public void runCopy() {
+    BlobInfo info =
+        BlobInfo.newBuilder(testBucket, generator.randomObjectName() + "copy/src").build();
+    Blob cpySrc = storage.create(info, helloWorldTextBytes, BlobTargetOption.doesNotExist());
+    BlobInfo dst =
+        BlobInfo.newBuilder(testBucket, generator.randomObjectName() + "copy/dst").build();
+    CopyRequest copyRequest =
+        CopyRequest.newBuilder()
+            .setSource(cpySrc.getBlobId())
+            .setSourceOptions(BlobSourceOption.generationMatch(cpySrc.getGeneration()))
+            .setTarget(dst, BlobTargetOption.doesNotExist())
+            .build();
+    CopyWriter copyWriter = storage.copy(copyRequest);
+    BlobInfo result = copyWriter.getResult();
+    assertThat(result).isNotNull();
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+    Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("copy")));
+  }
+
+  @Test
+  public void runWriter() throws IOException {
+    BlobInfo info = BlobInfo.newBuilder(testBucket, generator.randomObjectName()).build();
+    try (WriteChannel writer = storage.writer(info)) {
+      // Do nothing
+    }
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+    Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("writer")));
+  }
+
+  @Test
+  public void runReader() throws IOException {
+    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/plain").build();
+    storage.create(blobInfo, helloWorldTextBytes);
+    try (ReadChannel reader = storage.reader(blobId)) {
+      // Do nothing
+    }
+    TestExporter testExported = (TestExporter) exporter;
+    List<SpanData> spanData = testExported.getExportedSpans();
+    checkCommonAttributes(spanData);
+    Assert.assertTrue(spanData.stream().anyMatch(x -> x.getName().contains("reader")));
+  }
+
+  private void checkCommonAttributes(List<SpanData> spanData) {
+    for (SpanData span : spanData) {
+      Assert.assertEquals("Storage", getAttributeValue(span, "gcp.client.service"));
+      Assert.assertEquals("googleapis/java-storage", getAttributeValue(span, "gcp.client.repo"));
+      Assert.assertEquals(
+          "com.google.cloud:google-cloud-storage", getAttributeValue(span, "gcp.client.artifact"));
+      Assert.assertEquals(transport.name().toLowerCase(), getAttributeValue(span, "rpc.system"));
+    }
+  }
+
+  private String getAttributeValue(SpanData spanData, String key) {
+    return spanData.getAttributes().get(AttributeKey.stringKey(key)).toString();
+  }
+}
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
index d40a29954..470006c9c 100644
--- a/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/TestBench.java
@@ -112,7 +112,7 @@ private TestBench(
       String dockerImageName,
       String dockerImageTag,
       String containerName) {
-    this.ignorePullError = ignorePullError;
+    this.ignorePullError = true;
     this.baseUri = baseUri;
     this.gRPCBaseUri = gRPCBaseUri;
     this.dockerImageName = dockerImageName;
diff --git a/google-cloud-storage/src/test/java/com/google/cloud/storage/otel/TestExporter.java b/google-cloud-storage/src/test/java/com/google/cloud/storage/otel/TestExporter.java
new file mode 100644
index 000000000..1532fa836
--- /dev/null
+++ b/google-cloud-storage/src/test/java/com/google/cloud/storage/otel/TestExporter.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage.otel;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+public class TestExporter implements SpanExporter {
+
+  public final List<SpanData> exportedSpans = Collections.synchronizedList(new ArrayList<>());
+
+  @Override
+  public CompletableResultCode export(Collection<SpanData> spans) {
+    exportedSpans.addAll(spans);
+    return CompletableResultCode.ofSuccess();
+  }
+
+  @Override
+  public CompletableResultCode flush() {
+    return null;
+  }
+
+  @Override
+  public CompletableResultCode shutdown() {
+    return null;
+  }
+
+  public List<SpanData> getExportedSpans() {
+    return exportedSpans;
+  }
+}