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.

[tomazfernandes]

Hi everyone, just wanted to let you know that we're paying attention to this issue and should be able to work on the Micrometer integration and review related PRs soon.

There's a lot going on in terms of the threading model for the SQS integration and we want to make sure we get everything right for this, while also coming up with a solution that will be extendable to all modules.

Thank you for the patience, this being a community, non-sponsored project we need to find the time to work on issues such as this that requires more attention from the team.

Thanks!

[tomazfernandes]

Hey everyone!

I have some exciting news to share - I've opened up a PR to add Micrometer Observability support for SQS: #1369.

The main challenge was of course figuring out the best way to handle context propagation and automatic scope opening through CompletableFutures and arbitrary (potentially async) user logic in the components.

Features

In this PR, we'll automatically open scopes for any Blocking (non-async) components, so for a large majority of user use-cases it should just work OOTB for Listeners, Interceptors and Error Handlers.

We offer manual propagation support for async components via message headers - there's room for improvement there, but should be enough to start with.

We also support the same features as other Spring messaging projects, such as auto-wiring ObservationRegistries and allowing custom ObservationConventions, with similar default tags.

@jonatan-ivanov, I had a great time walking through Micrometer's code, and I'm happy with the solution in the PR - it just shows how the abstractions you folks built are precise and flexible enough to handle even a non-trivial use-case such as this.

@sondemar, I appreciate very much the PR you opened and it helped me a lot in getting things started - thank you! I ended up going with a different approach, but did use some of the tests you built, you're credited. If you're up for it, it would be awesome if you'd like to contribute the Batch message support you had in your PR once this PR is merged.

It would be really amazing if any / all of you - also tagging @jkuipers @jeevjyot @nikola-djuran - could take a look at the PR and / or run it locally and see if there are issues / improvements to be made. PR turned out a bit bigger than I expected, but a lot of it is tests, and most of it is just what was necessary to add. Of course, no hurry, and also no problem if you folks can't, I appreciate us getting this far regardless.

@maciejwalkowiak @MatejNedic, you're more than welcome to take a look at it too.

That's it, I apologize for the long write-up, and please let me know if you have any questions / suggestions.

Thank you!

[jkuipers]

@tomazfernandes that’s awesome! I’ll try to make some time next week to give this a test run in my own project to see if it works as expected and will provide my feedback.

[jonatan-ivanov]

I had a great time walking through Micrometer's code, and I'm happy with the solution in the PR - it just shows how the abstractions you folks built are precise and flexible enough to handle even a non-trivial use-case such as this.

@tomazfernandes That's awesome to read! I can take a look at the PR next week (we release on Monday) if that's ok. Also, please let us know if you have any ideas for improvements either by opening an issue or posting a message in https://slack.micrometer.io

[tomazfernandes]

I can take a look at the PR next week (we release on Monday) if that's ok.

Hey @jonatan-ivanov, sure, that's great, thanks!

Also, please let us know if you have any ideas for improvements either by opening an issue or posting a message in https://slack.micrometer.io/

Sure, will do. Overall I think it all just works super well, and once I figured out how the overall solution should look like all the flexibility I needed was just there already. I'll let you guys know if I can think of any suggestions.