Skip to content

feat(storage): add fast open tracing attributes #14877

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Dhriti07 wants to merge 1 commit into
base: fast-open/07-public-api
Choose a base branch
from
Draft

feat(storage): add fast open tracing attributes #14877

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions ...age/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1488,6 +1488,7 @@ public <Projection> ApiFuture<BlobReadSession> 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();
Expand All @@ -1500,7 +1501,11 @@ public <Projection> ApiFuture<BlobReadSession> blobReadSession(
delegate -> {
ready.end();
return new OtelDecoratingBlobReadSession(
delegate, id, blobReadSessionContext, blobReadSessionSpan);
delegate,
id,
blobReadSessionContext,
blobReadSessionSpan,
fastOpenConfig != null);
},
MoreExecutors.directExecutor());
ApiFutures.addCallback(
Expand Down Expand Up @@ -1909,6 +1914,7 @@ ReadProjectionConfig<Projection> unwrap() {

@Override
void onFastOpenReadClaimed(Projection projection) {
parentSpan.setAttribute("gcs.bidi.cache_status", "HIT");
if (!(projection instanceof ApiFuture)) {
parentSpan.end();
return;
Expand Down Expand Up @@ -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
Expand All @@ -2036,6 +2045,10 @@ public <Projection> Projection readAs(ReadProjectionConfig<Projection> 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<Projection> c =
new OtelReadProjectionConfig<>(config, readRangeSpan);
Expand Down
72 changes: 72 additions & 0 deletions ...orage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITFastOpenFakeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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<Boolean> FAST_OPEN =
AttributeKey.booleanKey("gcs.bidi.fast_open");
private static final AttributeKey<String> CACHE_STATUS =
AttributeKey.stringKey("gcs.bidi.cache_status");

private static final BidiReadObjectRequest OPEN =
BidiReadObjectRequest.newBuilder()
Expand Down Expand Up @@ -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()));
Expand Down Expand Up @@ -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 <T> List<T> attributeOf(
TestExporter exporter, String spanName, AttributeKey<T> 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<ApiFuture<byte[]>> {
private final ReadAsFutureBytes delegate;
Expand Down
Loading