micronaut-tracing aws instrumentation sqs support is not working

[chenlin2001]

Expected Behavior

Propagation of traces to SQS messages is not working.
Using the following config:

otel:
  register:
    global: true
  javaagent:
    debug: true
  propagators: tracecontext,baggage
  traces:
    exporter: logging

  instrumentation:
    messaging:
      experimental:
        receive-telemetry:
          enabled: true
    aws-sdk:
      experimental-use-propagator-for-messaging: true
      experimental-span-attributes: true

I would expect to be able to pass the traceId to the message attributes. However, even though traces are availables in my logs:

INFO: 'test-sqs-queue-name publish' : 4868b868b771d46a7db7ad00b63feb8d f009ef0728c09e05 PRODUCER [tracer: io.opentelemetry.aws-sdk-2.2:1.33.5-alpha] AttributesMap{data={aws.queue.url=http://sqs.us-east-1.localhost:4566/000000000000/test-sqs-queue-name, http.url=http://127.0.0.1:35113/, messaging.destination.name=test-sqs-queue-name, messaging.operation=publish, http.request_content_length=208, rpc.system=aws-api, messaging.system=AmazonSQS, rpc.service=Sqs, net.peer.port=35113, messaging.message.id=f90b76cb-f3f3-4d66-8aa5-6f6ddb424a01, http.method=POST, net.peer.name=127.0.0.1, http.status_code=200, rpc.method=SendMessage, http.response_content_length=171}, capacity=128, totalAddedValues=15}

They are not propagated to the message itself.

According to the documentation
https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/instrumentation/aws-sdk/aws-sdk-2.2/library/README.md

for sqs in addition to adding ExecutionInterceptor you also need:

For SQS an additional step is needed

SqsClientBuilder sqsClientBuilder = SqsClient.builder();
...
SqsClient sqsClient = telemetry.wrap(sqsClientBuilder.build());
SqsAsyncClientBuilder sqsAsyncClientBuilder = SqsAsyncClient.builder();
...
SqsAsyncClient sqsAsyncClient = telemetry.wrap(sqsAsyncClientBuilder.build());

I don't see you setting this, which might explain the issue

Actual Behaviour

traces should be propagated to message attributes

Steps To Reproduce

Use following config

otel:
  register:
    global: true
  javaagent:
    debug: true
  propagators: tracecontext,baggage
  traces:
    exporter: logging

  instrumentation:
    messaging:
      experimental:
        receive-telemetry:
          enabled: true
    aws-sdk:
      experimental-use-propagator-for-messaging: true
      experimental-span-attributes: true

Create a SQS queue listener and send a message through SQSClient ( sqsClient.sendMessage(sendMessageRequest);)

Environment Information

JDK17
MAC

Example Application

No response

Version

4.7.6

[ojebari-ma]

Can you please provide a reproducible example?

[chenlin2001]

I can create a simple example, but truthfully just use sqs to send to a queue and listen to the same queue and see the otel attributes are not being sent. The following fixed the sending context part:

@Replaces(bean = SdkClientBuilderListener.class)
@Slf4j
@Singleton
public class SdkClientBuilderListenerReplacement  implements BeanCreatedEventListener<SdkClientBuilder<?, ?>> {
    private final BeanProvider<OpenTelemetry> openTelemetry;
    public SdkClientBuilderListenerReplacement(BeanProvider<OpenTelemetry> openTelemetry) {
        this.openTelemetry = openTelemetry;
    }
    @Override
    public SdkClientBuilder<?, ?> onCreated(@NonNull BeanCreatedEvent<SdkClientBuilder<?, ?>> event) {
        if (log.isTraceEnabled()) {
            log.trace("Registering OpenTelemetry tracing interceptor to {}", event.getBean().getClass().getSimpleName());
        }
        AwsSdkTelemetry awsSdkTelemetry = AwsSdkTelemetry.builder(openTelemetry.get())
                .setMessagingReceiveInstrumentationEnabled(true)
                .setUseConfiguredPropagatorForMessaging(true)
                .build();
        SdkClientBuilder<?, ?> sdkClientBuilder = event.getBean().overrideConfiguration(builder ->
                builder.addExecutionInterceptor(awsSdkTelemetry.newExecutionInterceptor()));
        return sdkClientBuilder;
    }
}

But while I can see the attributes now arrive on the queue (by adding @MessageHeader("traceparent") @nullable String traceparent ), the full OtelContext isn't being set and that means it isn't propogated to anything else in the application

[chenlin2001]

Attached a simple project with a test that sends a message to a queue showing traceparent is null unless I implement SdkClientBuilderListenerReplacement. However even when then the traceparent is then at the attribute level it's not set in the context as needed. So I'm able to fix the issue of sending the context, but not receiving.
otel-sqs-issues 2.zip

[chenlin2001]

I believe that the problem why just wrapping the SqsClient doesn't fix the issue for receiving is that the way the JmsListener is created in Micronaut (in AbstractJMSListenerMethodProcessor class) - it is not using the SQSClient so the recieveMessage in SQSClient which is the one wrapped by the
io.opentelemetry.instrumentation.awssdk.v2_2.SqsImpl class is never called.

static SqsClient wrap(SqsClient sqsClient) {
    // proxy SqsClient so we could replace the messages list in ReceiveMessageResponse returned from
    // receiveMessage call
    return (SqsClient)
        Proxy.newProxyInstance(
            sqsClient.getClass().getClassLoader(),
            new Class<?>[] {SqsClient.class},
            (proxy, method, args) -> {
              if ("receiveMessage".equals(method.getName())) {
                SqsTracingContext sqsTracingContext = new SqsTracingContext();
                try (Scope ignored =
                    io.opentelemetry.context.Context.current()
                        .with(sqsTracingContext)
                        .makeCurrent()) {
                  Object result = invokeProxyMethod(method, sqsClient, args);
                  TracingList tracingList = sqsTracingContext.get();
                  if (tracingList != null) {
                    ReceiveMessageResponse response = (ReceiveMessageResponse) result;
                    SqsImpl.setMessages(response, tracingList);
                    return response;
                  }
                  return result;
                }
              } else {
                return invokeProxyMethod(method, sqsClient, args);
              }
            });
  }```