Customize the Observation created by MicrometerObservationListener

[nvervelle]

Motivation

The current integration between reactor and micrometer, with tap() and Micrometer.observation(), only allows simple use cases for the creation of the Observation that will observe the reactive pipeline : it's currently not possible to provide a context supplier or a convention that will be used to create the Observation.

One example where the current solution is limiting is when using Kafka in spring boot where the listener is a Flux, each kafka message becoming an element of the Flux. It's not possible to extract the traceparent header from the kafka message to use it when creating the Observation.

Desired solution

I'd like to be able to give more control on what Micrometer.observation() is doing. My idea is to be able to provide an optional Function<ObservationRegistry, Observation> so that the Observation will be created with as much customization as required.

Considered alternatives

I first tried a solution that will not imply any modification on reactor-micrometer library : I had to copy/paste 3 package private classes from the library (MicrometerObservationListener, MicrometerObservationListenerConfiguration and MicrometerObservationListenerFactory) to be able to modify them. It's not a good solution in the long term, as it needs to keep the 3 modified classes in sync with the library.

Additional context

I have submitted a PR with my ideas about the Function<>: #3456

[OlegDokuka]

Hi, @nvervelle!

Thanks for good idea. Can you please provide a code sample of how that could be used if you have such API.

Best,
Oleh

[nvervelle]

Hi @OlegDokuka,

Yes, sure. It would be useful for example for observing kafka listeners in Spring Boot. A very simplified example compared to what I currently have to do to observe single messages in the receiver flux. Tell me if it's enough.

import io.micrometer.observation.Observation;
import org.springframework.kafka.support.micrometer.KafkaListenerObservation;
import org.springframework.kafka.support.micrometer.KafkaRecordReceiverContext;
import reactor.core.observability.micrometer.MicrometerObservationListenerFactory;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverRecord;

KafkaReceiver.create(buildReceiverOptions(...))
    .receive()
    .flatMap(this::receiveRecord);

private Mono<Boolean> receiveRecord(ReceiverRecord receiverRecord) {
    return Mono.defer(() -> Mono.defer(...).contextCapture())
        .name("receiveRecord")
        .tap(MicrometerObservationListenerFactory.observation(
            observationRegistry,
            // HERE: I can customize the observation created by tap() with both
            // * a convention specialized for kafka messages
            // * a context supplier that will retrieve context information from the kafka message header
            registry -> Observation.createNotStarted(
                KafkaListenerObservation.DefaultKafkaListenerObservationConvention.INSTANCE,
                () -> new KafkaRecordReceiverContext(receiverRecord, "listenerId", () -> "clusterId"),
                registry)));
}

[OlegDokuka]

Makes sense to me now. Do you mind signing CLA so i can merge your PR?

[nvervelle]

I asked my legal department to sign the CLA or give me permission to sign it, I hope I will have their approval in the next few days...

[nvervelle]

Hi @OlegDokuka, I've signed the CLA, so you can merge the PR if you want 👍

[nvervelle]

Hi @OlegDokuka , any hope to merge the PR in the near future ?

[chemicL]

@nvervelle I asked for some tests and documentation in the PR, hopefully we can ship it in the next release :)

[nvervelle]

@chemicL I've pushed a commit for the tests and documentation

[ajax-osadchuk-r]

This improvement will help my team a lot, we are also looking for a way to properly use the micrometer with the reactor-kafka.
@nvervelle your solution seems to actually what we need.
@chemicL Looking forward to get this feature released.

[chemicL]

@nvervelle thank you for the contribution! This change will be part of 3.5.8 and 3.6.0-M1 🚀 Looking forward to more contributions 👏