Skip to content

Commit bac10e9

Browse files
PerfectSlayerPerfectSlayer
committed
feat(dsm): Clean up API
Remove DSM injection method from AgentPropagation Make checkpoint use DataStreamsContext
1 parent 1521ae2 commit bac10e9

File tree

34 files changed

+185
-472
lines changed

34 files changed

+185
-472
lines changed

‎dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/decorator/HttpServerDecorator.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package datadog.trace.bootstrap.instrumentation.decorator;
22

33
import static datadog.trace.api.cache.RadixTreeCache.UNSET_STATUS;
4+
import static datadog.trace.api.datastreams.DataStreamsContext.fromTags;
45
import static datadog.trace.api.gateway.Events.EVENTS;
56
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.traceConfig;
67
import static datadog.trace.bootstrap.instrumentation.decorator.http.HttpResourceDecorator.HTTP_RESOURCE_DECORATOR;
@@ -148,7 +149,7 @@ public AgentSpan startSpan(
148149
}
149150
AgentPropagation.ContextVisitor<REQUEST_CARRIER> getter = getter();
150151
if (null != carrier && null != getter) {
151-
tracer().getDataStreamsMonitoring().setCheckpoint(span, SERVER_PATHWAY_EDGE_TAGS, 0, 0);
152+
tracer().getDataStreamsMonitoring().setCheckpoint(span, fromTags(SERVER_PATHWAY_EDGE_TAGS));
152153
}
153154
return span;
154155
}

‎dd-java-agent/instrumentation/armeria-grpc/src/main/java/datadog/trace/instrumentation/armeria/grpc/server/TracingServerInterceptor.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package datadog.trace.instrumentation.armeria.grpc.server;
22

3+
import static datadog.trace.api.datastreams.DataStreamsContext.fromTags;
34
import static datadog.trace.api.gateway.Events.EVENTS;
45
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan;
56
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.propagate;
@@ -71,7 +72,7 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
7172

7273
AgentTracer.get()
7374
.getDataStreamsMonitoring()
74-
.setCheckpoint(span, SERVER_PATHWAY_EDGE_TAGS, 0, 0);
75+
.setCheckpoint(span, fromTags(SERVER_PATHWAY_EDGE_TAGS));
7576

7677
RequestContext reqContext = span.getRequestContext();
7778
if (reqContext != null) {

‎dd-java-agent/instrumentation/aws-java-sdk-1.11.0/src/main/java/datadog/trace/instrumentation/aws/v0/AwsSdkClientDecorator.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package datadog.trace.instrumentation.aws.v0;
22

3+
import static datadog.trace.api.datastreams.DataStreamsContext.create;
34
import static datadog.trace.bootstrap.instrumentation.api.ResourceNamePriorities.RPC_COMMAND_NAME;
45
import static datadog.trace.core.datastreams.TagsProcessor.DIRECTION_OUT;
56

@@ -264,7 +265,7 @@ public AgentSpan onServiceResponse(
264265

265266
AgentTracer.get()
266267
.getDataStreamsMonitoring()
267-
.setCheckpoint(span, sortedTags, 0, responseSize);
268+
.setCheckpoint(span, create(sortedTags, 0, responseSize));
268269
}
269270

270271
if ("PutObjectRequest".equalsIgnoreCase(awsOperation)
@@ -285,7 +286,7 @@ public AgentSpan onServiceResponse(
285286

286287
AgentTracer.get()
287288
.getDataStreamsMonitoring()
288-
.setCheckpoint(span, sortedTags, 0, payloadSize);
289+
.setCheckpoint(span, create(sortedTags, 0, payloadSize));
289290
}
290291
}
291292
}

‎dd-java-agent/instrumentation/aws-java-sdk-1.11.0/src/main/java/datadog/trace/instrumentation/aws/v0/TracingRequestHandler.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package datadog.trace.instrumentation.aws.v0;
22

