FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

feat(bigquery): Integrate Otel Tracing in storage lib by whuffman36 · Pull Request #2983 · googleapis/java-bigquerystorage · GitHub

This repository was archived by the owner on Feb 24, 2026. It is now read-only.
/ java-bigquerystorage Public archive

feat(bigquery): Integrate Otel Tracing in storage lib #2983

Merged
whuffman36 merged 6 commits into from
May 30, 2025
Conversation 3 Commits 6 Checks 0 Files changed
Merged

feat(bigquery): Integrate Otel Tracing in storage lib #2983

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
67bbe12
feat(bigquery): Integrate Otel Tracing in storage lib
whuffman36 May 23, 2025
1780e86
fix style
whuffman36 May 23, 2025
4a4770c
add sdk dependencies
whuffman36 May 23, 2025
5e217e8
Improve IT test coverage
whuffman36 May 23, 2025
b8bb754
fix IT test failures
whuffman36 May 27, 2025
8f6a581
Move otel init to specific test
whuffman36 May 28, 2025
File filter

Filter by extension

Filter by extension .java  (5) .xml  (2) All 2 file types selected
Only manifest files
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
19 changes: 19 additions & 0 deletions google-cloud-bigquerystorage/pom.xml
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -167,8 +167,27 @@
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-api</artifactId>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-context</artifactId>
</dependency>

<!-- Test dependencies -->
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk-common</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.opentelemetry</groupId>
<artifactId>opentelemetry-sdk-trace</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
Expand Down
217 changes: 207 additions & 10 deletions ...igquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/BigQueryReadClient.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,9 @@
import com.google.api.gax.rpc.ServerStreamingCallable;
import com.google.api.gax.rpc.UnaryCallable;
import com.google.cloud.bigquery.storage.v1.stub.EnhancedBigQueryReadStub;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.context.Scope;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

Expand Down Expand Up @@ -128,7 +131,9 @@ protected BigQueryReadClient(BigQueryReadSettings settings) throws IOException {
this.settings = settings;
this.stub =
EnhancedBigQueryReadStub.create(
settings.getTypedStubSettings(), settings.getReadRowsRetryAttemptListener());
settings.getTypedStubSettings(),
settings.getReadRowsRetryAttemptListener(),
settings.isOpenTelemetryEnabled());
}

@BetaApi("A restructuring of stub classes is planned, so this may break in the future")
Expand Down Expand Up @@ -229,7 +234,32 @@ public final ReadSession createReadSession(
* @throws com.google.api.gax.rpc.ApiException if the remote call fails
*/
public final ReadSession createReadSession(CreateReadSessionRequest request) {
return createReadSessionCallable().call(request);
Span createReadSession = null;
if (settings.isOpenTelemetryEnabled()) {
createReadSession =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.createReadSession")
.setAttribute(
"bq.storage.read_session.request.parent", getFieldAsString(request.getParent()))
.setAttribute(
"bq.storage.read_session.request.max_stream_count", request.getMaxStreamCount())
.setAttribute(
"bq.storage.read_session.request.preferred_min_stream_count",
request.getPreferredMinStreamCount())
.setAttribute(
"bq.storage.read_session.request.serialized_size", request.getSerializedSize())
.setAllAttributes(otelAttributesFrom(request.getReadSession()))
.startSpan();
}
try (Scope createReadSessionScope =
createReadSession != null ? createReadSession.makeCurrent() : null) {
return createReadSessionCallable().call(request);
} finally {
if (createReadSession != null) {
createReadSession.end();
}
}
}

/**
Expand Down Expand Up @@ -262,7 +292,22 @@ public final ReadSession createReadSession(CreateReadSessionRequest request) {
* </code></pre>
*/
public final UnaryCallable<CreateReadSessionRequest, ReadSession> createReadSessionCallable() {
return stub.createReadSessionCallable();
Span createReadSessionCallable = null;
if (settings.isOpenTelemetryEnabled()) {
createReadSessionCallable =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.createReadSessionCallable")
.startSpan();
}
try (Scope createReadSessionCallableScope =
createReadSessionCallable != null ? createReadSessionCallable.makeCurrent() : null) {
return stub.createReadSessionCallable();
} finally {
if (createReadSessionCallable != null) {
createReadSessionCallable.end();
}
}
}

/**
Expand All @@ -287,7 +332,22 @@ public final UnaryCallable<CreateReadSessionRequest, ReadSession> createReadSess
* </code></pre>
*/
public final ServerStreamingCallable<ReadRowsRequest, ReadRowsResponse> readRowsCallable() {
return stub.readRowsCallable();
Span readRowsCallable = null;
if (settings.isOpenTelemetryEnabled()) {
readRowsCallable =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.readRowsCallable")

shollyman May 28, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

The behavior here is we'll get a single span that represents the lifetime of the server streaming rpc? Can/should we add an attribute implicating the stream ID requested from the ReadRowsRequest?

whuffman36 May 28, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

Stream ID seems like a good field to capture. I could be mistaken, but it seems like the ReadRowsRequest is handled by generated code and in this context we don't have access to the stream ID (ie can't put it in the attributes). Do you know of a way we can get the stream ID in the ReadClient or its enhanced stub @Neenu1995 ?

Neenu1995 May 29, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality

The jdbc gets the StreamName from the readSession. I don't know if stream ID is accessible anywhere else.
.startSpan();
}
try (Scope readRowsCallableScope =
readRowsCallable != null ? readRowsCallable.makeCurrent() : null) {
return stub.readRowsCallable();
} finally {
if (readRowsCallable != null) {
readRowsCallable.end();
}
}
}

