fix(storage): add App Hub storage.googleapis.com prefix to destination.id (#14527)

## Description
Fixes the formatting of the `gcp.resource.destination.id` OpenTelemetry
span attribute emitted by the Storage SDK to comply with the Cloud Trace
App Hub URI extractor.

The previous implementation emitted bare
`projects/{project}/buckets/{bucket}` paths, which are rejected by Cloud
Trace. This PR ensures the resource payload explicitly starts with
`//storage.googleapis.com/` (e.g.
`//storage.googleapis.com/projects/{project}/buckets/{bucket}`).

Follows up on PR #13248.

Author: cpriti-os

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

@@ -36,7 +36,8 @@ final class AcoSpanBuilder implements SpanBuilder {
   private static final String MULTI_REGION = "multi-region";
   private static final String DUAL_REGION = "dual-region";
   private static final String PLACEHOLDER_BUCKET_LOCATION = "global";
-  private static final String PLACEHOLDER_RESOURCE_PREFIX = "projects/_/buckets/";
+  private static final String PLACEHOLDER_RESOURCE_PREFIX =
+      "//storage.googleapis.com/projects/_/buckets/";
 
   private final SpanBuilder delegate;
   private final OtelStorageDecorator parent;
@@ -184,7 +185,7 @@ static Tuple<String, String> fetch(Storage delegate, String bucketName) {
     String projectId = bucket.getProject() != null ? bucket.getProject().toString() : null;
     String resource;
     if (projectId != null && !projectId.isEmpty()) {
-      resource = "projects/" + projectId + "/buckets/" + bucketName;
+      resource = "//storage.googleapis.com/projects/" + projectId + "/buckets/" + bucketName;
     } else {
       resource = PLACEHOLDER_RESOURCE_PREFIX + bucketName;
     }

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java

@@ -185,7 +185,7 @@ public void testAcoForbiddenBucketFallbackAttributes() throws Exception {
     assertAll(
         () ->
             assertThat(getAttributeValue(getSpan2, "gcp.resource.destination.id"))
-                .isEqualTo("projects/_/buckets/test"),
+                .isEqualTo("//storage.googleapis.com/projects/_/buckets/test"),
         () ->
             assertThat(getAttributeValue(getSpan2, "gcp.resource.destination.location"))
                 .isEqualTo("global"));

java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OtelStorageDecoratorAcoUnitTest.java

@@ -84,7 +84,7 @@ public void testAcoSuccessFlow() throws Exception {
 
       BucketMetadataCache.BucketMetadata meta = osd.acoContext.getCache().get("success-bucket");
       assertNotNull(meta);
-      assertEquals("projects/12345/buckets/success-bucket", meta.resource);
+      assertEquals("//storage.googleapis.com/projects/12345/buckets/success-bucket", meta.resource);
       assertEquals("us-east1", meta.location);
       assertFalse(meta.fetchPending);
     }
@@ -176,7 +176,7 @@ public void testAco403ForbiddenFlow() throws Exception {
       // Forbidden -> Fallback values retained with pending = false (Do Not Retry)
       BucketMetadataCache.BucketMetadata meta = osd.acoContext.getCache().get("forbidden-bucket");
       assertNotNull(meta);
-      assertEquals("projects/_/buckets/forbidden-bucket", meta.resource);
+      assertEquals("//storage.googleapis.com/projects/_/buckets/forbidden-bucket", meta.resource);
       assertEquals("global", meta.location);
       assertFalse(meta.fetchPending);
     }
@@ -232,7 +232,11 @@ public void testAcoAcoSpanEndSkipsPending() throws Exception {
       // Manually put a pending placeholder entry
       osd.acoContext
           .getCache()
-          .put("pending-bucket", "projects/_/buckets/pending-bucket", "global", true);
+          .put(
+              "pending-bucket",
+              "//storage.googleapis.com/projects/_/buckets/pending-bucket",
+              "global",
+              true);
 
       Span mockSpan = mock(Span.class);
       AcoSpan acoSpan = new AcoSpan(mockSpan, "pending-bucket", osd);
@@ -257,7 +261,11 @@ public void testAcoAcoSpanEndAppliesResolved() throws Exception {
       // Manually put a resolved non-pending entry
       osd.acoContext
           .getCache()
-          .put("resolved-bucket", "projects/123/buckets/resolved-bucket", "us-east1", false);
+          .put(
+              "resolved-bucket",
+              "//storage.googleapis.com/projects/123/buckets/resolved-bucket",
+              "us-east1",
+              false);
 
       Span mockSpan = mock(Span.class);
       AcoSpan acoSpan = new AcoSpan(mockSpan, "resolved-bucket", osd);
@@ -266,7 +274,9 @@ public void testAcoAcoSpanEndAppliesResolved() throws Exception {
 
       // Verify OTel span attributes were set successfully
       Mockito.verify(mockSpan)
-          .setAttribute("gcp.resource.destination.id", "projects/123/buckets/resolved-bucket");
+          .setAttribute(
+              "gcp.resource.destination.id",
+              "//storage.googleapis.com/projects/123/buckets/resolved-bucket");
       Mockito.verify(mockSpan).setAttribute("gcp.resource.destination.location", "us-east1");
     }
   }