Skip to content
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

Jump to bottom

Generate only consumer span for sqs receive message #9652

Merged
merged 1 commit into from
Oct 18, 2023
Merged

Generate only consumer span for sqs receive message #9652

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
11 changes: 11 additions & 0 deletions ...ion-api/src/main/java/io/opentelemetry/instrumentation/api/instrumenter/Instrumenter.java
View file
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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
Open in desktop
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);
Copy link
Member

Choose a reason for hiding this comment

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

This is another good case for the suppressSpan flag mentioned in open-telemetry/opentelemetry-java#5886
For not this hack is probably fine though

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
Loading