/**
Expand Down Expand Up @@ -315,7 +375,23 @@ public final ServerStreamingCallable<ReadRowsRequest, ReadRowsResponse> readRows
* @throws com.google.api.gax.rpc.ApiException if the remote call fails
*/
public final SplitReadStreamResponse splitReadStream(SplitReadStreamRequest request) {
return splitReadStreamCallable().call(request);
Span splitReadStream = null;
if (settings.isOpenTelemetryEnabled()) {
splitReadStream =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.splitReadStream")
.setAllAttributes(otelAttributesFrom(request))
.startSpan();
}
try (Scope splitReadStreamScope =
splitReadStream != null ? splitReadStream.makeCurrent() : null) {
return splitReadStreamCallable().call(request);
} finally {
if (splitReadStream != null) {
splitReadStream.end();
}
}
}

/**
Expand Down Expand Up @@ -343,17 +419,60 @@ public final SplitReadStreamResponse splitReadStream(SplitReadStreamRequest requ
*/
public final UnaryCallable<SplitReadStreamRequest, SplitReadStreamResponse>
splitReadStreamCallable() {
return stub.splitReadStreamCallable();
Span splitReadStreamCallable = null;
if (settings.isOpenTelemetryEnabled()) {
splitReadStreamCallable =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.splitReadStreamCallable")
.startSpan();
}
try (Scope readRowsCallableScope =
splitReadStreamCallable != null ? splitReadStreamCallable.makeCurrent() : null) {
return stub.splitReadStreamCallable();
} finally {
if (splitReadStreamCallable != null) {
splitReadStreamCallable.end();
}
}
}

@Override
public final void close() {
stub.close();
Span close = null;
if (settings.isOpenTelemetryEnabled()) {
close =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.close")
.startSpan();
}
try (Scope closeScope = close != null ? close.makeCurrent() : null) {
stub.close();
} finally {
if (close != null) {
close.end();
}
}
}

@Override
public void shutdown() {
stub.shutdown();
Span shutdown = null;
if (settings.isOpenTelemetryEnabled()) {
shutdown =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.shutdown")
.startSpan();
}
try (Scope shutdownScope = shutdown != null ? shutdown.makeCurrent() : null) {
stub.shutdown();
} finally {
if (shutdown != null) {
shutdown.end();
}
}
}

