Skip to content

Commit

Permalink
Generate only consumer span for sqs receive message (#9652)
Browse files Browse the repository at this point in the history
  • Loading branch information
@laurit
laurit committed Oct 18, 2023
1 parent 76d0090 commit 3dbc301
Show file tree
Hide file tree
Showing 10 changed files with 275 additions and 230 deletions.
11 changes: 11 additions & 0 deletions ...ion-api/src/main/java/io/opentelemetry/instrumentation/api/instrumenter/Instrumenter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,17 @@ public <RQ, RS> Context startAndEnd(
return instrumenter.startAndEnd(
parentContext, request, response, error, startTime, endTime);
}

@Override
public <REQUEST, RESPONSE> Context suppressSpan(
Instrumenter<REQUEST, RESPONSE> instrumenter,
Context parentContext,
REQUEST request) {
SpanKind spanKind = instrumenter.spanKindExtractor.extract(request);

return instrumenter.spanSuppressor.storeInContext(
parentContext, spanKind, Span.getInvalid());
}
});
}
}
3 changes: 3 additions & 0 deletions ...n-api/src/main/java/io/opentelemetry/instrumentation/api/internal/InstrumenterAccess.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,4 +24,7 @@ <REQUEST, RESPONSE> Context startAndEnd(
@Nullable Throwable error,
Instant startTime,
Instant endTime);

<REQUEST, RESPONSE> Context suppressSpan(
Instrumenter<REQUEST, RESPONSE> instrumenter, Context parentContext, REQUEST request);
}
5 changes: 5 additions & 0 deletions ...ion-api/src/main/java/io/opentelemetry/instrumentation/api/internal/InstrumenterUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,11 @@ public static <REQUEST, RESPONSE> Context startAndEnd(
instrumenter, parentContext, request, response, error, startTime, endTime);
}

public static <REQUEST, RESPONSE> Context suppressSpan(
Instrumenter<REQUEST, RESPONSE> instrumenter, Context parentContext, REQUEST request) {
return instrumenterAccess.suppressSpan(instrumenter, parentContext, request);
}

