Skip to content

Commit 050b164

Browse files
surbhigarg92surbhigarg92
committed
Refactor
1 parent fe6e4a5 commit 050b164

28 files changed

+651
-779
lines changed

‎google-cloud-spanner/pom.xml

+27-25
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616
<properties>
1717
<site.installationModule>google-cloud-spanner</site.installationModule>
1818
<opencensus.version>0.31.1</opencensus.version>
19-
<opentelemetry.version>1.28.0</opentelemetry.version>
19+
<opentelemetry.version>1.30.1</opentelemetry.version>
2020
<graalvm.version>22.3.3</graalvm.version>
2121
<spanner.testenv.config.class>com.google.cloud.spanner.GceTestEnvConfig</spanner.testenv.config.class>
2222
<spanner.testenv.instance>projects/gcloud-devel/instances/spanner-testing-east1</spanner.testenv.instance>
@@ -249,30 +249,6 @@
249249
<artifactId>opentelemetry-context</artifactId>
250250
<version>${opentelemetry.version}</version>
251251
</dependency>
252-
<dependency>
253-
<groupId>io.opentelemetry</groupId>
254-
<artifactId>opentelemetry-sdk</artifactId>
255-
<version>${opentelemetry.version}</version>
256-
<scope>test</scope>
257-
</dependency>
258-
<dependency>
259-
<groupId>io.opentelemetry</groupId>
260-
<artifactId>opentelemetry-sdk-metrics</artifactId>
261-
<version>${opentelemetry.version}</version>
262-
<scope>test</scope>
263-
</dependency>
264-
<dependency>
265-
<groupId>io.opentelemetry</groupId>
266-
<artifactId>opentelemetry-sdk-trace</artifactId>
267-
<version>${opentelemetry.version}</version>
268-
<scope>test</scope>
269-
</dependency>
270-
<dependency>
271-
<groupId>io.opentelemetry</groupId>
272-
<artifactId>opentelemetry-sdk-testing</artifactId>
273-
<version>${opentelemetry.version}</version>
274-
<scope>test</scope>
275-
</dependency>
276252
<dependency>
277253
<groupId>com.google.auth</groupId>
278254
<artifactId>google-auth-library-oauth2-http</artifactId>
@@ -434,6 +410,32 @@
434410
<scope>test</scope>
435411
</dependency>
436412

413+
<!-- OpenTelemetry test dependencies -->
414+
<dependency>
415+
<groupId>io.opentelemetry</groupId>
416+
<artifactId>opentelemetry-sdk</artifactId>
417+
<version>${opentelemetry.version}</version>
418+
<scope>test</scope>
419+
</dependency>
420+
<dependency>
421+
<groupId>io.opentelemetry</groupId>
422+
<artifactId>opentelemetry-sdk-metrics</artifactId>
423+
<version>${opentelemetry.version}</version>
424+
<scope>test</scope>
425+
</dependency>
426+
<dependency>
427+
<groupId>io.opentelemetry</groupId>
428+
<artifactId>opentelemetry-sdk-trace</artifactId>
429+
<version>${opentelemetry.version}</version>
430+
<scope>test</scope>
431+
</dependency>
432+
<dependency>
433+
<groupId>io.opentelemetry</groupId>
434+
<artifactId>opentelemetry-sdk-testing</artifactId>
435+
<version>${opentelemetry.version}</version>
436+
<scope>test</scope>
437+
</dependency>
438+
437439
</dependencies>
438440

439441
<profiles>

‎google-cloud-spanner/src/main/java/com/google/cloud/spanner/AbstractReadContext.java