3+
import static datadog.trace.api.datastreams.DataStreamsContext.create;
34
import static datadog.trace.bootstrap.instrumentation.api.AgentPropagation.XRAY_TRACING_CONCERN;
45
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan;
56
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.blackholeSpan;
@@ -19,6 +20,7 @@
1920
import datadog.context.propagation.Propagators;
2021
import datadog.trace.api.Config;
2122
import datadog.trace.api.datastreams.AgentDataStreamsMonitoring;
23+
import datadog.trace.api.datastreams.DataStreamsContext;
2224
import datadog.trace.api.datastreams.PathwayContext;
2325
import datadog.trace.bootstrap.ContextStore;
2426
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
@@ -122,8 +124,8 @@ public void afterResponse(final Request<?> request, final Response<?> response)
122124
AgentDataStreamsMonitoring dataStreamsMonitoring =
123125
AgentTracer.get().getDataStreamsMonitoring();
124126
PathwayContext pathwayContext = dataStreamsMonitoring.newPathwayContext();
125-
pathwayContext.setCheckpoint(
126-
sortedTags, dataStreamsMonitoring::add, arrivalTime.getTime());
127+
DataStreamsContext context = create(sortedTags, arrivalTime.getTime(), 0);
128+
pathwayContext.setCheckpoint(context, dataStreamsMonitoring::add);
127129
if (!span.context().getPathwayContext().isStarted()) {
128130
span.context().mergePathwayContext(pathwayContext);
129131
}

‎dd-java-agent/instrumentation/aws-java-sdk-2.2/src/main/java/datadog/trace/instrumentation/aws/v2/AwsSdkClientDecorator.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package datadog.trace.instrumentation.aws.v2;
22

3+
import static datadog.trace.api.datastreams.DataStreamsContext.create;
34
import static datadog.trace.core.datastreams.TagsProcessor.DIRECTION_IN;
45
import static datadog.trace.core.datastreams.TagsProcessor.DIRECTION_OUT;
56
import static datadog.trace.core.datastreams.TagsProcessor.DIRECTION_TAG;
@@ -361,7 +362,8 @@ public AgentSpan onSdkResponse(
361362
AgentTracer.get().getDataStreamsMonitoring();
362363
PathwayContext pathwayContext = dataStreamsMonitoring.newPathwayContext();
363364
pathwayContext.setCheckpoint(
364-
sortedTags, dataStreamsMonitoring::add, arrivalTime.toEpochMilli());
365+
create(sortedTags, arrivalTime.toEpochMilli(), 0),
366+
dataStreamsMonitoring::add);
365367
if (!span.context().getPathwayContext().isStarted()) {
366368
span.context().mergePathwayContext(pathwayContext);
367369
}
@@ -391,7 +393,7 @@ public AgentSpan onSdkResponse(
391393

392394
AgentTracer.get()
393395
.getDataStreamsMonitoring()
394-
.setCheckpoint(span, sortedTags, 0, responseSize);
396+
.setCheckpoint(span, create(sortedTags, 0, responseSize));
395397
}
396398

397399
if ("PutObject".equalsIgnoreCase(awsOperation)) {
@@ -411,7 +413,7 @@ public AgentSpan onSdkResponse(
411413

412414
AgentTracer.get()
413415
.getDataStreamsMonitoring()
414-
.setCheckpoint(span, sortedTags, 0, payloadSize);
416+
.setCheckpoint(span, create(sortedTags, 0, payloadSize));
415417
}
416418
}
417419
}

‎dd-java-agent/instrumentation/aws-java-sqs-1.0/src/main/java/datadog/trace/instrumentation/aws/v1/sqs/TracingIterator.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package datadog.trace.instrumentation.aws.v1.sqs;
22

