Add support for Micrometer's Observation API

[maciejwalkowiak]

• simplify configuring AWS SDK v2 tracing for Brave and Otel
  • Otel: https://github.com/open-telemetry/opentelemetry-java-instrumentation/tree/main/instrumentation/aws-sdk/aws-sdk-2.2/library
  • Brave: https://github.com/openzipkin/zipkin-aws/blob/master/brave/instrumentation-aws-java-sdk-v2-core/README.md
• create spans for SQS listeners

Also #566

[jonatan-ivanov]

You might want to consider using the Observation API so that you can get Metrics and Tracing (and any other arbitrary thing that users need/want to implement): https://micrometer.io/docs/observation

[maciejwalkowiak]

@jonatan-ivanov thanks for a hint. Would you like to contribute it maybe?🙂

[jonatan-ivanov]

I would but in the next few months I'm not sure I will be able to do it. Let me talk to the rest of the team to see if we can prioritize this.

[nikola-djuran]

Just stumbled on this after 2 days of examining the code and trying to figure out a solution for, at least, propagating the trace through SQS to keep log aggregation for the entire logical operation... In this short venture in the code I concluded there are two points for integration:

• Add an ExecutionInterceptor in a AwsClientCustomizer bean for the sending service that will add the trace header to the outgoing message(something like TraceIdExecutionInterceptor AWS made for Lambdas)
• Add a MessageInterceptor for the receiving service that will pickup the trace from the message attributes and create the trace context with it.

Does that sound like I'm on the right path or am I missing something?

[jkuipers]

Hey @nikola-djuran, I looked into this recently and documented my solution here:
#902

This seems to work well. I'm currently working on the migration from 2.4 to 3.0 and this was a crucial thing for us too.

[nikola-djuran]

Hi @jkuipers thanks for reaching out and sharing this. I just looked at it and it seems like that's the think I'm trying to figure out as well. Looked at most of the classes you're working with(apart from Brave specific), however I'm trying to do it with the Micrometer Tracing API, to abstract Brave if possible. :)

If you're still working with this, take a look at the classes I mentioned above...I might be wrong, but they seem to pin-point the integration parts. Not that your approach is invalid, I'm just a fan of changing as little as possible if there is a place in the framework for it than re-declaring higher level beans. I feel like there's much more that can go wrong that way 😅 .

[jonatan-ivanov]

What do you think about:

1. Repurposing the issue to use the Observation API instead of "just" Micrometer Tracing
2. Asking the AWS SDK maintainers to instrument the SDK so that not only Spring-Cloud AWS users can benefit from the instrumentation but everyone: Add support for Micrometer's Observation API  aws/aws-sdk-java-v2#4611

[nikola-djuran]

Definitely agree with the idea! That would be the most sensible solution, I was just trying to get a hint how to implement a short term solution that would cover the tracing part until this goes into a release.

[jkuipers]

About instrumenting the SDK instead: even though that sounds nice, I'm wondering if it would work with all the work that's performed using CompletableFutures in Spring Cloud AWS. To me it wasn't very clear when I looked at the code where thread-switching might occur. That's one of the reasons I chose the extension points I'm currently using: for those I'm sure that the work happens on the correct thread.
Doesn't mean it can't be done, just that I'm not smart enough to do the work ;)

[jeevjyot]

Hi there,

We are working on a project where we need to trace all messages once consumed from SQS. We are attempting to add the tracing context to the messages as shown below:

SqsAsyncClient.builder()
                .overrideConfiguration(ClientOverrideConfiguration.builder()
                        .addExecutionInterceptor(new TracingExecutionInterceptor())
                        .build())
                .build();

and when receive the message,

       private Mono<ProcessingStatus> processMessage(Message message) {

        Propagator.Getter<Message> getter = new Propagator.Getter<>() {
            /**
             * retrieves the traceId from the sqs messsage, assuming the key is "X-B3-TraceId"
             *
             * @param carrier carrier of propagation fields, such as an http request.
             * @param key     the key of the field.
             * @return
             */
            @Override
            public String get(Message carrier, String key) {
                return carrier.attributesAsStrings().get(key);
            }
        };
        ReceiverContext<Message> context = new ReceiverContext<>(getter);

        var sqsObservation = Observation.createNotStarted("sqs", () -> context, observationRegistry);
        var observation = new NullObservation(observationRegistry)
                .observe(() -> sqsObservation);
        return observation.observe(() -> Mono.defer(() -> processObserved(message)
                .contextWrite(Context.of(ObservationThreadLocalAccessor.KEY, observation))));
    }

Our question is: Is there a way to integrate OpenTelemetry (otel) instead of using our own Propagator.Getter? We are looking for a working example, possibly leveraging this OpenTelemetry AWS SDK instrumentation.

[jonatan-ivanov]

Fyi: since Micrometer Tracing supports OTel, if SQS would be instrumented with the Observation API, OTel would work too.

[tomazfernandes]

I can't speak for AWS SDK, but I think we should be able to instrument SCAWS SQS to provide Micrometer support OOTB.

Off the top of my head, while the async / threading nature of the integration has its challenges and it's hard to give strong guarantees for large chunks of code, basically it'll only hop threads when doing some I/O such as using an async method from AWS SDK or from user code in components such as Interceptors / Error Handlers.

So we do have some control we should be able to leverage in keeping any necessary context.

Another option, and perhaps a more idiomatic way to pass this kind of metadata along in a Messaging framework, would be using MessageHeaders as the transport - if we can have some kind of component that would pick information from the Thread early in the process and transform it into headers, we should be able to pass it along to any necessary downstream components.

[jonatan-ivanov]

@maciejwalkowiak Would you mind renaming this issue for something like "Add support for Micrometer's Observation API"?

[maciejwalkowiak]

@jonatan-ivanov done!

[0x006EA1E5]

Is there still plans to do this?

Do you need help implementing it?

[tomazfernandes]

Hey @0x006EA1E5, yup, PRs are definitely welcome.

[sondemar]

Hey @0x006EA1E5, yup, PRs are definitely welcome.

Hi, I am working on it and plan to deliver a PR in the upcoming days.

[0x006EA1E5]

hey @sondemar are you just working on SQS? Do you intend adding a general ExecutionInterceptor?

[sondemar]

hey @sondemar are you just working on SQS? Do you intend adding a general ExecutionInterceptor?

Hi @0x006EA1E5 yes, I am working on SQS processing as described in the following issue. Regarding the general ExecutionInterceptor. What would you like to achieve by creating this dedicated interceptor? I am asking because, in the attached PR, you can find an example of using a custom interceptor in an observation manner.