@Override
Expand All @@ -368,11 +487,89 @@ public boolean isTerminated() {

@Override
public void shutdownNow() {
stub.shutdownNow();
Span shutdownNow = null;
if (settings.isOpenTelemetryEnabled()) {
shutdownNow =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.shutdownNow")
.startSpan();
}
try (Scope shutdownNowScope = shutdownNow != null ? shutdownNow.makeCurrent() : null) {
stub.shutdownNow();
} finally {
if (shutdownNow != null) {
shutdownNow.end();
}
}
}

@Override
public boolean awaitTermination(long duration, TimeUnit unit) throws InterruptedException {
return stub.awaitTermination(duration, unit);
Span awaitTermination = null;
if (settings.isOpenTelemetryEnabled()) {
awaitTermination =
settings
.getOpenTelemetryTracer()
.spanBuilder("com.google.cloud.bigquery.storage.v1.read.awaitTermination")
.setAttribute("duration", duration)
.setAttribute("unit", unit.toString())
.startSpan();
}
try (Scope awaitTerminationScope =
awaitTermination != null ? awaitTermination.makeCurrent() : null) {
return stub.awaitTermination(duration, unit);
} finally {
if (awaitTermination != null) {
awaitTermination.end();
}
}
}

public void disableOpenTelemetryTracing() {
settings.setEnableOpenTelemetryTracing(false);
}

public void enableOpenTelemetryTracing() {
settings.setEnableOpenTelemetryTracing(true);
}

private static String getFieldAsString(Object field) {
return field == null ? "null" : field.toString();
}

private Attributes otelAttributesFrom(ReadSession readSession) {
return Attributes.builder()
.put("bq.storage.read_session.name", getFieldAsString(readSession.getName()))
.put(
"bq.storage.read_session.data_format_value",
getFieldAsString(readSession.getDataFormatValue()))
.put(
"bq.storage.read_session.serialized_size",
getFieldAsString(readSession.getSerializedSize()))
.put("bq.storage.read_session.table", getFieldAsString(readSession.getTable()))
.put("bq.storage.read_session.estimated_row_count", readSession.getEstimatedRowCount())
.put(
"bq.storage.read_session.estimated_total_bytes_scanned",
readSession.getEstimatedTotalBytesScanned())
.put(
"bq.storage.read_session.estimated_total_physical_bytes",
readSession.getEstimatedTotalPhysicalFileSize())
.put("bq.storage.read_session.streams_count", readSession.getStreamsCount())
.put("bq.storage.read_session.trace_id", getFieldAsString(readSession.getTraceId()))
.put("bq.storage.read_session.expire_time", getFieldAsString(readSession.getExpireTime()))
.build();
}

private Attributes otelAttributesFrom(SplitReadStreamRequest request) {
return Attributes.builder()
.put("bq.storage.split_read_stream_request.name", getFieldAsString(request.getName()))
.put(
"bq.storage.split_read_stream_request.serialized_size",
getFieldAsString(request.getSerializedSize()))
.put(
"bq.storage.split_read_stream_request.fraction",
getFieldAsString(request.getFraction()))
.build();
}
}
29 changes: 29 additions & 0 deletions ...querystorage/src/main/java/com/google/cloud/bigquery/storage/v1/BigQueryReadSettings.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import com.google.cloud.bigquery.storage.v1.stub.EnhancedBigQueryReadStubSettings;
import io.grpc.Metadata;
import io.grpc.Status;
import io.opentelemetry.api.trace.Tracer;
import java.io.IOException;
import java.util.List;

Expand Down Expand Up @@ -76,6 +77,8 @@ public static interface RetryAttemptListener {
}

private RetryAttemptListener readRowsRetryAttemptListener = null;
private boolean enableOpenTelemetryTracing = false;
private Tracer openTelemetryTracer = null;

/**
* If a non null readRowsRetryAttemptListener is provided, client will call onRetryAttempt
Expand All @@ -87,10 +90,29 @@ public void setReadRowsRetryAttemptListener(RetryAttemptListener readRowsRetryAt
this.readRowsRetryAttemptListener = readRowsRetryAttemptListener;
}

public void setEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing) {
this.enableOpenTelemetryTracing = enableOpenTelemetryTracing;
if (enableOpenTelemetryTracing) {
this.openTelemetryTracer =
Singletons.getOpenTelemetry()
.getTracerProvider()
.tracerBuilder("com.google.cloud.bigquery.storage.v1.read")
.build();
}
}

public RetryAttemptListener getReadRowsRetryAttemptListener() {
return readRowsRetryAttemptListener;
}

public boolean isOpenTelemetryEnabled() {
return this.enableOpenTelemetryTracing;
}

public Tracer getOpenTelemetryTracer() {
return this.openTelemetryTracer;
}

/** Returns the object with the settings used for calls to splitReadStream. */
public UnaryCallSettings<SplitReadStreamRequest, SplitReadStreamResponse>
splitReadStreamSettings() {
Expand Down Expand Up @@ -199,13 +221,19 @@ public Builder applyToAllUnaryMethods(
}

private RetryAttemptListener readRowsRetryAttemptListener = null;
private boolean enableOpenTelemetryTracing = false;

public Builder setReadRowsRetryAttemptListener(
RetryAttemptListener readRowsRetryAttemptListener) {
this.readRowsRetryAttemptListener = readRowsRetryAttemptListener;
return this;
}

public Builder setEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing) {
this.enableOpenTelemetryTracing = enableOpenTelemetryTracing;
return this;
}

/** Returns the builder for the settings used for calls to createReadSession. */
public UnaryCallSettings.Builder<CreateReadSessionRequest, ReadSession>
createReadSessionSettings() {
Expand All @@ -228,6 +256,7 @@ public Builder setReadRowsRetryAttemptListener(
public BigQueryReadSettings build() throws IOException {
BigQueryReadSettings settings = new BigQueryReadSettings(this);
settings.setReadRowsRetryAttemptListener(readRowsRetryAttemptListener);
settings.setEnableOpenTelemetryTracing(enableOpenTelemetryTracing);
return settings;
}
}
Expand Down
Loading
Back | FazBrowse Home | New Git URL