+23-29
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,10 @@
3737
import com.google.cloud.spanner.Options.ReadOption;
3838
import com.google.cloud.spanner.SessionImpl.SessionTransaction;
3939
import com.google.cloud.spanner.spi.v1.SpannerRpc;
40+
import com.google.cloud.spanner.tracing.DualSpan;
41+
import com.google.cloud.spanner.tracing.ISpan;
4042
import com.google.common.annotations.VisibleForTesting;
43+
import com.google.common.collect.ImmutableMap;
4144
import com.google.common.util.concurrent.MoreExecutors;
4245
import com.google.protobuf.ByteString;
4346
import com.google.spanner.v1.BeginTransactionRequest;
@@ -69,9 +72,10 @@ abstract class AbstractReadContext
6972
abstract static class Builder<B extends Builder<?, T>, T extends AbstractReadContext> {
7073
private SessionImpl session;
7174
private SpannerRpc rpc;
72-
private io.opentelemetry.api.trace.Span openTelemetrySpan =
73-
io.opentelemetry.api.trace.Span.fromContext(Context.current());
74-
private Span span = Tracing.getTracer().getCurrentSpan();
75+
private ISpan span =
76+
new DualSpan(
77+
Tracing.getTracer().getCurrentSpan(),
78+
io.opentelemetry.api.trace.Span.fromContext(Context.current()));
7579
private int defaultPrefetchChunks = SpannerOptions.Builder.DEFAULT_PREFETCH_CHUNKS;
7680
private QueryOptions defaultQueryOptions = SpannerOptions.Builder.DEFAULT_QUERY_OPTIONS;
7781
private ExecutorProvider executorProvider;
@@ -93,16 +97,11 @@ B setRpc(SpannerRpc rpc) {
9397
return self();
9498
}
9599

96-
B setSpan(Span span) {
100+
B setSpan(ISpan span) {
97101
this.span = span;
98102
return self();
99103
}
100104

101-
B setOpenTelemetrySpan(io.opentelemetry.api.trace.Span span) {
102-
this.openTelemetrySpan = span;
103-
return self();
104-
}
105-
106105
B setDefaultPrefetchChunks(int defaultPrefetchChunks) {
107106
this.defaultPrefetchChunks = defaultPrefetchChunks;
108107
return self();
@@ -356,7 +355,6 @@ void initTransaction() {
356355
if (transactionId != null) {
357356
return;
358357
}
359-
OpenTelemetryTraceUtil.addEvent(openTelemetrySpan, "Creating Transaction");
360358
span.addAnnotation("Creating Transaction");
361359
try {
362360
TransactionOptions.Builder options = TransactionOptions.newBuilder();
@@ -383,16 +381,16 @@ void initTransaction() {
383381
ErrorCode.INTERNAL, "Bad value in transaction.read_timestamp metadata field", e);
384382
}
385383
transactionId = transaction.getId();
386-
OpenTelemetryTraceUtil.addEvent(
387-
openTelemetrySpan,
388-
"Transaction Creation Done",
389-
OpenTelemetryTraceUtil.getTransactionAnnotations(transaction));
390384
span.addAnnotation(
391-
"Transaction Creation Done", TraceUtil.getTransactionAnnotations(transaction));
385+
"Transaction Creation Done",
386+
ImmutableMap.of(
387+
"id",
388+
transaction.getId().toStringUtf8(),
389+
"Timestamp",
390+
Timestamp.fromProto(transaction.getReadTimestamp()).toString()));
391+
392392
} catch (SpannerException e) {
393-
OpenTelemetryTraceUtil.addEventWithException(
394-
openTelemetrySpan, "Transaction Creation Failed", e);
395-
span.addAnnotation("Transaction Creation Failed", TraceUtil.getExceptionAnnotations(e));
393+
span.addAnnotation("Transaction Creation Failed", e);
396394
throw e;
397395
}
398396
}
@@ -403,8 +401,7 @@ void initTransaction() {
403401
final SessionImpl session;
404402
final SpannerRpc rpc;
405403
final ExecutorProvider executorProvider;
406-
Span span;
407-
io.opentelemetry.api.trace.Span openTelemetrySpan;
404+
ISpan span;
408405
private final int defaultPrefetchChunks;
409406
private final QueryOptions defaultQueryOptions;
410407

@@ -431,19 +428,19 @@ void initTransaction() {
431428
this.defaultPrefetchChunks = builder.defaultPrefetchChunks;
432429
this.defaultQueryOptions = builder.defaultQueryOptions;
433430
this.span = builder.span;
434-
this.openTelemetrySpan = builder.openTelemetrySpan;
435431
this.executorProvider = builder.executorProvider;
436432
}
437433

438434
@Override
439-
public void setSpan(Span span) {
435+
public void setSpan(ISpan span) {
440436
this.span = span;
441437
}
442438

439+
/**
440+
* No-op method needed to implement SessionTransaction interface.
441+
*/
443442
@Override
444-
public void setOpenTelemetrySpan(io.opentelemetry.api.trace.Span span) {
445-
this.openTelemetrySpan = span;
446-
}
443+
public void setSpan(Span span) {}
447444

448445
long getSeqNo() {
449446
return seqNo.incrementAndGet();
@@ -690,7 +687,6 @@ ResultSet executeQueryInternalWithOptions(
690687
MAX_BUFFERED_CHUNKS,
691688
SpannerImpl.QUERY,
692689
span,
693-
openTelemetrySpan,
694690
rpc.getExecuteQueryRetrySettings(),
695691
rpc.getExecuteQueryRetryableCodes()) {
696692
@Override
@@ -750,8 +746,7 @@ public final void invalidate() {
750746

751747
@Override
752748
public void close() {
753-
OpenTelemetryTraceUtil.endSpan(openTelemetrySpan);
754-
span.end(TraceUtil.END_SPAN_OPTIONS);
749+
span.end();
755750
synchronized (lock) {
756751
isClosed = true;
757752
}
@@ -831,7 +826,6 @@ ResultSet readInternalWithOptions(
831826
MAX_BUFFERED_CHUNKS,
832827
SpannerImpl.READ,
833828
span,
834-
openTelemetrySpan,
835829
rpc.getReadRetrySettings(),
836830
rpc.getReadRetryableCodes()) {
837831
@Override

‎google-cloud-spanner/src/main/java/com/google/cloud/spanner/AbstractResultSet.java

+20-56
Original file line numberDiff line numberDiff line change
@@ -33,11 +33,13 @@
3333
import com.google.cloud.Timestamp;
3434
import com.google.cloud.spanner.Type.StructField;
3535
import com.google.cloud.spanner.spi.v1.SpannerRpc;
36+
import com.google.cloud.spanner.tracing.IScope;
37+
import com.google.cloud.spanner.tracing.ISpan;
38+
import com.google.cloud.spanner.tracing.TraceWrapper;
3639
import com.google.cloud.spanner.v1.stub.SpannerStubSettings;
3740
import com.google.common.annotations.VisibleForTesting;
3841
import com.google.common.base.Preconditions;
3942
import com.google.common.collect.AbstractIterator;
40-
import com.google.common.collect.ImmutableMap;
4143
import com.google.common.collect.Lists;
4244
import com.google.common.util.concurrent.Uninterruptibles;
4345
import com.google.protobuf.ByteString;
@@ -50,12 +52,7 @@
5052
import com.google.spanner.v1.Transaction;
5153
import com.google.spanner.v1.TypeCode;
5254
import io.grpc.Context;
53-
import io.opencensus.common.Scope;
54-
import io.opencensus.trace.AttributeValue;
55-
import io.opencensus.trace.Span;
56-
import io.opencensus.trace.Tracer;
5755
import io.opencensus.trace.Tracing;
58-
import io.opentelemetry.api.common.Attributes;
5956
import java.io.IOException;
6057
import java.io.Serializable;
6158
import java.math.BigDecimal;
@@ -83,7 +80,7 @@
8380

8481
/** Implementation of {@link ResultSet}. */
8582
abstract class AbstractResultSet<R> extends AbstractStructReader implements ResultSet {
86-
private static final Tracer tracer = Tracing.getTracer();
83+
private static final TraceWrapper tracer = new TraceWrapper(Tracing.getTracer());
8784
private static final com.google.protobuf.Value NULL_VALUE =
8885
com.google.protobuf.Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build();
8986

@@ -1094,8 +1091,7 @@ abstract static class ResumableStreamIterator extends AbstractIterator<PartialRe
10941091
private final BackOff backOff;
10951092
private final LinkedList<PartialResultSet> buffer = new LinkedList<>();
10961093
private final int maxBufferSize;
1097-
private final Span span;
1098-
private final io.opentelemetry.api.trace.Span openTelemetrySpan;
1094+
private final ISpan span;
10991095
private CloseableIterator<PartialResultSet> stream;
11001096
private ByteString resumeToken;
11011097
private boolean finished;
@@ -1109,16 +1105,12 @@ abstract static class ResumableStreamIterator extends AbstractIterator<PartialRe
11091105
protected ResumableStreamIterator(
11101106
int maxBufferSize,
11111107
String streamName,
1112-
Span parent,
1113-
io.opentelemetry.api.trace.Span openTelemetryParent,
1108+
ISpan parent,
11141109
RetrySettings streamingRetrySettings,
11151110
Set<Code> retryableCodes) {
11161111
checkArgument(maxBufferSize >= 0);
11171112
this.maxBufferSize = maxBufferSize;
1118-
this.openTelemetrySpan =
1119-
OpenTelemetryTraceUtil.spanBuilderWithExplicitParent(
1120-
SpannerOptions.getTracer(), streamName, openTelemetryParent);
1121-
this.span = tracer.spanBuilderWithExplicitParent(streamName, parent).startSpan();
1113+
this.span = tracer.spanBuilderWithExplicitParent(streamName, parent);
11221114
this.streamingRetrySettings = Preconditions.checkNotNull(streamingRetrySettings);
11231115
this.retryableCodes = Preconditions.checkNotNull(retryableCodes);
11241116
this.backOff = newBackOff();
@@ -1174,15 +1166,7 @@ private static long nextBackOffMillis(BackOff backoff) throws SpannerException {
11741166
}
11751167

11761168
private void backoffSleep(Context context, long backoffMillis) throws SpannerException {
1177-
tracer
1178-
.getCurrentSpan()
1179-
.addAnnotation(
1180-
"Backing off",
1181-
ImmutableMap.of("Delay", AttributeValue.longAttributeValue(backoffMillis)));
1182-
OpenTelemetryTraceUtil.addEvent(
1183-
io.opentelemetry.api.trace.Span.fromContext(io.opentelemetry.context.Context.current()),
1184-
"Backing off",
1185-
Attributes.builder().put("Delay", backoffMillis).build());
1169+
tracer.getCurrentSpan().addAnnotation("Backing off", "Delay", backoffMillis);
11861170
final CountDownLatch latch = new CountDownLatch(1);
11871171
final Context.CancellationListener listener =
11881172
ignored -> {
@@ -1222,8 +1206,7 @@ public void execute(Runnable command) {
12221206
public void close(@Nullable String message) {
12231207
if (stream != null) {
12241208
stream.close(message);
1225-
span.end(TraceUtil.END_SPAN_OPTIONS);
1226-
OpenTelemetryTraceUtil.endSpan(openTelemetrySpan);
1209+
span.end();
12271210
stream = null;
12281211
}
12291212
}
@@ -1239,20 +1222,12 @@ protected PartialResultSet computeNext() {
12391222
while (true) {
12401223
// Eagerly start stream before consuming any buffered items.
12411224
if (stream == null) {
1242-
OpenTelemetryTraceUtil.addEvent(
1243-
openTelemetrySpan,
1244-
"Starting/Resuming stream",
1245-
Attributes.builder()
1246-
.put("ResumeToken", resumeToken == null ? "null" : resumeToken.toStringUtf8())
1247-
.build());
12481225
span.addAnnotation(
12491226
"Starting/Resuming stream",
1250-
ImmutableMap.of(
1251-
"ResumeToken",
1252-
AttributeValue.stringAttributeValue(
1253-
resumeToken == null ? "null" : resumeToken.toStringUtf8())));
1254-
try (Scope s = tracer.withSpan(span);
1255-
io.opentelemetry.context.Scope ss = openTelemetrySpan.makeCurrent()) {
1227+
"ResumeToken",
1228+
resumeToken == null ? "null" : resumeToken.toStringUtf8());
1229+
1230+
try (IScope sss = tracer.withSpan(span)) {
12561231
// When start a new stream set the Span as current to make the gRPC Span a child of
12571232
// this Span.
12581233
stream = checkNotNull(startStream(resumeToken));
@@ -1292,22 +1267,16 @@ protected PartialResultSet computeNext() {
12921267
}
12931268
} catch (SpannerException spannerException) {
12941269
if (safeToRetry && isRetryable(spannerException)) {
1295-
OpenTelemetryTraceUtil.addEvent(
1296-
openTelemetrySpan,
1297-
"Stream broken. Safe to retry",
1298-
OpenTelemetryTraceUtil.getExceptionAnnotations(spannerException));
1299-
span.addAnnotation(
1300-
"Stream broken. Safe to retry",
1301-
TraceUtil.getExceptionAnnotations(spannerException));
1270+
1271+
span.addAnnotation("Stream broken. Safe to retry", spannerException);
13021272
logger.log(Level.FINE, "Retryable exception, will sleep and retry", spannerException);
13031273
// Truncate any items in the buffer before the last retry token.
13041274
while (!buffer.isEmpty() && buffer.getLast().getResumeToken().isEmpty()) {
13051275
buffer.removeLast();
13061276
}
13071277
assert buffer.isEmpty() || buffer.getLast().getResumeToken().equals(resumeToken);
13081278
stream = null;
1309-
try (Scope s = tracer.withSpan(span);
1310-
io.opentelemetry.context.Scope ss = openTelemetrySpan.makeCurrent()) {
1279+
try (IScope s = tracer.withSpan(span)) {
13111280
long delay = spannerException.getRetryDelayInMillis();
13121281
if (delay != -1) {
13131282
backoffSleep(context, delay);
@@ -1318,17 +1287,12 @@ protected PartialResultSet computeNext() {
13181287

13191288
continue;
13201289
}
1321-
span.addAnnotation("Stream broken. Not safe to retry");
1322-
TraceUtil.setWithFailure(span, spannerException);
1323-
OpenTelemetryTraceUtil.addEventWithExceptionAndSetFailure(
1324-
openTelemetrySpan, "Stream broken. Safe to retry", spannerException);
1290+
span.addAnnotation("Stream broken. Not safe to retry", spannerException);
1291+
span.setStatus(spannerException);
13251292
throw spannerException;
13261293
} catch (RuntimeException e) {
1327-
span.addAnnotation("Stream broken. Not safe to retry");
1328-
TraceUtil.setWithFailure(span, e);
1329-
OpenTelemetryTraceUtil.addEventWithExceptionAndSetFailure(
1330-
openTelemetrySpan, "Stream broken. Safe to retry", e);
1331-
1294+
span.addAnnotation("Stream broken. Not safe to retry", e);
1295+
span.setStatus(e);
13321296
throw e;
13331297
}
13341298
}

0 commit comments

Comments
 (0)