Add Support for JMS Tracing in Quarkus with Artemis

[T-Celor]

Hello,

In a Quarkus application using quarkus-artemis-jms-ra with quarkus-ironjacamar for JMS messaging (ActiveMQ), I noticed that OpenTelemetry spans are not automatically managed for JMS messages even with OpenTelemetry enabled. I would expect similar automatic tracing for JMS messages as Kafka does (spans are automatically managed with naming conventions like topic + publish/receive), including span creation/end for publish and receive operations.

Questions:

Do you plan to support this natively in Quarkus?

Context:

Connector: quarkus-artemis-jms-ra with quarkus-ironjacamar
JMS Provider: ActiveMQ
Observability Tools: OpenTelemetry, LGTM

Thank you for considering this feature request. Let me know if additional details are needed.

[vsevel]

see zulip for additional context

[turing85]

FTR: I have written something like this in the past. The process is not too terrible complicated. We would need to transfer the uber-trace-id and the trace-id at the sender and unpack both on the reciever (and set up the span context).

The critical part is performance, though.

There are additional considerations wrt. baggage propagation. But those - I think - should be secondary concerns.

I dug a little bit. Interceptors seem like a good fit. Maybe @zhfeng has some insight too?

[T-Celor]

Hello turing85,

I did a fast workaround.

For the receive part, inside an interceptor :

     @Inject
    JMSCorrelationIdDelegate delegate;

    @Inject
    JmsObservability jmsObservability;

    @AroundInvoke
    public Object around(InvocationContext ctx) throws Exception {
        Method method = ctx.getMethod();
        Class<?>[] parameterTypes = method.getParameterTypes();
        if (method.getName().equals("onMessage") && parameterTypes.length == 1 && parameterTypes[0] == Message.class) {
            Message message = (Message) ctx.getParameters()[0];
            return jmsObservability.observabilityTraceJmsReceive(message, () -> delegate.onMessage(message, ctx));
        } else {
            return ctx.proceed();
        }
    }

For the publish part, inside an helper :

    @Inject
    JmsObservability jmsObservability;

    public JMSProducer send(JMSProducer producer, Destination destination, Message message) {
        try {
            return jmsObservability.observabilityTraceJmsPublish(destination, message, () -> {
                return producer.send(destination, message);
            });
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) { // Needed to respect callable signature
            throw new RuntimeException(e);
        }
    }

And the logic of the observability :

@ApplicationScoped
public class JmsObservability {

    private static final String SPAN_TYPE_MESSAGE = "message";

    private final ContextConfig config;
    private final Tracer tracer;
    private final JmsContextPropagation jmsContextPropagation;

    private final Attributes attributes;

    @Inject
    public JmsObservability(Tracer tracer, JmsContextPropagation jmsContextPropagation) {
        this.tracer = tracer;
        this.jmsContextPropagation = jmsContextPropagation;
        this.attributes = buildGenericAttributes();
    }

    public <T> T observabilityTraceJmsReceive(Message message, Callable<T> action) throws Exception {
        Span span = tracer.spanBuilder(String.format("%s receive", getDestinationName(message.getJMSDestination())))
                .setSpanKind(SpanKind.CONSUMER)
                .setParent(jmsContextPropagation.extractContext(message))
                .startSpan();

        return observabilityTrace(
                message,
                span,
                action);
    }

    public <T> T observabilityTraceJmsPublish(Destination destination, Message message, Callable<T> action) throws Exception {
        Span span = tracer.spanBuilder(String.format("%s publish", getDestinationName(destination)))
                .setSpanKind(SpanKind.PRODUCER)
                .startSpan();

        jmsContextPropagation.injectContext(Context.current(), message);

        return observabilityTrace(
                message,
                span,
                action);
    }

    private <T> T observabilityTrace(Message message, Span span, Callable<T> action) throws Exception {
        try (Scope scope = span.makeCurrent()) {
            setSpanAttributesFromMessage(span, message);
            return action.call();
        } catch (Exception e) {
            handleSpanException(span, e);
            throw e;
        } finally {
            span.end();
        }
    }

    private Attributes buildGenericAttributes() {
        AttributesBuilder attributesBuilder = Attributes.builder();
        attributesBuilder.put(JmsObservabilityPreAllocatedKey.ATTRIBUT_KEY_TYPE, SPAN_TYPE_MESSAGE);
        return attributesBuilder.build();
    }

    private void addOptionalAttribute(AttributesBuilder builder, AttributeKey<String> key, java.util.Optional<String> value) {
        value.ifPresent(v -> builder.put(String.valueOf(key), v));
    }

    private void setSpanAttributesFromMessage(Span span, Message message) {
        try {
            span.setAttribute(JmsObservabilityPreAllocatedKey.ATTRIBUT_KEY_JMS_MESSAGE_ID, message.getJMSMessageID())
                    .setAttribute(JmsObservabilityPreAllocatedKey.ATTRIBUT_KEY_JMS_MESSAGE_CORRELATION_ID,
                            message.getJMSCorrelationID())
                    .setAttribute(JmsObservabilityPreAllocatedKey.ATTRIBUT_KEY_JMS_MESSAGE_TIMESTAMP, message.getJMSTimestamp())
                    .setAllAttributes(attributes);
        } catch (JMSException e) {
            Log.error("Error setting JMS message attributes on span", e);
        }
    }

    private void handleSpanException(Span span, Throwable e) {
        if (span != null) {
            span.recordException(e);
            span.setStatus(StatusCode.ERROR);
            span.setAttribute("error", e.getMessage());
        }
    }