3+
import static datadog.trace.api.datastreams.DataStreamsContext.create;
34
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateNext;
45
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.closePrevious;
56
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.propagate;
@@ -90,7 +91,7 @@ protected void startNewMessageSpan(Message message) {
9091
sortedTags.put(DIRECTION_TAG, DIRECTION_IN);
9192
sortedTags.put(TOPIC_TAG, urlFileName(queueUrl));
9293
sortedTags.put(TYPE_TAG, "sqs");
93-
AgentTracer.get().getDataStreamsMonitoring().setCheckpoint(span, sortedTags, 0, 0);
94+
AgentTracer.get().getDataStreamsMonitoring().setCheckpoint(span, create(sortedTags, 0, 0));
9495

9596
CONSUMER_DECORATE.afterStart(span);
9697
CONSUMER_DECORATE.onConsume(span, queueUrl);

‎dd-java-agent/instrumentation/aws-java-sqs-2.0/src/main/java/datadog/trace/instrumentation/aws/v2/sqs/TracingIterator.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package datadog.trace.instrumentation.aws.v2.sqs;
22

3+
import static datadog.trace.api.datastreams.DataStreamsContext.create;
34
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateNext;
45
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.closePrevious;
56
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.propagate;
@@ -92,7 +93,7 @@ protected void startNewMessageSpan(Message message) {
9293
sortedTags.put(DIRECTION_TAG, DIRECTION_IN);
9394
sortedTags.put(TOPIC_TAG, urlFileName(queueUrl));
9495
sortedTags.put(TYPE_TAG, "sqs");
95-
AgentTracer.get().getDataStreamsMonitoring().setCheckpoint(span, sortedTags, 0, 0);
96+
AgentTracer.get().getDataStreamsMonitoring().setCheckpoint(span, create(sortedTags, 0, 0));
9697

9798
CONSUMER_DECORATE.afterStart(span);
9899
CONSUMER_DECORATE.onConsume(span, queueUrl, requestId);

‎dd-java-agent/instrumentation/google-pubsub/src/main/java/datadog/trace/instrumentation/googlepubsub/PubSubDecorator.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import datadog.trace.api.Functions;
1414
import datadog.trace.api.cache.DDCache;
1515
import datadog.trace.api.cache.DDCaches;
16+
import datadog.trace.api.datastreams.DataStreamsContext;
1617
import datadog.trace.api.naming.SpanNaming;
1718
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
1819
import datadog.trace.bootstrap.instrumentation.api.AgentSpanContext;
@@ -141,9 +142,10 @@ public AgentSpan onConsume(final PubsubMessage message, final String subscriptio
141142
.getDataStreamsMonitoring()
142143
.setCheckpoint(
143144
span,
144-
sortedTags,
145-
publishTime.getSeconds() * 1_000 + publishTime.getNanos() / (int) 1e6,
146-
message.getSerializedSize());
145+
DataStreamsContext.create(
146+
sortedTags,
147+
publishTime.getSeconds() * 1_000 + publishTime.getNanos() / (int) 1e6,
148+
message.getSerializedSize()));
147149
afterStart(span);
148150
span.setResourceName(
149151
CONSUMER_RESOURCE_NAME_CACHE.computeIfAbsent(parsedSubscription, CONSUMER_PREFIX));

‎dd-java-agent/instrumentation/grpc-1.5/src/main/java/datadog/trace/instrumentation/grpc/server/TracingServerInterceptor.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package datadog.trace.instrumentation.grpc.server;
22

3+
import static datadog.trace.api.datastreams.DataStreamsContext.fromTags;
34
import static datadog.trace.api.gateway.Events.EVENTS;
45
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan;
56
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.propagate;
@@ -70,7 +71,7 @@ public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
7071

7172
AgentTracer.get()
7273
.getDataStreamsMonitoring()
73-
.setCheckpoint(span, SERVER_PATHWAY_EDGE_TAGS, 0, 0);
74+
.setCheckpoint(span, fromTags(SERVER_PATHWAY_EDGE_TAGS));
7475

7576
RequestContext reqContext = span.getRequestContext();
7677
if (reqContext != null) {

‎dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/TracingIterator.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package datadog.trace.instrumentation.kafka_clients;
22

3-
import static datadog.trace.api.datastreams.DataStreamsContext.fromKafka;
3+
import static datadog.trace.api.datastreams.DataStreamsContext.create;
44
import static datadog.trace.bootstrap.instrumentation.api.AgentPropagation.DSM_CONCERN;
55
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateNext;
66
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.closePrevious;
@@ -113,7 +113,7 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {
113113
if (STREAMING_CONTEXT.isDisabledForTopic(val.topic())) {
114114
AgentTracer.get()
115115
.getDataStreamsMonitoring()
116-
.setCheckpoint(span, sortedTags, val.timestamp(), payloadSize);
116+
.setCheckpoint(span, create(sortedTags, val.timestamp(), payloadSize));
117117
} else {
118118
// when we're in a streaming context we want to consume only from source topics
119119
if (STREAMING_CONTEXT.isSourceTopic(val.topic())) {
@@ -122,7 +122,7 @@ protected void startNewRecordSpan(ConsumerRecord<?, ?> val) {
122122
// some other instance of the application, breaking the context propagation
123123
// for DSM users
124124
Propagator dsmPropagator = Propagators.forConcern(DSM_CONCERN);
125-
DataStreamsContext dsmContext = fromKafka(sortedTags, val.timestamp(), payloadSize);
125+
DataStreamsContext dsmContext = create(sortedTags, val.timestamp(), payloadSize);
126126
dsmPropagator.inject(span.with(dsmContext), val.headers(), SETTER);
127127
}
128128
}

0 commit comments

Comments
 (0)