Propagate and Observation from Reactive context #3999
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
To propagate an `Observation` from reactive stream (e.g. WebFlux) we have to capture its context and set it into the current thread scope. * Add a `io.micrometer:context-propagation` dependency to support reactive context propagation * Populate a `parentObservation` in the `IntegrationObservation.PRODUCER.observation()` since it is not available for tracing on `Observation.onStart()`. Might be tentative until upcoming fix in Micrometer Observation * Populate from reactive context in the `WebFluxInboundEndpoint` where we use just `send()` operation downstream * Populate from reactive context in the `MessagingGatewaySupport` where we use `send()` operation downstream or `FluxMessageChannel.subscribeTo()` * Use `contextCapture()` in the `FluxMessageChannel` to gather a `ThreadLocal` info into a Reactor context and then set back to `ThreadLocal` in the `transformDeferredContextual()` which really happens on a different thread * Verify a trace propagation from WebFlux to an integration flow via Brave instrumentation in the `WebFluxObservationPropagationTests`
garyrussell
reviewed
Feb 1, 2023
| .flatMap(replyMessage -> populateResponse(exchange, replyMessage)); | ||
| } | ||
| else { | ||
| var scope = ContextSnapshot.setAllThreadLocalsFrom(contextView); |
There was a problem hiding this comment.
Why is this not required in the send/receive case?
There was a problem hiding this comment.
It is covered by the MessagingGatewaySupport change in this PR.
We need such an action only when we lose a reactive scope.
This send() returns nothing, but sendAndReceiveMessageReactive() returns a Mono where respective context is still present. Therefore we have to repeat a set to ThreadLocal over there where we lose the reactive stream context again.
That's why I have those separate tests in this PR.
NOTE: The RSocket support will work as well just because we have a fix in the MessagingGatewaySupport and it does not use a plain send().
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
To propagate an
Observationfrom reactive stream (e.g. WebFlux) we have to capture its context and set it into the current thread scope.io.micrometer:context-propagationdependency to support reactive context propagationparentObservationin theIntegrationObservation.PRODUCER.observation()since it is not available for tracing onObservation.onStart(). Might be tentative until upcoming fix in Micrometer ObservationWebFluxInboundEndpointwhere we use justsend()operation downstreamMessagingGatewaySupportwhere we usesend()operation downstream orFluxMessageChannel.subscribeTo()contextCapture()in theFluxMessageChannelto gather aThreadLocalinfo into a Reactor context and then set back toThreadLocalin thetransformDeferredContextual()which really happens on a different threadWebFluxObservationPropagationTests