public static <REQUEST, RESPONSE> Instrumenter<REQUEST, RESPONSE> buildUpstreamInstrumenter(
InstrumenterBuilder<REQUEST, RESPONSE> builder,
TextMapGetter<REQUEST> getter,
Expand Down
130 changes: 12 additions & 118 deletions instrumentation/aws-sdk/aws-sdk-1.11/javaagent/src/test/groovy/S3TracingTest.groovy
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
awsConnector.purgeQueue(queueUrl)

then:
assertTraces(12) {
assertTraces(10) {
trace(0, 1) {

span(0) {
Expand Down Expand Up @@ -168,31 +168,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(5, 1) {
span(0) {
name "SQS.ReceiveMessage"
kind CLIENT
hasNoParent()
attributes {
"aws.agent" "java-aws-sdk"
"aws.endpoint" String
"rpc.method" "ReceiveMessage"
"aws.queue.url" queueUrl
"rpc.system" "aws-api"
"rpc.service" "AmazonSQS"
"http.method" "POST"
"http.status_code" 200
"http.url" String
"net.peer.name" String
"$SemanticAttributes.NET_PROTOCOL_NAME" "http"
"$SemanticAttributes.NET_PROTOCOL_VERSION" "1.1"
"net.peer.port" { it == null || Number }
"$SemanticAttributes.HTTP_REQUEST_CONTENT_LENGTH" { it == null || it instanceof Long }
"$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" { it == null || it instanceof Long }
}
}
}
trace(6, 2) {
trace(5, 2) {
span(0) {
name "S3.PutObject"
kind CLIENT
Expand Down Expand Up @@ -239,36 +215,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}

/**
* This span represents HTTP "sending of receive message" operation. It's always single, while there can be multiple CONSUMER spans (one per consumed message).
* This one could be suppressed (by IF in TracingRequestHandler#beforeRequest but then HTTP instrumentation span would appear
*/
trace(7, 1) {
span(0) {
name "SQS.ReceiveMessage"
kind CLIENT
hasNoParent()
attributes {
"aws.agent" "java-aws-sdk"
"aws.endpoint" String
"rpc.method" "ReceiveMessage"
"aws.queue.url" queueUrl
"rpc.system" "aws-api"
"rpc.service" "AmazonSQS"
"http.method" "POST"
"http.status_code" 200
"http.url" String
"net.peer.name" String
"$SemanticAttributes.NET_PROTOCOL_NAME" "http"
"$SemanticAttributes.NET_PROTOCOL_VERSION" "1.1"
"net.peer.port" { it == null || Number }
"$SemanticAttributes.HTTP_REQUEST_CONTENT_LENGTH" { it == null || it instanceof Long }
"$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" { it == null || it instanceof Long }
}
}
}
trace(8, 1) {
trace(6, 1) {
span(0) {
name "S3.ListObjects"
kind CLIENT
Expand All @@ -292,7 +239,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(9, 1) {
trace(7, 1) {
span(0) {
name "S3.DeleteObject"
kind CLIENT
Expand All @@ -316,7 +263,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(10, 1) {
trace(8, 1) {
span(0) {
name "S3.DeleteBucket"
kind CLIENT
Expand All @@ -340,7 +287,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(11, 1) {
trace(9, 1) {
span(0) {
name "SQS.PurgeQueue"
kind CLIENT
Expand Down Expand Up @@ -393,7 +340,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
awsConnector.purgeQueue(queueUrl)

then:
assertTraces(16) {
assertTraces(14) {
trace(0, 1) {
span(0) {
name "SQS.CreateQueue"
Expand Down Expand Up @@ -583,32 +530,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
// test even receive
trace(8, 1) {
span(0) {
name "SQS.ReceiveMessage"
kind CLIENT
hasNoParent()
attributes {
"aws.agent" "java-aws-sdk"
"aws.endpoint" String
"rpc.method" "ReceiveMessage"
"aws.queue.url" queueUrl
"rpc.system" "aws-api"
"rpc.service" "AmazonSQS"
"http.method" "POST"
"http.status_code" 200
"http.url" String
"net.peer.name" String
"$SemanticAttributes.NET_PROTOCOL_NAME" "http"
"$SemanticAttributes.NET_PROTOCOL_VERSION" "1.1"
"net.peer.port" { it == null || Number }
"$SemanticAttributes.HTTP_REQUEST_CONTENT_LENGTH" { it == null || it instanceof Long }
"$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" { it == null || it instanceof Long }
}
}
}
trace(9, 1) {
span(0) {
name "S3.PutObject"
kind CLIENT
Expand All @@ -632,35 +554,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
/**
* This span represents HTTP "sending of receive message" operation. It's always single, while there can be multiple CONSUMER spans (one per consumed message).
* This one could be suppressed (by IF in TracingRequestHandler#beforeRequest but then HTTP instrumentation span would appear
*/
trace(10, 1) {
span(0) {
name "SQS.ReceiveMessage"
kind CLIENT
hasNoParent()
attributes {
"aws.agent" "java-aws-sdk"
"aws.endpoint" String
"rpc.method" "ReceiveMessage"
"aws.queue.url" queueUrl
"rpc.system" "aws-api"
"rpc.service" "AmazonSQS"
"http.method" "POST"
"http.status_code" 200
"http.url" String
"net.peer.name" String
"$SemanticAttributes.NET_PROTOCOL_NAME" "http"
"$SemanticAttributes.NET_PROTOCOL_VERSION" "1.1"
"net.peer.port" { it == null || Number }
"$SemanticAttributes.HTTP_REQUEST_CONTENT_LENGTH" { it == null || it instanceof Long }
"$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" { it == null || it instanceof Long }
}
}
}
trace(11, 1) {
trace(9, 1) {
span(0) {
name "SQS.ReceiveMessage"
kind CONSUMER
Expand All @@ -685,7 +579,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(12, 1) {
trace(10, 1) {
span(0) {
name "S3.ListObjects"
kind CLIENT
Expand All @@ -709,7 +603,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(13, 1) {
trace(11, 1) {
span(0) {
name "S3.DeleteObject"
kind CLIENT
Expand All @@ -733,7 +627,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(14, 1) {
trace(12, 1) {
span(0) {
name "S3.DeleteBucket"
kind CLIENT
Expand All @@ -757,7 +651,7 @@ class S3TracingTest extends AgentInstrumentationSpecification {
}
}
}
trace(15, 1) {
trace(13, 1) {
span(0) {
name "SQS.PurgeQueue"
kind CLIENT
Expand Down
29 changes: 1 addition & 28 deletions instrumentation/aws-sdk/aws-sdk-1.11/javaagent/src/test/groovy/SnsTracingTest.groovy
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ class SnsTracingTest extends AgentInstrumentationSpecification {
awsConnector.receiveMessage(queueUrl)

then:
assertTraces(7) {
assertTraces(6) {
trace(0, 1) {

span(0) {
Expand Down Expand Up @@ -198,33 +198,6 @@ class SnsTracingTest extends AgentInstrumentationSpecification {
}
}
}
/**
* This span represents HTTP "sending of receive message" operation. It's always single, while there can be multiple CONSUMER spans (one per consumed message).
* This one could be suppressed (by IF in TracingRequestHandler#beforeRequest but then HTTP instrumentation span would appear
*/
trace(6, 1) {
span(0) {
name "SQS.ReceiveMessage"
kind CLIENT
hasNoParent()
attributes {
"aws.agent" "java-aws-sdk"
"aws.endpoint" String
"aws.queue.url" queueUrl
"rpc.system" "aws-api"
"rpc.service" "AmazonSQS"
"rpc.method" "ReceiveMessage"
"http.method" "POST"
"http.status_code" 200
"http.url" String
"net.peer.name" String
"$SemanticAttributes.NET_PROTOCOL_NAME" "http"
"$SemanticAttributes.NET_PROTOCOL_VERSION" "1.1"
"net.peer.port" { it == null || Number }
"$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" Long
}
}
}
}
}
}
41 changes: 41 additions & 0 deletions ...ry/src/main/java/io/opentelemetry/instrumentation/awssdk/v1_11/TracingRequestHandler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,22 @@
import com.amazonaws.handlers.RequestHandler2;
import com.google.errorprone.annotations.CanIgnoreReturnValue;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.ContextKey;
import io.opentelemetry.contrib.awsxray.propagator.AwsXrayPropagator;
import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
import io.opentelemetry.instrumentation.api.internal.InstrumenterUtil;
import java.time.Instant;
import javax.annotation.Nullable;

/** Tracing Request Handler. */
final class TracingRequestHandler extends RequestHandler2 {

static final HandlerContextKey<Context> CONTEXT =
new HandlerContextKey<>(Context.class.getName());
private static final ContextKey<Instant> REQUEST_START_KEY =
ContextKey.named(TracingRequestHandler.class.getName() + ".RequestStart");
private static final ContextKey<Context> PARENT_CONTEXT_KEY =
ContextKey.named(TracingRequestHandler.class.getName() + ".ParentContext");

private final Instrumenter<Request<?>, Response<?>> requestInstrumenter;
private final Instrumenter<Request<?>, Response<?>> consumerInstrumenter;
Expand All @@ -47,6 +54,21 @@ public void beforeRequest(Request<?> request) {
if (!requestInstrumenter.shouldStart(parentContext, request)) {
return;
}

// Skip creating request span for AmazonSQSClient.receiveMessage if there is no parent span and
// also suppress the span from the underlying http client. Request/http client span appears in a
// separate trace from message producer/consumer spans if there is no parent span just having
// a trace with only the request/http client span isn't useful.
if (Context.root() == parentContext
&& "com.amazonaws.services.sqs.model.ReceiveMessageRequest"
.equals(request.getOriginalRequest().getClass().getName())) {
Context context = InstrumenterUtil.suppressSpan(requestInstrumenter, parentContext, request);
context = context.with(REQUEST_START_KEY, Instant.now());
context = context.with(PARENT_CONTEXT_KEY, parentContext);
request.addHandlerContext(CONTEXT, context);
return;
}

Context context = requestInstrumenter.start(parentContext, request);

AwsXrayPropagator.getInstance().inject(context, request, HeaderSetter.INSTANCE);
Expand Down Expand Up @@ -81,6 +103,25 @@ private void finish(Request<?> request, Response<?> response, @Nullable Throwabl
return;
}
request.addHandlerContext(CONTEXT, null);

// see beforeRequest, requestStart is only set when we skip creating request span for sqs
// AmazonSQSClient.receiveMessage calls
Instant requestStart = context.get(REQUEST_START_KEY);
if (requestStart != null) {
// create request span if there was an error
if (error != null) {
InstrumenterUtil.startAndEnd(
requestInstrumenter,
context.get(PARENT_CONTEXT_KEY),
request,
response,
error,
requestStart,
Instant.now());
}
return;
}

requestInstrumenter.end(context, request, response, error);
}
}
Loading

0 comments on commit 3dbc301

Please sign in to comment.