From 6bd1b9aef3ff3804d948384e6642e7a7f402349a Mon Sep 17 00:00:00 2001 From: Dhriti Chopra Date: Sat, 31 Oct 2026 15:46:16 +0000 Subject: [PATCH] feat(storage): add fast open tracing attributes Eighth step of fast open BiDi reads (go/fast-open-bidi-read, b/552263062). With fast open, the blobReadSession span also covers reading the first range, and a readAs served from the fast open read takes ~0 ms with no request. Add span attributes so traces explain this: - gcs.bidi.fast_open (true/false) on the blobReadSession span: whether a read was sent with the open. - gcs.bidi.cache_status (HIT/MISS) on each readAs span of a session opened with fast open: HIT when the readAs was served from the fast open read, MISS when it sent a new read. Sessions without fast open don't get it. --- .../cloud/storage/OtelStorageDecorator.java | 17 ++++- .../cloud/storage/ITFastOpenFakeTest.java | 72 +++++++++++++++++++ 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java index 3546027c2fdd..cbf67ef9914e 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java @@ -1488,6 +1488,7 @@ public ApiFuture blobReadSession( tracer .spanBuilder(BLOB_READ_SESSION) .setAttribute("gsutil.uri", id.toGsUtilUriWithGeneration()) + .setAttribute("gcs.bidi.fast_open", fastOpenConfig != null) .startSpan(); try (Scope ignore1 = blobReadSessionSpan.makeCurrent()) { Context blobReadSessionContext = Context.current(); @@ -1500,7 +1501,11 @@ public ApiFuture blobReadSession( delegate -> { ready.end(); return new OtelDecoratingBlobReadSession( - delegate, id, blobReadSessionContext, blobReadSessionSpan); + delegate, + id, + blobReadSessionContext, + blobReadSessionSpan, + fastOpenConfig != null); }, MoreExecutors.directExecutor()); ApiFutures.addCallback( @@ -1909,6 +1914,7 @@ ReadProjectionConfig unwrap() { @Override void onFastOpenReadClaimed(Projection projection) { + parentSpan.setAttribute("gcs.bidi.cache_status", "HIT"); if (!(projection instanceof ApiFuture)) { parentSpan.end(); return; @@ -2011,16 +2017,19 @@ class OtelDecoratingBlobReadSession implements BlobReadSession { private final BlobId id; private final Context blobReadSessionContext; private final Span blobReadSessionSpan; + private final boolean fastOpen; private OtelDecoratingBlobReadSession( BlobReadSession delegate, BlobId id, Context blobReadSessionContext, - Span blobReadSessionSpan) { + Span blobReadSessionSpan, + boolean fastOpen) { this.delegate = delegate; this.id = id; this.blobReadSessionContext = blobReadSessionContext; this.blobReadSessionSpan = blobReadSessionSpan; + this.fastOpen = fastOpen; } @Override @@ -2036,6 +2045,10 @@ public Projection readAs(ReadProjectionConfig config) { .setAttribute("gsutil.uri", id.toGsUtilUriWithGeneration()) .setParent(blobReadSessionContext) .startSpan(); + if (fastOpen) { + // changed to HIT if this readAs is served from the fast open read + readRangeSpan.setAttribute("gcs.bidi.cache_status", "MISS"); + } try (Scope ignore2 = readRangeSpan.makeCurrent()) { OtelReadProjectionConfig c = new OtelReadProjectionConfig<>(config, readRangeSpan); diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java index 694ce957e903..5a23fb3f4045 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java @@ -50,6 +50,7 @@ import io.grpc.Metadata; import io.grpc.Status; import io.grpc.stub.StreamObserver; +import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.trace.StatusCode; import io.opentelemetry.sdk.OpenTelemetrySdk; import io.opentelemetry.sdk.trace.SdkTracerProvider; @@ -62,6 +63,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicInteger; @@ -80,6 +82,10 @@ public final class ITFastOpenFakeTest { .build(); private static final GrpcCallContext CTX = GrpcCallContext.createDefault(); private static final BlobId BLOB_ID = BlobId.of("b", "o"); + private static final AttributeKey FAST_OPEN = + AttributeKey.booleanKey("gcs.bidi.fast_open"); + private static final AttributeKey CACHE_STATUS = + AttributeKey.stringKey("gcs.bidi.cache_status"); private static final BidiReadObjectRequest OPEN = BidiReadObjectRequest.newBuilder() @@ -560,6 +566,48 @@ public void blobReadSession_withTracing_whenTheFastOpenReadFails_recordsTheError assertThat(readAsStatus).containsExactly(StatusCode.ERROR); } + @Test + public void blobReadSession_withTracing_recordsFastOpenAndCacheStatus() throws Exception { + FakeStorage server = + fakeServer( + ImmutableMap.of( + OPEN_WITH_READ_OF_10_TO_20, + metadataAndBytes(1, 10, 10), + read(2, 20, 10), + bytes(2, 20, 10))); + TestExporter exporter = new TestExporter(); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = tracedStorage(fakeServer, exporter)) { + try (BlobReadSession session = + storage.blobReadSession(BLOB_ID, READ_10_TO_20).get(5, SECONDS)) { + session.readAs(READ_10_TO_20).get(2, SECONDS); + session.readAs(READ_20_TO_30).get(2, SECONDS); + } + } + + assertThat(attributeOf(exporter, "blobReadSession", FAST_OPEN)).containsExactly(true); + assertThat(attributeOf(exporter, "blobReadSession/readAs", CACHE_STATUS)) + .containsExactly("HIT", "MISS"); + } + + @Test + public void blobReadSession_withTracing_withoutFastOpen_recordsNoCacheStatus() throws Exception { + FakeStorage server = + fakeServer(ImmutableMap.of(OPEN, metadataOnly(), read(1, 10, 10), bytes(1, 10, 10))); + TestExporter exporter = new TestExporter(); + + try (FakeServer fakeServer = FakeServer.of(server); + Storage storage = tracedStorage(fakeServer, exporter)) { + try (BlobReadSession session = storage.blobReadSession(BLOB_ID).get(5, SECONDS)) { + session.readAs(READ_10_TO_20).get(2, SECONDS); + } + } + + assertThat(attributeOf(exporter, "blobReadSession", FAST_OPEN)).containsExactly(false); + assertThat(attributeOf(exporter, "blobReadSession/readAs", CACHE_STATUS)).isEmpty(); + } + @Test public void blobReadSession_withANullFastOpenConfig_isAPlainOpen() throws Exception { FakeStorage server = fakeServer(ImmutableMap.of(OPEN, metadataOnly())); @@ -713,6 +761,30 @@ private static byte[] slice(int offset, int length) { return Arrays.copyOfRange(ALL_OBJECT_BYTES, offset, offset + length); } + private static Storage tracedStorage(FakeServer fakeServer, TestExporter exporter) { + OpenTelemetrySdk otel = + OpenTelemetrySdk.builder() + .setTracerProvider( + SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(exporter)) + .build()) + .build(); + return fakeServer.getGrpcStorageOptions().toBuilder() + .setOpenTelemetry(otel) + .build() + .getService(); + } + + /** The non-null values of {@code key} on the ended spans named {@code spanName}. */ + private static List attributeOf( + TestExporter exporter, String spanName, AttributeKey key) { + return exporter.getExportedSpans().stream() + .filter(s -> s.getName().equals("com.google.cloud.storage.Storage/" + spanName)) + .map(s -> s.getAttributes().get(key)) + .filter(Objects::nonNull) + .collect(toList()); + } + /** Wraps a config the way tracing does, and records each fast open hit. */ private static final class WrappedConfig extends ReadProjectionConfig> { private final ReadAsFutureBytes delegate;