    private String getDestinationName(Object destination) {
        if (destination instanceof ActiveMQDestination) {
            return ((ActiveMQDestination) destination).getName();
        }
        return "unknown";
    }
}

And the propagator :

@ApplicationScoped
public class JmsContextPropagation {

    private final ContextPropagators propagators;

    @Inject
    public JmsContextPropagation(OpenTelemetry openTelemetry) {
        this.propagators = openTelemetry.getPropagators();
    }

    public Context extractContext(Message message) {
        Map<String, String> carrier = new HashMap<>();
        try {
            Enumeration<?> propertyNames = message.getPropertyNames();
            while (propertyNames.hasMoreElements()) {
                String key = (String) propertyNames.nextElement();
                carrier.put(key, message.getStringProperty(key));
            }
        } catch (JMSException e) {
            throw new RuntimeException("Error while reading JMS message properties", e);
        }

        return propagators.getTextMapPropagator().extract(Context.current(), carrier, new TextMapGetter<>() {
            @Override
            public Iterable<String> keys(Map<String, String> carrier) {
                return carrier.keySet();
            }

            @Override
            public String get(Map<String, String> carrier, String key) {
                return carrier.get(key);
            }
        });
    }

    public void injectContext(Context context, Message message) {
        propagators.getTextMapPropagator().inject(context, message, new TextMapSetter<>() {
            @Override
            public void set(Message carrier, String key, String value) {
                try {
                    carrier.setStringProperty(key, value);
                } catch (JMSException e) {
                    throw new RuntimeException("Failed to set property on JMS message", e);
                }
            }
        });
    }
}

• Does my implementation is optimal in performance term ?
• Do you see any issues with how I propagate the trace and uber-trace IDs?
• Are there best practices for baggage propagation I should consider?
• Is my use of interceptors appropriate, or would you recommend another pattern?
• Any advice on testing or edge cases I should cover?

[brunobat]

You cannot do something like: jmsContextPropagation.injectContext(Context.current(), message);
OTel Context must not be sent way.
The propagation related header must follow W3C standard (default) or other.
The OTel java instrumentation API provides interface for that, however is mode convoluted than the low level Java API used here in the examples.

[T-Celor]

@brunobat, beside the jmsContextPropagation.injectContext i use propagators.getTextMapPropagator() from io.opentelemetry.context.propagation.ContextPropagators, this is not the good way ?

• Do you have an example of how to implement context propagation using the OTel Java instrumentation API, following the W3C standard?
• What interfaces or classes from the OTel API should I use for injecting and extracting propagation headers in JMS messages?
• Are there recommended patterns or best practices for this scenario?
• Is there anything specific I should avoid or pay attention to when implementing this with JMS?

[brunobat]

Yes, I actually have a prototype but never got the time to port the implementation to the connector.
See this demo. Has 2 services with send and receive.
Sender: https://github.com/brunobat/quarkus-observability-demo/blob/main/quarkus-observability-demo-activemq/quarkus-observability-demo-activemq-legume/src/main/java/com/brunobat/rest/message/MessageSender.java
Receiver: https://github.com/brunobat/quarkus-observability-demo/blob/main/quarkus-observability-demo-activemq/quarkus-observability-demo-activemq-superhero/src/main/java/com/brunobat/activemq/superhero/message/SuperHeroCreator.java

However, it has a problem because I wrote this before the messaging OTel spec was written... Per spec, Child/Parent relationship should not happen as a general case, using links instead would be generic.
If we can make sure of a few things, we can use parent/child:

1. There is no Bulk message send receive happening
2. On consumption there is no ambiguity on who's the parent. Meaning there is no active span on the consumer, previously to the message consumption.
3. There were no other spans generated between the sender and the consumer... Meaning, there were no spans generated in the broker
4. W3C propagation context was not modified in transit.

Some of these things are very hard to grant on a message crossing process boundaries and I don't think we should commit to them, therefore I propose to place a link to the parent on the consumer side span, instead of setting a parent relationship.

[T-Celor]

And is using io.opentelemetry.context.propagation.ContextPropagators not an option in this case? It seems to follow the recommendation.

[brunobat]

Yes, that should be what we need to use, in concrete, the W3C implementation

[T-Celor]

Thank you, that clarifies things. This is essentially the approach I used for context injection (public void injectContext(Context context, Message message);) and extraction (public Context extractContext(Message message);), although perhaps the method names could be improved.

To confirm my understanding: was my mistake performing the injection outside the try-with-resources block for the span? Should I instead structure it as follows, ensuring that context injection or extraction occurs within the span’s scope?

 try (Scope scope = span.makeCurrent()) {
            setSpanAttributesFromMessage(span, message);
             // <------ Inject or extract context here, depending on send/receive.
            return action.call();
        } catch (Exception e) {
            handleSpanException(span, e);
            throw e;
        } finally {
            span.end();
        }

Would this be the correct approach?

[brunobat]

That needs to happen as well, yes.
Sorry, I didn't look at this:

public void injectContext(Context context, Message message) {
        propagators.getTextMapPropagator().inject(context, message, new TextMapSetter<>() {
            @Override
            public void set(Message carrier, String key, String value) {
                try {
                    carrier.setStringProperty(key, value);
                } catch (JMSException e) {
                    throw new RuntimeException("Failed to set property on JMS message", e);
                }
            }
        });
    }

And I think this will work.

We might need to set additional attributes in the span, at least the ones that make sense for JMS and are listed here: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes

[T-Celor]

Yes, i have conclusive test with local env (Otel-collector, kibana, elastic and apm).

We might need to set additional attributes in the span, at least the ones that make sense for JMS and are listed here: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes

I will add them to the code, thanks to you !