-
Notifications
You must be signed in to change notification settings - Fork 2.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Reactive Messaging OTel tracing decorator update for 4.11.0
- Loading branch information
1 parent
adf8f7d
commit 6a3982b
Showing
4 changed files
with
108 additions
and
38 deletions.
There are no files selected for viewing
This file contains 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
50 changes: 50 additions & 0 deletions
50
...me/tracing/intrumentation/reactivemessaging/ReactiveMessagingTracingEmitterDecorator.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| package io.quarkus.opentelemetry.runtime.tracing.intrumentation.reactivemessaging; | ||
|
|
||
| import static io.quarkus.opentelemetry.runtime.tracing.intrumentation.reactivemessaging.ReactiveMessagingTracingOutgoingDecorator.decorateOutgoing; | ||
|
|
||
| import jakarta.enterprise.context.ApplicationScoped; | ||
| import jakarta.inject.Inject; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Message; | ||
|
|
||
| import io.smallrye.mutiny.Multi; | ||
| import io.smallrye.reactive.messaging.ChannelRegistry; | ||
| import io.smallrye.reactive.messaging.PublisherDecorator; | ||
|
|
||
| /** | ||
| * Intercepts outgoing messages from emitters from Reactive Messaging connectors. | ||
| * <p> | ||
| * For outgoing messages from emitters, if the message doesn't already contain a tracing metadata, it attaches one with the | ||
| * current | ||
| * OpenTelemetry context. | ||
| * Reactive messaging outbound connectors, if tracing is supported, will use that context as parent span to trace outbound | ||
| * message transmission. | ||
| */ | ||
| @ApplicationScoped | ||
| public class ReactiveMessagingTracingEmitterDecorator implements PublisherDecorator { | ||
|
|
||
| @Override | ||
| public int getPriority() { | ||
| // Place the decorator before all others including the ContextDecorator which is priority 0 | ||
| // This is only important for the emitter case | ||
| return -1000; | ||
| } | ||
|
|
||
| @Inject | ||
| ChannelRegistry registry; | ||
|
|
||
| /** | ||
| * Incoming messages | ||
| */ | ||
| @Override | ||
| public Multi<? extends Message<?>> decorate(Multi<? extends Message<?>> publisher, | ||
| String channelName, boolean isConnector) { | ||
| Multi<? extends Message<?>> multi = publisher; | ||
| if (!isConnector && registry.getEmitterNames().contains(channelName)) { | ||
| // Emitter is a special case for the emitter publisher | ||
| multi = decorateOutgoing(multi); | ||
| } | ||
| return multi; | ||
| } | ||
|
|
||
| } |
This file contains 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
50 changes: 50 additions & 0 deletions
50
...e/tracing/intrumentation/reactivemessaging/ReactiveMessagingTracingOutgoingDecorator.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| package io.quarkus.opentelemetry.runtime.tracing.intrumentation.reactivemessaging; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| import jakarta.enterprise.context.ApplicationScoped; | ||
|
|
||
| import org.eclipse.microprofile.reactive.messaging.Message; | ||
|
|
||
| import io.quarkus.opentelemetry.runtime.QuarkusContextStorage; | ||
| import io.smallrye.mutiny.Multi; | ||
| import io.smallrye.reactive.messaging.SubscriberDecorator; | ||
| import io.smallrye.reactive.messaging.TracingMetadata; | ||
|
|
||
| /** | ||
| * Intercepts outgoing messages from Reactive Messaging connectors. | ||
| * <p> | ||
| * For outgoing messages, if the message doesn't already contain a tracing metadata, it attaches one with the current | ||
| * OpenTelemetry context. | ||
| * Reactive messaging outbound connectors, if tracing is supported, will use that context as parent span to trace outbound | ||
| * message transmission. | ||
| */ | ||
| @ApplicationScoped | ||
| public class ReactiveMessagingTracingOutgoingDecorator implements SubscriberDecorator { | ||
|
|
||
| /** | ||
| * Outgoing messages | ||
| */ | ||
| @Override | ||
| public Multi<? extends Message<?>> decorate(Multi<? extends Message<?>> toBeSubscribed, | ||
| List<String> channelName, boolean isConnector) { | ||
| Multi<? extends Message<?>> multi = toBeSubscribed; | ||
| if (isConnector) { | ||
| // add TracingMetadata to the outgoing message if it doesn't exist already | ||
| multi = decorateOutgoing(multi); | ||
| } | ||
| return multi; | ||
| } | ||
|
|
||
| static Multi<? extends Message<?>> decorateOutgoing(Multi<? extends Message<?>> multi) { | ||
| return multi.map(m -> { | ||
| Message<?> message = m; | ||
| if (m.getMetadata(TracingMetadata.class).isEmpty()) { | ||
| var otelContext = QuarkusContextStorage.INSTANCE.current(); | ||
| message = m.addMetadata(TracingMetadata.withCurrent(otelContext)); | ||
| } | ||
| return message; | ||
| }); | ||
| } | ||
|
|
||
| } |