You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jb...@apache.org on 2026/10/19 04:02:54 UTC
(activemq) branch main updated: feat(otel): Add OpenTelemetry distributed tracing support (#1769)
This is an automated email from the ASF dual-hosted git repository.
jbonofre pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/main by this push:
new 7d05f6fda9 feat(otel): Add OpenTelemetry distributed tracing support (#1769)
7d05f6fda9 is described below
commit 7d05f6fda99c0396027f1bf52516d415c59fcfb2
Author: JB Onofré <jb...@apache.org>
AuthorDate: Mon Oct 19 05:02:48 2026 +0100
feat(otel): Add OpenTelemetry distributed tracing support (#1769)
* Add OpenTelemetry distributed tracing support
New activemq-opentelemetry module that instruments the broker using the
OpenTelemetry API. The plugin traces send (PRODUCER), dispatch (CONSUMER),
and acknowledge (INTERNAL) operations with W3C TraceContext propagation
and standard OTel messaging semantic conventions.
Depends on opentelemetry-api only; users bring their own SDK and exporter
at runtime. Included in the distribution with example configuration.
* Address jeanouii's review comments on OpenTelemetry plugin
- Fix OTel semantic convention attribute names:
messaging.system -> messaging.system.name
messaging.operation -> messaging.operation.type
messaging.client_id -> messaging.client.id (dot notation)
- Initialize propagator and tracer once in constructor instead of per-invocation
- Bound dispatchSpans map to prevent unbounded growth / span leaks;
oldest span is ended and evicted when the limit is exceeded
- Add comment explaining setMarshalledProperties(null) in TextMapSetter
- Replace Thread.sleep() in tests with Wait.waitFor() pattern
- Document topic fan-out span behavior in class Javadoc
* Fix activemq-opentelemetry parent version to match reactor (6.4.0-SNAPSHOT)
The module's parent POM referenced activemq-parent:6.3.0-SNAPSHOT, which
no longer exists since the root pom.xml was bumped to 6.4.0-SNAPSHOT,
breaking CI reactor resolution.
* feat(otel): Add activemq-opentelemetry to activemq-spring schema and fix example config
- Wire activemq-opentelemetry into activemq-spring (optional dependency and maven-xbean-plugin includes) so that openTelemetryPlugin is generated in activemq.xsd and recognized by XBean Spring
- Remove unnecessary jetty.xml import and ws transport connector from activemq-opentelemetry.xml to pass XML validation and standalone broker start tests
* Address mattrpav's review: configurable messaging system, attribute constants, semconv fixes
- Add messagingSystem property (default 'activemq')
- Use AttributeKey constants for attribute keys and constants for values
- Use messaging.system and the 'settle' operation type per semantic conventions
- Resolve OpenTelemetry lazily and allow explicit injection via setOpenTelemetry
- Log trace context read failures; null-safe span error status
- Document deliver/ack span semantics and dispatch span cap
---
activemq-opentelemetry/pom.xml | 115 ++++++++
.../ActiveMQMessageTextMapGetter.java | 57 ++++
.../ActiveMQMessageTextMapSetter.java | 42 +++
.../opentelemetry/OpenTelemetryBrokerPlugin.java | 296 +++++++++++++++++++++
.../OpenTelemetryBrokerPluginTest.java | 269 +++++++++++++++++++
activemq-spring/pom.xml | 6 +
assembly/pom.xml | 4 +
assembly/src/release/conf/activemq.xml | 9 +
.../examples/conf/activemq-opentelemetry.xml | 118 ++++++++
bom/pom.xml | 5 +
pom.xml | 16 ++
11 files changed, 937 insertions(+)
diff --git a/activemq-opentelemetry/pom.xml b/activemq-opentelemetry/pom.xml
new file mode 100644
index 0000000000..f0023949e4
--- /dev/null
+++ b/activemq-opentelemetry/pom.xml
@@ -0,0 +1,115 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-parent</artifactId>
+ <version>6.4.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>activemq-opentelemetry</artifactId>
+ <packaging>bundle</packaging>
+ <name>ActiveMQ :: OpenTelemetry</name>
+ <description>ActiveMQ OpenTelemetry tracing support</description>
+
+ <dependencies>
+
+ <!-- =============================== -->
+ <!-- Required Dependencies -->
+ <!-- =============================== -->
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>activemq-broker</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-context</artifactId>
+ </dependency>
+
+ <!-- =============================== -->
+ <!-- Testing Dependencies -->
+ <!-- =============================== -->
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>activemq-kahadb-store</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>activemq-broker</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk-testing</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <extensions>true</extensions>
+ <configuration>
+ <instructions>
+ <Bundle-SymbolicName>org.apache.activemq.opentelemetry</Bundle-SymbolicName>
+ <Export-Package>org.apache.activemq.broker.util.opentelemetry*;version=${project.version};-noimport:=true;-split-package:=merge-first</Export-Package>
+ <Import-Package>
+ org.apache.activemq*;version=${project.version};resolution:=optional,
+ io.opentelemetry.*;resolution:=optional,
+ *
+ </Import-Package>
+ <_noee>true</_noee>
+ </instructions>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/ActiveMQMessageTextMapGetter.java b/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/ActiveMQMessageTextMapGetter.java
new file mode 100644
index 0000000000..850ded8d32
--- /dev/null
+++ b/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/ActiveMQMessageTextMapGetter.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.util.opentelemetry;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import io.opentelemetry.context.propagation.TextMapGetter;
+import org.apache.activemq.command.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ActiveMQMessageTextMapGetter implements TextMapGetter<Message> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ActiveMQMessageTextMapGetter.class);
+
+ public static final ActiveMQMessageTextMapGetter INSTANCE = new ActiveMQMessageTextMapGetter();
+
+ @Override
+ public Iterable<String> keys(Message message) {
+ try {
+ if (message.getProperties() != null) {
+ return message.getProperties().keySet();
+ }
+ } catch (IOException e) {
+ LOG.debug("Failed to read message properties for trace context extraction", e);
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public String get(Message message, String key) {
+ try {
+ Object value = message.getProperty(key);
+ if (value instanceof String) {
+ return (String) value;
+ }
+ } catch (IOException e) {
+ LOG.debug("Failed to read trace context property '{}' from message", key, e);
+ }
+ return null;
+ }
+}
diff --git a/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/ActiveMQMessageTextMapSetter.java b/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/ActiveMQMessageTextMapSetter.java
new file mode 100644
index 0000000000..d5b3518541
--- /dev/null
+++ b/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/ActiveMQMessageTextMapSetter.java
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.util.opentelemetry;
+
+import java.io.IOException;
+
+import io.opentelemetry.context.propagation.TextMapSetter;
+import org.apache.activemq.command.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ActiveMQMessageTextMapSetter implements TextMapSetter<Message> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ActiveMQMessageTextMapSetter.class);
+
+ public static final ActiveMQMessageTextMapSetter INSTANCE = new ActiveMQMessageTextMapSetter();
+
+ @Override
+ public void set(Message message, String key, String value) {
+ try {
+ message.setProperty(key, value);
+ // Clear the cached marshalled form so the injected property is included on next serialization.
+ message.setMarshalledProperties(null);
+ } catch (IOException e) {
+ LOG.warn("Failed to set trace context property '{}' on message", key, e);
+ }
+ }
+}
diff --git a/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/OpenTelemetryBrokerPlugin.java b/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/OpenTelemetryBrokerPlugin.java
new file mode 100644
index 0000000000..de0a8282dd
--- /dev/null
+++ b/activemq-opentelemetry/src/main/java/org/apache/activemq/broker/util/opentelemetry/OpenTelemetryBrokerPlugin.java
@@ -0,0 +1,296 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.util.opentelemetry;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import org.apache.activemq.broker.BrokerPluginSupport;
+import org.apache.activemq.broker.ConsumerBrokerExchange;
+import org.apache.activemq.broker.ProducerBrokerExchange;
+import org.apache.activemq.command.Message;
+import org.apache.activemq.command.MessageAck;
+import org.apache.activemq.command.MessageDispatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A broker plugin that provides OpenTelemetry distributed tracing support.
+ * Traces message send, dispatch, and acknowledge operations with proper
+ * context propagation via W3C TraceContext.
+ *
+ * <p>For topic destinations, each subscriber dispatch creates its own span,
+ * so a single published message may produce multiple deliver spans.
+ *
+ * <p>The deliver span is opened in {@code preProcessDispatch} and ended in
+ * {@code postProcessDispatch}, so it measures the broker-side dispatch of the
+ * message to the consumer's transport, not the consumer's own processing time.
+ * It is a child of the publish span. The acknowledge span is an independent
+ * span: an acknowledgement carries no message trace context, so it cannot be
+ * parented to the publish/deliver spans.
+ *
+ * <p>The OpenTelemetry instance is resolved lazily on first use (from
+ * {@link GlobalOpenTelemetry} unless one is set explicitly via
+ * {@link #setOpenTelemetry(OpenTelemetry)}), so the SDK may be registered
+ * after the plugin is instantiated but must be registered before the first
+ * traced message.
+ *
+ * @org.apache.xbean.XBean element="openTelemetryPlugin"
+ */
+public class OpenTelemetryBrokerPlugin extends BrokerPluginSupport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenTelemetryBrokerPlugin.class);
+ private static final String INSTRUMENTATION_NAME = "org.apache.activemq";
+
+ // Semantic convention attribute keys (https://opentelemetry.io/docs/specs/semconv/messaging/)
+ static final AttributeKey<String> ATTR_MESSAGING_SYSTEM = AttributeKey.stringKey("messaging.system");
+ static final AttributeKey<String> ATTR_DESTINATION_NAME = AttributeKey.stringKey("messaging.destination.name");
+ static final AttributeKey<String> ATTR_OPERATION_TYPE = AttributeKey.stringKey("messaging.operation.type");
+ static final AttributeKey<String> ATTR_MESSAGE_ID = AttributeKey.stringKey("messaging.message.id");
+ static final AttributeKey<String> ATTR_CLIENT_ID = AttributeKey.stringKey("messaging.client.id");
+
+ // Semantic convention values
+ static final String DEFAULT_MESSAGING_SYSTEM = "activemq";
+ static final String OPERATION_PUBLISH = "publish";
+ static final String OPERATION_DELIVER = "deliver";
+ static final String OPERATION_SETTLE = "settle";
+
+ // Span name suffixes
+ static final String SPAN_SUFFIX_PUBLISH = " publish";
+ static final String SPAN_SUFFIX_DELIVER = " deliver";
+ static final String SPAN_SUFFIX_ACK = " ack";
+
+ private static final String UNKNOWN_DESTINATION = "unknown";
+
+ // Cap in-flight dispatch spans to avoid unbounded growth when postProcessDispatch is missed.
+ // The eldest entry is ended and evicted if this limit is exceeded, so under more than this
+ // many concurrent in-flight dispatches the oldest deliver spans end early (truncated).
+ private static final int MAX_DISPATCH_SPANS = 1000;
+
+ private boolean traceProducer = true;
+ private boolean traceConsumer = true;
+ private boolean traceAcknowledge = true;
+
+ private String messagingSystem = DEFAULT_MESSAGING_SYSTEM;
+
+ private volatile OpenTelemetry openTelemetry;
+ private volatile TextMapPropagator propagator;
+ private volatile Tracer tracer;
+
+ private final Map<MessageDispatch, Span> dispatchSpans = Collections.synchronizedMap(
+ new LinkedHashMap<MessageDispatch, Span>(256, 0.75f, false) {
+ @Override
+ protected boolean removeEldestEntry(Map.Entry<MessageDispatch, Span> eldest) {
+ if (size() > MAX_DISPATCH_SPANS) {
+ LOG.warn("dispatchSpans exceeded max size; ending oldest span to prevent leak");
+ eldest.getValue().end();
+ return true;
+ }
+ return false;
+ }
+ });
+
+ public OpenTelemetryBrokerPlugin() {
+ }
+
+ /**
+ * Use the given OpenTelemetry instance instead of {@link GlobalOpenTelemetry}.
+ */
+ public void setOpenTelemetry(OpenTelemetry openTelemetry) {
+ this.openTelemetry = openTelemetry;
+ this.tracer = null;
+ this.propagator = null;
+ }
+
+ public String getMessagingSystem() {
+ return messagingSystem;
+ }
+
+ /**
+ * Value of the {@code messaging.system} attribute set on every span (default: {@code activemq}).
+ */
+ public void setMessagingSystem(String messagingSystem) {
+ this.messagingSystem = messagingSystem;
+ }
+
+ private Tracer tracer() {
+ Tracer t = tracer;
+ if (t == null) {
+ OpenTelemetry otel = openTelemetry;
+ t = otel != null ? otel.getTracer(INSTRUMENTATION_NAME) : GlobalOpenTelemetry.getTracer(INSTRUMENTATION_NAME);
+ tracer = t;
+ }
+ return t;
+ }
+
+ private TextMapPropagator propagator() {
+ TextMapPropagator p = propagator;
+ if (p == null) {
+ OpenTelemetry otel = openTelemetry;
+ p = (otel != null ? otel.getPropagators() : GlobalOpenTelemetry.getPropagators()).getTextMapPropagator();
+ propagator = p;
+ }
+ return p;
+ }
+
+ private SpanBuilder newSpan(String name, SpanKind kind, String destinationName, String operationType) {
+ return tracer().spanBuilder(name)
+ .setSpanKind(kind)
+ .setAttribute(ATTR_MESSAGING_SYSTEM, messagingSystem)
+ .setAttribute(ATTR_DESTINATION_NAME, destinationName)
+ .setAttribute(ATTR_OPERATION_TYPE, operationType);
+ }
+
+ public boolean isTraceProducer() {
+ return traceProducer;
+ }
+
+ public void setTraceProducer(boolean traceProducer) {
+ this.traceProducer = traceProducer;
+ }
+
+ public boolean isTraceConsumer() {
+ return traceConsumer;
+ }
+
+ public void setTraceConsumer(boolean traceConsumer) {
+ this.traceConsumer = traceConsumer;
+ }
+
+ public boolean isTraceAcknowledge() {
+ return traceAcknowledge;
+ }
+
+ public void setTraceAcknowledge(boolean traceAcknowledge) {
+ this.traceAcknowledge = traceAcknowledge;
+ }
+
+ @Override
+ public void send(ProducerBrokerExchange producerExchange, Message messageSend) throws Exception {
+ if (!traceProducer) {
+ super.send(producerExchange, messageSend);
+ return;
+ }
+
+ Context parentContext = propagator().extract(Context.current(), messageSend, ActiveMQMessageTextMapGetter.INSTANCE);
+
+ String destinationName = messageSend.getDestination().getPhysicalName();
+ SpanBuilder spanBuilder = newSpan(destinationName + SPAN_SUFFIX_PUBLISH, SpanKind.PRODUCER,
+ destinationName, OPERATION_PUBLISH)
+ .setParent(parentContext);
+
+ if (messageSend.getMessageId() != null) {
+ spanBuilder.setAttribute(ATTR_MESSAGE_ID, messageSend.getMessageId().toString());
+ }
+ if (producerExchange.getConnectionContext() != null
+ && producerExchange.getConnectionContext().getClientId() != null) {
+ spanBuilder.setAttribute(ATTR_CLIENT_ID, producerExchange.getConnectionContext().getClientId());
+ }
+
+ Span span = spanBuilder.startSpan();
+ try {
+ Context contextWithSpan = parentContext.with(span);
+ propagator().inject(contextWithSpan, messageSend, ActiveMQMessageTextMapSetter.INSTANCE);
+
+ super.send(producerExchange, messageSend);
+ } catch (Exception e) {
+ span.setStatus(StatusCode.ERROR, String.valueOf(e.getMessage()));
+ span.recordException(e);
+ throw e;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public void preProcessDispatch(MessageDispatch messageDispatch) {
+ if (traceConsumer && messageDispatch != null && messageDispatch.getMessage() != null) {
+ try {
+ Message message = messageDispatch.getMessage();
+ Context extractedContext = propagator().extract(Context.current(), message, ActiveMQMessageTextMapGetter.INSTANCE);
+
+ String destinationName = message.getDestination().getPhysicalName();
+ SpanBuilder spanBuilder = newSpan(destinationName + SPAN_SUFFIX_DELIVER, SpanKind.CONSUMER,
+ destinationName, OPERATION_DELIVER)
+ .setParent(extractedContext);
+
+ if (message.getMessageId() != null) {
+ spanBuilder.setAttribute(ATTR_MESSAGE_ID, message.getMessageId().toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+ dispatchSpans.put(messageDispatch, span);
+ } catch (Exception e) {
+ LOG.warn("Failed to create deliver span", e);
+ }
+ }
+ super.preProcessDispatch(messageDispatch);
+ }
+
+ @Override
+ public void postProcessDispatch(MessageDispatch messageDispatch) {
+ if (messageDispatch != null) {
+ Span span = dispatchSpans.remove(messageDispatch);
+ if (span != null) {
+ span.end();
+ }
+ }
+ super.postProcessDispatch(messageDispatch);
+ }
+
+ @Override
+ public void acknowledge(ConsumerBrokerExchange consumerExchange, MessageAck ack) throws Exception {
+ if (!traceAcknowledge) {
+ super.acknowledge(consumerExchange, ack);
+ return;
+ }
+
+ String destinationName = ack.getDestination() != null ? ack.getDestination().getPhysicalName() : UNKNOWN_DESTINATION;
+ SpanBuilder spanBuilder = newSpan(destinationName + SPAN_SUFFIX_ACK, SpanKind.INTERNAL,
+ destinationName, OPERATION_SETTLE);
+
+ if (ack.getLastMessageId() != null) {
+ spanBuilder.setAttribute(ATTR_MESSAGE_ID, ack.getLastMessageId().toString());
+ }
+ if (consumerExchange.getConnectionContext() != null
+ && consumerExchange.getConnectionContext().getClientId() != null) {
+ spanBuilder.setAttribute(ATTR_CLIENT_ID, consumerExchange.getConnectionContext().getClientId());
+ }
+
+ Span span = spanBuilder.startSpan();
+ try {
+ super.acknowledge(consumerExchange, ack);
+ } catch (Exception e) {
+ span.setStatus(StatusCode.ERROR, String.valueOf(e.getMessage()));
+ span.recordException(e);
+ throw e;
+ } finally {
+ span.end();
+ }
+ }
+}
diff --git a/activemq-opentelemetry/src/test/java/org/apache/activemq/broker/util/opentelemetry/OpenTelemetryBrokerPluginTest.java b/activemq-opentelemetry/src/test/java/org/apache/activemq/broker/util/opentelemetry/OpenTelemetryBrokerPluginTest.java
new file mode 100644
index 0000000000..370fb6abb0
--- /dev/null
+++ b/activemq-opentelemetry/src/test/java/org/apache/activemq/broker/util/opentelemetry/OpenTelemetryBrokerPluginTest.java
@@ -0,0 +1,269 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.broker.util.opentelemetry;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.List;
+
+import jakarta.jms.Connection;
+import jakarta.jms.DeliveryMode;
+import jakarta.jms.MessageConsumer;
+import jakarta.jms.MessageProducer;
+import jakarta.jms.Session;
+import jakarta.jms.TextMessage;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerPlugin;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.TransportConnector;
+import org.apache.activemq.util.Wait;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class OpenTelemetryBrokerPluginTest {
+
+ private BrokerService broker;
+ private TransportConnector connector;
+ private InMemorySpanExporter spanExporter;
+ private Connection connection;
+ private Session session;
+ private final String queueName = "TEST.OTEL";
+
+ @Before
+ public void setUp() throws Exception {
+ GlobalOpenTelemetry.resetForTest();
+
+ spanExporter = InMemorySpanExporter.create();
+ SdkTracerProvider tracerProvider = SdkTracerProvider.builder()
+ .addSpanProcessor(SimpleSpanProcessor.create(spanExporter))
+ .build();
+
+ OpenTelemetrySdk.builder()
+ .setTracerProvider(tracerProvider)
+ .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
+ .buildAndRegisterGlobal();
+
+ OpenTelemetryBrokerPlugin plugin = new OpenTelemetryBrokerPlugin();
+
+ broker = new BrokerService();
+ broker.setBrokerName("otelTestBroker");
+ broker.setPersistent(false);
+ broker.setUseJmx(false);
+ broker.setPlugins(new BrokerPlugin[]{plugin});
+ connector = broker.addConnector("tcp://localhost:0");
+ broker.start();
+ broker.waitUntilStarted();
+
+ connection = new ActiveMQConnectionFactory(connector.getConnectUri()).createConnection();
+ connection.start();
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (session != null) session.close();
+ if (connection != null) connection.close();
+ if (broker != null) {
+ broker.stop();
+ broker.waitUntilStopped();
+ }
+ GlobalOpenTelemetry.resetForTest();
+ }
+
+ @Test
+ public void testPublishSpanCreatedOnSend() throws Exception {
+ MessageProducer producer = session.createProducer(session.createQueue(queueName));
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+
+ TextMessage message = session.createTextMessage("test");
+ producer.send(message);
+ producer.close();
+
+ assertTrue("Expected at least one span", Wait.waitFor(() -> spanExporter.getFinishedSpanItems().size() >= 1));
+
+ List<SpanData> spans = spanExporter.getFinishedSpanItems();
+ SpanData publishSpan = findSpan(spans, queueName + " publish");
+ assertNotNull("Publish span should exist", publishSpan);
+ assertEquals(SpanKind.PRODUCER, publishSpan.getKind());
+ assertEquals("activemq", publishSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.system")));
+ assertEquals(queueName, publishSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.destination.name")));
+ assertEquals("publish", publishSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.operation.type")));
+ }
+
+ @Test
+ public void testTraceContextPropagatedInMessageProperties() throws Exception {
+ MessageProducer producer = session.createProducer(session.createQueue(queueName));
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+
+ TextMessage message = session.createTextMessage("propagation test");
+ producer.send(message);
+ producer.close();
+
+ MessageConsumer consumer = session.createConsumer(session.createQueue(queueName));
+ jakarta.jms.Message received = consumer.receive(5000);
+ assertNotNull("Should receive message", received);
+
+ String traceparent = received.getStringProperty("traceparent");
+ assertNotNull("traceparent property should be set", traceparent);
+ assertTrue("traceparent should follow W3C format", traceparent.startsWith("00-"));
+
+ consumer.close();
+ }
+
+ @Test
+ public void testDeliverSpanCreatedOnDispatch() throws Exception {
+ MessageConsumer consumer = session.createConsumer(session.createQueue(queueName));
+
+ MessageProducer producer = session.createProducer(session.createQueue(queueName));
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ producer.send(session.createTextMessage("deliver test"));
+ producer.close();
+
+ jakarta.jms.Message received = consumer.receive(5000);
+ assertNotNull("Should receive message", received);
+ consumer.close();
+
+ assertTrue("Expected deliver span", Wait.waitFor(() -> findSpan(spanExporter.getFinishedSpanItems(), queueName + " deliver") != null));
+
+ List<SpanData> spans = spanExporter.getFinishedSpanItems();
+ SpanData deliverSpan = findSpan(spans, queueName + " deliver");
+ assertNotNull("Deliver span should exist", deliverSpan);
+ assertEquals(SpanKind.CONSUMER, deliverSpan.getKind());
+ assertEquals("activemq", deliverSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.system")));
+ assertEquals("deliver", deliverSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.operation.type")));
+ }
+
+ @Test
+ public void testAckSpanCreatedOnAcknowledge() throws Exception {
+ MessageConsumer consumer = session.createConsumer(session.createQueue(queueName));
+
+ MessageProducer producer = session.createProducer(session.createQueue(queueName));
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ producer.send(session.createTextMessage("ack test"));
+ producer.close();
+
+ jakarta.jms.Message received = consumer.receive(5000);
+ assertNotNull("Should receive message", received);
+ consumer.close();
+
+ assertTrue("Expected ack span", Wait.waitFor(() -> findSpan(spanExporter.getFinishedSpanItems(), queueName + " ack") != null));
+
+ List<SpanData> spans = spanExporter.getFinishedSpanItems();
+ SpanData ackSpan = findSpan(spans, queueName + " ack");
+ assertNotNull("Ack span should exist", ackSpan);
+ assertEquals(SpanKind.INTERNAL, ackSpan.getKind());
+ assertEquals("activemq", ackSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.system")));
+ assertEquals("settle", ackSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.operation.type")));
+ }
+
+ @Test
+ public void testDisabledProducerTracing() throws Exception {
+ connection.close();
+ broker.stop();
+ broker.waitUntilStopped();
+ spanExporter.reset();
+
+ OpenTelemetryBrokerPlugin plugin = new OpenTelemetryBrokerPlugin();
+ plugin.setTraceProducer(false);
+ plugin.setTraceConsumer(false);
+ plugin.setTraceAcknowledge(false);
+
+ broker = new BrokerService();
+ broker.setBrokerName("otelTestBroker2");
+ broker.setPersistent(false);
+ broker.setUseJmx(false);
+ broker.setPlugins(new BrokerPlugin[]{plugin});
+ connector = broker.addConnector("tcp://localhost:0");
+ broker.start();
+ broker.waitUntilStarted();
+
+ connection = new ActiveMQConnectionFactory(connector.getConnectUri()).createConnection();
+ connection.start();
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ MessageProducer producer = session.createProducer(session.createQueue(queueName));
+ producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ producer.send(session.createTextMessage("no trace"));
+ producer.close();
+
+ MessageConsumer consumer = session.createConsumer(session.createQueue(queueName));
+ consumer.receive(5000);
+ consumer.close();
+
+ // Brief wait then assert nothing was traced
+ Wait.waitFor(() -> false, 500, 100);
+
+ List<SpanData> spans = spanExporter.getFinishedSpanItems();
+ assertFalse("Should have no publish spans when tracing disabled",
+ spans.stream().anyMatch(s -> s.getName().contains("publish")));
+ assertFalse("Should have no deliver spans when tracing disabled",
+ spans.stream().anyMatch(s -> s.getName().contains("deliver")));
+ assertFalse("Should have no ack spans when tracing disabled",
+ spans.stream().anyMatch(s -> s.getName().contains("ack")));
+ }
+
+ @Test
+ public void testMessagingSystemConfigurable() throws Exception {
+ broker.stop();
+ broker.waitUntilStopped();
+
+ OpenTelemetryBrokerPlugin plugin = new OpenTelemetryBrokerPlugin();
+ plugin.setMessagingSystem("my-broker");
+ broker = new BrokerService();
+ broker.setBrokerName("otelTestBroker2");
+ broker.setPersistent(false);
+ broker.setUseJmx(false);
+ broker.setPlugins(new BrokerPlugin[]{plugin});
+ connector = broker.addConnector("tcp://localhost:0");
+ broker.start();
+ broker.waitUntilStarted();
+
+ Connection c = new ActiveMQConnectionFactory(connector.getConnectUri()).createConnection();
+ try {
+ c.start();
+ Session s = c.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ s.createProducer(s.createQueue(queueName)).send(s.createTextMessage("test"));
+ } finally {
+ c.close();
+ }
+
+ assertTrue("Expected publish span", Wait.waitFor(() -> findSpan(spanExporter.getFinishedSpanItems(), queueName + " publish") != null));
+ SpanData publishSpan = findSpan(spanExporter.getFinishedSpanItems(), queueName + " publish");
+ assertEquals("my-broker", publishSpan.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("messaging.system")));
+ }
+
+ private SpanData findSpan(List<SpanData> spans, String name) {
+ return spans.stream()
+ .filter(s -> s.getName().equals(name))
+ .findFirst()
+ .orElse(null);
+ }
+}
diff --git a/activemq-spring/pom.xml b/activemq-spring/pom.xml
index 9d6d1898e4..68e82096f2 100644
--- a/activemq-spring/pom.xml
+++ b/activemq-spring/pom.xml
@@ -108,6 +108,11 @@
<artifactId>activemq-stomp</artifactId>
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-opentelemetry</artifactId>
+ <optional>true</optional>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>activemq-pool</artifactId>
@@ -234,6 +239,7 @@
<include>${basedir}/../activemq-mqtt/src/main/java</include>
<include>${basedir}/../activemq-stomp/src/main/java</include>
<include>${basedir}/../activemq-runtime-config/src/main/java</include>
+ <include>${basedir}/../activemq-opentelemetry/src/main/java</include>
</includes>
<strictXsdOrder>false</strictXsdOrder>
<namespace>http://activemq.apache.org/schema/core</namespace>
diff --git a/assembly/pom.xml b/assembly/pom.xml
index aa46add1c1..3769b54a98 100644
--- a/assembly/pom.xml
+++ b/assembly/pom.xml
@@ -136,6 +136,10 @@
<groupId>${project.groupId}</groupId>
<artifactId>activemq-shiro</artifactId>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>activemq-opentelemetry</artifactId>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>activemq-spring</artifactId>
diff --git a/assembly/src/release/conf/activemq.xml b/assembly/src/release/conf/activemq.xml
index 551bdb4a15..e483181d9c 100644
--- a/assembly/src/release/conf/activemq.xml
+++ b/assembly/src/release/conf/activemq.xml
@@ -160,6 +160,15 @@
</systemUsage>
</systemUsage>
+ <!--
+ The plugins element can be used to configure broker plugins.
+ For example, to enable OpenTelemetry distributed tracing, uncomment the following:
+
+ <plugins>
+ <openTelemetryPlugin traceProducer="true" traceConsumer="true" traceAcknowledge="true"/>
+ </plugins>
+ -->
+
<!--
The transport connectors expose ActiveMQ over a given protocol to
clients and other brokers. For more information, see:
diff --git a/assembly/src/release/examples/conf/activemq-opentelemetry.xml b/assembly/src/release/examples/conf/activemq-opentelemetry.xml
new file mode 100644
index 0000000000..25028e7dd3
--- /dev/null
+++ b/assembly/src/release/examples/conf/activemq-opentelemetry.xml
@@ -0,0 +1,118 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+-->
+<!--
+ ActiveMQ with OpenTelemetry distributed tracing
+
+ This configuration enables OpenTelemetry tracing for the broker.
+ The plugin traces message send, dispatch, and acknowledge operations
+ using the OpenTelemetry API with W3C TraceContext propagation.
+
+ To use this, you need to provide an OpenTelemetry SDK implementation
+ and exporter at runtime (e.g., opentelemetry-sdk, opentelemetry-exporter-otlp).
+
+ You can configure the OpenTelemetry SDK via environment variables:
+ OTEL_SERVICE_NAME=activemq
+ OTEL_TRACES_EXPORTER=otlp
+ OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
+
+ Or via the Java agent:
+ ACTIVEMQ_OPTS="-javaagent:/path/to/opentelemetry-javaagent.jar"
+
+ To run ActiveMQ with this configuration:
+ $ bin/activemq console xbean:examples/conf/activemq-opentelemetry.xml
+-->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+ <!-- Allows us to use system properties as variables in this configuration file -->
+ <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
+ <property name="locations">
+ <value>file:${activemq.conf}/credentials.properties</value>
+ </property>
+ </bean>
+
+ <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}">
+
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry topic=">" >
+ <pendingMessageLimitStrategy>
+ <constantPendingMessageLimitStrategy limit="1000"/>
+ </pendingMessageLimitStrategy>
+ </policyEntry>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+
+ <!--
+ OpenTelemetry tracing plugin.
+
+ Attributes:
+ traceProducer - trace message send operations (default: true)
+ traceConsumer - trace message dispatch/deliver operations (default: true)
+ traceAcknowledge - trace message acknowledgment operations (default: true)
+ messagingSystem - value of the messaging.system attribute (default: activemq)
+
+ Span details:
+ send() -> "{destination} publish" (PRODUCER span)
+ preProcessDispatch() -> "{destination} deliver" (CONSUMER span)
+ acknowledge() -> "{destination} ack" (INTERNAL span, operation type "settle")
+
+ Each span includes standard OTel messaging attributes:
+ messaging.system, messaging.destination.name, messaging.operation.type,
+ messaging.message.id, messaging.client.id
+ -->
+ <plugins>
+ <openTelemetryPlugin traceProducer="true" traceConsumer="true" traceAcknowledge="true"/>
+ </plugins>
+
+ <persistenceAdapter>
+ <kahaDB directory="${activemq.data}/kahadb"/>
+ </persistenceAdapter>
+
+ <systemUsage>
+ <systemUsage>
+ <memoryUsage>
+ <memoryUsage percentOfJvmHeap="70" />
+ </memoryUsage>
+ <storeUsage>
+ <storeUsage limit="100 gb"/>
+ </storeUsage>
+ <tempUsage>
+ <tempUsage limit="50 gb"/>
+ </tempUsage>
+ </systemUsage>
+ </systemUsage>
+
+ <transportConnectors>
+ <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
+ <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
+ <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
+ <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
+ </transportConnectors>
+
+ <shutdownHooks>
+ <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
+ </shutdownHooks>
+
+ </broker>
+
+</beans>
diff --git a/bom/pom.xml b/bom/pom.xml
index 321244473f..eae60ac52d 100644
--- a/bom/pom.xml
+++ b/bom/pom.xml
@@ -147,6 +147,11 @@
<artifactId>activemq-shiro</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-opentelemetry</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-spring</artifactId>
diff --git a/pom.xml b/pom.xml
index 46b33149f2..ccb658a79c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -100,6 +100,7 @@
<netty-version>4.2.19.Final</netty-version>
<rome-version>2.1.0</rome-version>
<shiro-version>2.2.1</shiro-version>
+ <opentelemetry-version>1.43.0</opentelemetry-version>
<slf4j-version>2.0.20</slf4j-version>
<spring-version>7.0.9</spring-version>
<spring-version-range>[7,8)</spring-version-range>
@@ -217,6 +218,7 @@
<module>activemq-rar</module>
<module>activemq-run</module>
<module>activemq-shiro</module>
+ <module>activemq-opentelemetry</module>
<module>activemq-spring</module>
<module>activemq-runtime-config</module>
<module>activemq-tooling</module>
@@ -349,6 +351,11 @@
<artifactId>activemq-shiro</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.activemq</groupId>
+ <artifactId>activemq-opentelemetry</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-spring</artifactId>
@@ -667,6 +674,15 @@
<optional>true</optional>
</dependency>
+ <!-- Optional OpenTelemetry Support -->
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-bom</artifactId>
+ <version>${opentelemetry-version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+
<!-- Optional Spring Support -->
<dependency>
<groupId>org.springframework</groupId>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact