You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by pc...@apache.org on 2025/08/25 08:45:41 UTC
(camel) 01/03: feat(components): micrometer observability telemetry impl
This is an automated email from the ASF dual-hosted git repository.
pcongiusti pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 553ddd5f4a68a201189f310fb7f2d90a339a4a07
Author: Pasquale Congiusti <pa...@gmail.com>
AuthorDate: Mon May 12 15:12:09 2025 +0200
feat(components): micrometer observability telemetry impl
Closes CAMEL-21869
---
catalog/camel-allcomponents/pom.xml | 5 +
components/camel-micrometer-observability/pom.xml | 130 +++++++++++
.../MicrometerObservabilityTracerConfigurer.java | 69 ++++++
...ter.observability.MicrometerObservabilityTracer | 2 +
.../apache/camel/micrometer-observability-tracer | 2 +
.../services/org/apache/camel/other.properties | 7 +
.../resources/micrometer-observability.json | 15 ++
.../src/main/docs/micrometer-observability.adoc | 69 ++++++
.../MicrometerObservabilitySpanAdapter.java | 83 +++++++
.../MicrometerObservabilityTracer.java | 160 +++++++++++++
.../micrometer/observability/AsyncCXFTest.java | 171 ++++++++++++++
.../micrometer/observability/AsyncDirectTest.java | 146 ++++++++++++
.../micrometer/observability/AsyncWiretapTest.java | 149 ++++++++++++
.../observability/CamelOpenTelemetryExtension.java | 257 +++++++++++++++++++++
.../observability/DisableEndpointTest.java | 80 +++++++
.../observability/EnableProcessorsTest.java | 109 +++++++++
.../MicrometerObservabilityTrace.java | 53 +++++
...rObservabilityTracerPropagationTestSupport.java | 77 ++++++
.../MicrometerObservabilityTracerTest.java | 110 +++++++++
.../MicrometerObservabilityTracerTestSupport.java | 70 ++++++
.../observability/SpanPropagationTest.java | 82 +++++++
.../src/test/resources/log4j2.properties | 30 +++
parent/pom.xml | 15 ++
23 files changed, 1891 insertions(+)
diff --git a/catalog/camel-allcomponents/pom.xml b/catalog/camel-allcomponents/pom.xml
index 279c7f1d08f..6a4910f05ed 100644
--- a/catalog/camel-allcomponents/pom.xml
+++ b/catalog/camel-allcomponents/pom.xml
@@ -1271,6 +1271,11 @@
<artifactId>camel-micrometer</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-micrometer-observability</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-micrometer-prometheus</artifactId>
diff --git a/components/camel-micrometer-observability/pom.xml b/components/camel-micrometer-observability/pom.xml
new file mode 100644
index 00000000000..a03b8e9fc05
--- /dev/null
+++ b/components/camel-micrometer-observability/pom.xml
@@ -0,0 +1,130 @@
+<?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/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>4.12.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-micrometer-observability</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: Micrometer :: Observability 2</name>
+ <description>Micrometer Observability implementation of Camel Telemetry</description>
+
+ <properties>
+ <firstVersion>4.15.0</firstVersion>
+ <label>monitoring,microservice</label>
+ <title>Micrometer Observability 2</title>
+ <supportLevel>Preview</supportLevel>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-telemetry</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-tracing</artifactId>
+ </dependency>
+ <!-- Used as default development tracer option -->
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-tracing-test</artifactId>
+ </dependency>
+
+
+ <!-- START Required to validate a real context propagation implementation -->
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-tracing-bridge-otel</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk</artifactId>
+ <version>${opentelemetry-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk-testing</artifactId>
+ <version>${opentelemetry-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- END Required to validate a real context propagation implementation -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring-junit5</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- Required to test CXF async -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-cxf-rest</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-cxf-common</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.cxf</groupId>
+ <artifactId>cxf-rt-transports-http-undertow</artifactId>
+ <version>${cxf-version}</version>
+ <scope>test</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.undertow</groupId>
+ <artifactId>undertow-servlet</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>io.undertow</groupId>
+ <artifactId>undertow-servlet-jakarta</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>io.undertow</groupId>
+ <artifactId>undertow-core</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-undertow</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/components/camel-micrometer-observability/src/generated/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerConfigurer.java b/components/camel-micrometer-observability/src/generated/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerConfigurer.java
new file mode 100644
index 00000000000..014c73913f0
--- /dev/null
+++ b/components/camel-micrometer-observability/src/generated/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerConfigurer.java
@@ -0,0 +1,69 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.micrometer.observability;
+
+import javax.annotation.processing.Generated;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.micrometer.observability.MicrometerObservabilityTracer;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@Generated("org.apache.camel.maven.packaging.GenerateConfigurerMojo")
+@SuppressWarnings("unchecked")
+public class MicrometerObservabilityTracerConfigurer extends org.apache.camel.support.component.PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ org.apache.camel.micrometer.observability.MicrometerObservabilityTracer target = (org.apache.camel.micrometer.observability.MicrometerObservabilityTracer) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "camelcontext":
+ case "camelContext": target.setCamelContext(property(camelContext, org.apache.camel.CamelContext.class, value)); return true;
+ case "excludepatterns":
+ case "excludePatterns": target.setExcludePatterns(property(camelContext, java.lang.String.class, value)); return true;
+ case "spanlifecyclemanager":
+ case "spanLifecycleManager": target.setSpanLifecycleManager(property(camelContext, org.apache.camel.telemetry.SpanLifecycleManager.class, value)); return true;
+ case "traceprocessors":
+ case "traceProcessors": target.setTraceProcessors(property(camelContext, boolean.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "camelcontext":
+ case "camelContext": return org.apache.camel.CamelContext.class;
+ case "excludepatterns":
+ case "excludePatterns": return java.lang.String.class;
+ case "spanlifecyclemanager":
+ case "spanLifecycleManager": return org.apache.camel.telemetry.SpanLifecycleManager.class;
+ case "traceprocessors":
+ case "traceProcessors": return boolean.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ org.apache.camel.micrometer.observability.MicrometerObservabilityTracer target = (org.apache.camel.micrometer.observability.MicrometerObservabilityTracer) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "camelcontext":
+ case "camelContext": return target.getCamelContext();
+ case "excludepatterns":
+ case "excludePatterns": return target.getExcludePatterns();
+ case "spanlifecyclemanager":
+ case "spanLifecycleManager": return target.getSpanLifecycleManager();
+ case "traceprocessors":
+ case "traceProcessors": return target.isTraceProcessors();
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.micrometer.observability.MicrometerObservabilityTracer b/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.micrometer.observability.MicrometerObservabilityTracer
new file mode 100644
index 00000000000..878f9e382dc
--- /dev/null
+++ b/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/configurer/org.apache.camel.micrometer.observability.MicrometerObservabilityTracer
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.micrometer.observability.MicrometerObservabilityTracerConfigurer
diff --git a/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/micrometer-observability-tracer b/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/micrometer-observability-tracer
new file mode 100644
index 00000000000..e2c07c68f1d
--- /dev/null
+++ b/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/micrometer-observability-tracer
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.micrometer.observability.MicrometerObservabilityTracer
diff --git a/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/other.properties b/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/other.properties
new file mode 100644
index 00000000000..c2cf3f73b4f
--- /dev/null
+++ b/components/camel-micrometer-observability/src/generated/resources/META-INF/services/org/apache/camel/other.properties
@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+name=micrometer-observability
+groupId=org.apache.camel
+artifactId=camel-micrometer-observability
+version=4.12.0-SNAPSHOT
+projectName=Camel :: Micrometer :: Observability
+projectDescription=Micrometer Observability implementation of Camel Telemetry
diff --git a/components/camel-micrometer-observability/src/generated/resources/micrometer-observability.json b/components/camel-micrometer-observability/src/generated/resources/micrometer-observability.json
new file mode 100644
index 00000000000..afce5fe6f33
--- /dev/null
+++ b/components/camel-micrometer-observability/src/generated/resources/micrometer-observability.json
@@ -0,0 +1,15 @@
+{
+ "other": {
+ "kind": "other",
+ "name": "micrometer-observability",
+ "title": "Micrometer Observability",
+ "description": "Micrometer Observability implementation of Camel Telemetry",
+ "deprecated": false,
+ "firstVersion": "4.12.0",
+ "label": "monitoring,microservice",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-micrometer-observability",
+ "version": "4.12.0-SNAPSHOT"
+ }
+}
diff --git a/components/camel-micrometer-observability/src/main/docs/micrometer-observability.adoc b/components/camel-micrometer-observability/src/main/docs/micrometer-observability.adoc
new file mode 100644
index 00000000000..e2d8e70461e
--- /dev/null
+++ b/components/camel-micrometer-observability/src/main/docs/micrometer-observability.adoc
@@ -0,0 +1,69 @@
+= Telemetry Dev Component
+:doctitle: Micrometer Observability
+:shortname: micrometed-observability
+:artifactid: camel-micrometed-observability
+:description: Implementation of Camel Micrometer Observability based on the Camel Telemetry spec
+:since: 4.15
+:supportlevel: Preview
+:tabs-sync-option:
+
+*Since Camel {since}*
+
+This module is the implementation of the common `camel-telemetry` interface based on https://micrometer.io/[Micrometer Application Observability] technology. The name used here as `Micrometer Observability 2` is done to distinguish on the existing `camel-observation` component which was based on an older Camel tracing and Micrometer specification. You're invited to start replacing the older `camel-observation` with this one instead as it may become the default component in future version [...]
+
+NOTE: this component has slight differences compared to the `camel-observation` and is meant to solve a few inconsistencies identified.
+
+== Configuration
+
+The configuration properties for the component are:
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|=======================================================================
+|Option |Default |Description
+|`excludePatterns` | | Sets exclude pattern(s) that will disable tracing for Camel
+messages that matches the pattern. The content is a Set<String> where the key is a pattern. The pattern
+uses the rules from Intercept.
+|`traceProcessors` | `false` | Setting this to true will create new OpenTelemetry Spans for each Camel Processors.
+Use the excludePattern property to filter out Processors
+|=======================================================================
+
+== Tracing format
+
+The component provides a few different traces representation formats you can use, depending on the scope of your telemetry or the integration with any third party telemetry collectors.
+
+=== Tree
+
+As the component is mainly targeted to have a quick visualization of a running application, then, the `tree` format is the recommended one. The output will be like the following:
+
+```
+| d6c36bc6f3374ed0bd6edb638535f130
+├── | timer://start (camel-timer) [2030 millis] <--
+ ├── | process (to1-to) [2021 millis]
+ ├── | direct://new (camel-direct) [2020 millis] -->
+ ├── | direct://new (camel-direct) [2019 millis] <--
+ ├── | process (delay1-delay) [2002 millis]
+ ├── | process (log2-log) [2 millis]
+ └── | process (to3-to) [3 millis]
+ └── | log://new (camel-log) [2 millis] -->
+ └── | process (to4-to) [1 millis]
+ └── | mock://end (camel-mock) [0 millis] -->
+ ├── | process (log1-log) [0 millis]
+ └── | process (to2-to) [2 millis]
+ └── | log://info (camel-log) [1 millis] -->
+```
+
+This is a quick visual tree Trace representation of a given execution of a Camel route. You can quickly verify the Camel URI, the component and the time spent to execute each Span.
+
+NOTE: the --> and <-- represents an outgoing or incoming message.
+
+=== Json
+
+This format is suitable if you need to integrate with any third party. The output of each trace will come in the following format:
+
+```json
+{"traceId":"4ee161eb055644fca713f4c02ab2fb50","spans":[{"logEntries":[],"traceid":"4ee161eb055644fca713f4c02ab2fb50","spanid":"8785debcdc09409abeed65d08aaaca1b","exchangeId":"D1B1814C509A376-0000000000000000","op":"EVENT_RECEIVED","component":"camel-timer","url.path":"start","initTimestamp":"4085653460994","camel.uri":"timer://start","url.scheme":"timer","endTimestamp":"4087698388888","isDone":"true"},{"logEntries":[],"traceid":"4ee161eb055644fca713f4c02ab2fb50","spanid":"51cc3929d34e452 [...]
+```
+
+=== Default
+
+Default is a very simple serialization into the default Java object `toString()` representation.
diff --git a/components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilitySpanAdapter.java b/components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilitySpanAdapter.java
new file mode 100644
index 00000000000..44bdff3e920
--- /dev/null
+++ b/components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilitySpanAdapter.java
@@ -0,0 +1,83 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.util.Map;
+
+import org.apache.camel.telemetry.Span;
+
+public class MicrometerObservabilitySpanAdapter implements Span {
+
+ private static final String DEFAULT_EVENT_NAME = "log";
+
+ private final io.micrometer.tracing.Span span;
+
+ public MicrometerObservabilitySpanAdapter(io.micrometer.tracing.Span span) {
+ this.span = span;
+ }
+
+ @Override
+ public void log(Map<String, String> fields) {
+ String event = fields.get("event");
+ if ("error".equalsIgnoreCase(event)) {
+ setError(true);
+ } else if (fields.get("message") != null) {
+ // this is really the only expected event for logging
+ this.span.event("message=" + fields.get("message"));
+ } else {
+ this.span.event(DEFAULT_EVENT_NAME);
+ }
+ }
+
+ @Override
+ public void setComponent(String component) {
+ this.span.tag("component", component);
+ }
+
+ @Override
+ public void setError(boolean isError) {
+ this.span.tag("error", isError);
+ }
+
+ @Override
+ public void setTag(String key, String value) {
+ this.span.tag(key, value);
+ }
+
+ protected io.micrometer.tracing.Span getSpan() {
+ return this.span;
+ }
+
+ protected void activate() {
+ this.span.start();
+ }
+
+ protected void close() {
+ this.span.end();
+ }
+
+ protected void deactivate() {
+
+ }
+
+ @Override
+ public String toString() {
+ return "MicrometerObservabilitySpanAdapter [traceId=" +
+ span.context().traceId() + " spanId=" + span.context().spanId() + "]";
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracer.java b/components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracer.java
new file mode 100644
index 00000000000..fa6bba48ab7
--- /dev/null
+++ b/components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracer.java
@@ -0,0 +1,160 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import io.micrometer.observation.ObservationHandler;
+import io.micrometer.observation.ObservationRegistry;
+import io.micrometer.tracing.Span.Builder;
+import io.micrometer.tracing.Tracer;
+import io.micrometer.tracing.handler.DefaultTracingObservationHandler;
+import io.micrometer.tracing.handler.PropagatingReceiverTracingObservationHandler;
+import io.micrometer.tracing.handler.PropagatingSenderTracingObservationHandler;
+import io.micrometer.tracing.propagation.Propagator;
+import io.micrometer.tracing.test.simple.SimpleTracer;
+import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.spi.Configurer;
+import org.apache.camel.spi.annotations.JdkService;
+import org.apache.camel.support.CamelContextHelper;
+import org.apache.camel.telemetry.Span;
+import org.apache.camel.telemetry.SpanContextPropagationExtractor;
+import org.apache.camel.telemetry.SpanContextPropagationInjector;
+import org.apache.camel.telemetry.SpanLifecycleManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@JdkService("micrometer-observability-tracer")
+@Configurer
+@ManagedResource(description = "MicrometerObservabilityTracer")
+public class MicrometerObservabilityTracer extends org.apache.camel.telemetry.Tracer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MicrometerObservabilityTracer.class);
+
+ private Tracer tracer;
+ private ObservationRegistry observationRegistry;
+ private Propagator propagator;
+
+ @Override
+ protected void initTracer() {
+ if (tracer == null) {
+ tracer = CamelContextHelper.findSingleByType(getCamelContext(), Tracer.class);
+ }
+ if (tracer == null) {
+ tracer = new SimpleTracer();
+ LOG.warn("No tracer was provided. A default inmemory tracer is used. " +
+ "This can be useful for development only, avoid this in a production environment.");
+ }
+ if (observationRegistry == null) {
+ observationRegistry = CamelContextHelper.findSingleByType(getCamelContext(), ObservationRegistry.class);
+ }
+ if (observationRegistry == null) {
+ MeterRegistry meterRegistry = new SimpleMeterRegistry();
+ this.observationRegistry = ObservationRegistry.create();
+ this.observationRegistry.observationConfig().observationHandler(
+ new DefaultMeterObservationHandler(meterRegistry));
+ LOG.warn("No observation registry was provided. A default inmemory observation registry is used. " +
+ "This can be useful for development only, avoid this in a production environment.");
+ }
+
+ if (propagator == null) {
+ propagator = CamelContextHelper.findSingleByType(getCamelContext(), Propagator.class);
+ }
+ if (propagator == null) {
+ propagator = Propagator.NOOP;
+ LOG.warn("No propagator was provided. A NOOP implementation is used, you won't be able to trace " +
+ "upstream activity. " +
+ "This can be useful for development only, avoid this in a production environment.");
+ }
+
+ this.setSpanLifecycleManager(new MicrometerObservabilitySpanLifecycleManager());
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ LOG.info("Micrometer Observability enabled");
+ }
+
+ private class MicrometerObservabilitySpanLifecycleManager implements SpanLifecycleManager {
+
+ private MicrometerObservabilitySpanLifecycleManager() {
+ observationRegistry.observationConfig().observationHandler(
+ new ObservationHandler.FirstMatchingCompositeObservationHandler(
+ new PropagatingSenderTracingObservationHandler<>(tracer, propagator),
+ new PropagatingReceiverTracingObservationHandler<>(tracer, propagator),
+ new DefaultTracingObservationHandler(tracer)));
+ }
+
+ @Override
+ public Span create(String spanName, Span parent, SpanContextPropagationExtractor extractor) {
+ io.micrometer.tracing.Span span;
+ if (parent != null) {
+ MicrometerObservabilitySpanAdapter microObsParentSpan = (MicrometerObservabilitySpanAdapter) parent;
+ span = tracer.nextSpan(microObsParentSpan.getSpan());
+ } else if (extractor.get("traceparent") != null || extractor.get("X-B3-TraceId") != null) {
+ /*
+ * This part is a bit tricky. We need to verify if the extractor
+ * (ie, the Camel Exchange) holds a propagated parent.
+ * As the micrometer-observability is technology agnostic, we need to check against
+ * the available implementations (Opentelemetry and Zipkin at the moment of writing this comment).
+ * TODO: we could do this configurable if it is required.
+ */
+ Builder builder = propagator.extract(extractor, (carrier, key) -> {
+ return extractor.get(key) == null ? null : (String) extractor.get(key);
+ });
+
+ span = builder.start();
+ } else {
+ span = tracer.nextSpan();
+ }
+
+ span.name(spanName);
+
+ return new MicrometerObservabilitySpanAdapter(span);
+ }
+
+ @Override
+ public void activate(Span span) {
+ MicrometerObservabilitySpanAdapter microObsSpan = (MicrometerObservabilitySpanAdapter) span;
+ microObsSpan.activate();
+ }
+
+ @Override
+ public void close(Span span) {
+ MicrometerObservabilitySpanAdapter microObsSpan = (MicrometerObservabilitySpanAdapter) span;
+ microObsSpan.close();
+ }
+
+ @Override
+ public void deactivate(Span span) {
+ MicrometerObservabilitySpanAdapter microObsSpan = (MicrometerObservabilitySpanAdapter) span;
+ microObsSpan.deactivate();
+ }
+
+ @Override
+ public void inject(Span span, SpanContextPropagationInjector injector) {
+ MicrometerObservabilitySpanAdapter microObsSpan = (MicrometerObservabilitySpanAdapter) span;
+ propagator.inject(
+ microObsSpan.getSpan().context(),
+ injector,
+ (carrier, key, value) -> carrier.put(key, value));
+ }
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncCXFTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncCXFTest.java
new file mode 100644
index 00000000000..b77e25a0b67
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncCXFTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.micrometer.observability.CamelOpenTelemetryExtension.OtelTrace;
+import org.apache.camel.telemetry.Op;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/*
+ * AsyncCXFTest tests the execution of CXF async which was reported as a potential candidate to
+ * inconsistent Span creation in async mode.
+ *
+ * This test is also special as it requires a different setting to inherit the
+ * Opentelemetry propagation mechanism.
+ */
+public class AsyncCXFTest extends MicrometerObservabilityTracerPropagationTestSupport {
+
+ private static int cxfPort = AvailablePortFinder.getNextRandomAvailable();
+
+ @Test
+ void testRouteMultipleRequests() throws InterruptedException, IOException {
+ int j = 10;
+ MockEndpoint mock = getMockEndpoint("mock:end");
+ mock.expectedMessageCount(j);
+ mock.setAssertPeriod(5000);
+ for (int i = 0; i < j; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ mock.assertIsSatisfied(1000);
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(j, traces.size());
+ // Each trace should have the same structure
+ for (OtelTrace trace : traces.values()) {
+ checkTrace(trace);
+ }
+ }
+
+ private void checkTrace(OtelTrace trace) {
+ List<SpanData> spans = trace.getSpans();
+ assertEquals(8, spans.size());
+ SpanData testProducer = getSpan(spans, "direct://start", Op.EVENT_SENT);
+ SpanData direct = getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ SpanData directSendTo = getSpan(spans, "direct://send", Op.EVENT_SENT);
+ SpanData directSendFrom = getSpan(spans, "direct://send", Op.EVENT_RECEIVED);
+ SpanData cxfRs = getSpan(
+ spans,
+ "cxfrs://http://localhost:" + cxfPort + "/rest/helloservice/sayHello?synchronous=false",
+ Op.EVENT_SENT);
+ SpanData rest = getSpan(
+ spans,
+ "rest://post:/rest/helloservice:/sayHello?routeId=direct-hi",
+ Op.EVENT_RECEIVED);
+ SpanData log = getSpan(spans, "log://hi", Op.EVENT_SENT);
+ SpanData mock = getSpan(spans, "mock://end", Op.EVENT_SENT);
+
+ // Validate span completion
+ assertTrue(testProducer.hasEnded());
+ assertTrue(direct.hasEnded());
+ assertTrue(directSendTo.hasEnded());
+ assertTrue(directSendFrom.hasEnded());
+ assertTrue(cxfRs.hasEnded());
+ assertTrue(rest.hasEnded());
+ assertTrue(log.hasEnded());
+ assertTrue(mock.hasEnded());
+
+ // Validate same trace
+ assertEquals(testProducer.getSpanContext().getTraceId(), direct.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), directSendTo.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), directSendFrom.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), cxfRs.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), rest.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), log.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), mock.getSpanContext().getTraceId());
+
+ // Validate different Exchange ID
+ assertNotEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ rest.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ direct.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ directSendTo.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ directSendFrom.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ cxfRs.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(rest.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ log.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(rest.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ mock.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+
+ // Validate hierarchy
+ assertFalse(testProducer.getParentSpanContext().isValid());
+ assertEquals(testProducer.getSpanContext().getSpanId(), direct.getParentSpanContext().getSpanId());
+ assertEquals(direct.getSpanContext().getSpanId(), directSendTo.getParentSpanContext().getSpanId());
+ assertEquals(directSendTo.getSpanContext().getSpanId(), directSendFrom.getParentSpanContext().getSpanId());
+ assertEquals(directSendFrom.getSpanContext().getSpanId(), cxfRs.getParentSpanContext().getSpanId());
+ assertEquals(cxfRs.getSpanContext().getSpanId(), rest.getParentSpanContext().getSpanId());
+ assertEquals(rest.getSpanContext().getSpanId(), log.getParentSpanContext().getSpanId());
+ assertEquals(rest.getSpanContext().getSpanId(), mock.getParentSpanContext().getSpanId());
+
+ // Validate message logging
+ // The micrometer framework sets the name instead of the opentelemetry attributes
+ assertEquals("message=A direct message", directSendFrom.getEvents().get(0).getName());
+ assertEquals("message=say-hi", rest.getEvents().get(0).getName());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .routeId("myRoute")
+ .to("direct:send");
+
+ from("direct:send")
+ .log("A direct message")
+ .to("cxfrs:http://localhost:" + cxfPort
+ + "/rest/helloservice/sayHello?synchronous=false");
+
+ restConfiguration()
+ .port(cxfPort);
+
+ rest("/rest/helloservice")
+ .post("/sayHello")
+ .routeId("rest-GET-say-hi")
+ .to("direct:hi");
+
+ from("direct:hi")
+ .routeId("direct-hi")
+ .delay(2000)
+ .log("say-hi")
+ .to("log:hi")
+ .to("mock:end");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncDirectTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncDirectTest.java
new file mode 100644
index 00000000000..399cf87e3ea
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncDirectTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.telemetry.Op;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class AsyncDirectTest extends MicrometerObservabilityTracerTestSupport {
+
+ @Test
+ void testRouteMultipleRequests() throws InterruptedException, IOException {
+ int j = 10;
+ MockEndpoint mock = getMockEndpoint("mock:end");
+ mock.expectedMessageCount(j);
+ mock.setAssertPeriod(5000);
+ for (int i = 0; i < j; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ mock.assertIsSatisfied(1000);
+ Map<String, MicrometerObservabilityTrace> traces = traces();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(j, traces.size());
+ // Each trace should have the same structure
+ for (MicrometerObservabilityTrace trace : traces.values()) {
+ checkTrace(trace, "Hello!");
+ }
+
+ }
+
+ private void checkTrace(MicrometerObservabilityTrace trace, String expectedBody) {
+ List<SimpleSpan> spans = trace.getSpans();
+ assertEquals(7, spans.size());
+ SimpleSpan testProducer = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_SENT);
+ SimpleSpan direct = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ SimpleSpan newDirectTo = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://new", Op.EVENT_SENT);
+ SimpleSpan log = MicrometerObservabilityTracerTestSupport.getSpan(spans, "log://info", Op.EVENT_SENT);
+ SimpleSpan newDirectFrom = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://new", Op.EVENT_RECEIVED);
+ SimpleSpan newLog = MicrometerObservabilityTracerTestSupport.getSpan(spans, "log://new", Op.EVENT_SENT);
+ SimpleSpan newMock = MicrometerObservabilityTracerTestSupport.getSpan(spans, "mock://end", Op.EVENT_SENT);
+
+ // Validate span completion
+ assertNotEquals("", testProducer.getEndTimestamp());
+ assertNotEquals("", direct.getEndTimestamp());
+ assertNotEquals("", newDirectTo.getEndTimestamp());
+ assertNotEquals("", log.getEndTimestamp());
+ assertNotEquals("", newDirectFrom.getEndTimestamp());
+ assertNotEquals("", newLog.getEndTimestamp());
+ assertNotEquals("", newMock.getEndTimestamp());
+
+ // Validate same trace
+ assertEquals(testProducer.getTraceId(), direct.getTraceId());
+ assertEquals(testProducer.getTraceId(), newDirectTo.getTraceId());
+ assertEquals(testProducer.getTraceId(), log.getTraceId());
+ assertEquals(testProducer.getTraceId(), newDirectFrom.getTraceId());
+ assertEquals(testProducer.getTraceId(), newLog.getTraceId());
+ assertEquals(testProducer.getTraceId(), newMock.getTraceId());
+
+ // Validate same Exchange ID
+ // As it's a "direct" component, we expect the logic to happen within the same
+ // Exchange boundary
+ assertEquals(testProducer.getTags().get("exchangeId"), direct.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), newDirectTo.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), newDirectFrom.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), log.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), newLog.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), newMock.getTags().get("exchangeId"));
+
+ // // Validate hierarchy
+ assertEquals("", testProducer.getParentId());
+ assertEquals(testProducer.getSpanId(), direct.getParentId());
+ assertEquals(direct.getSpanId(), newDirectTo.getParentId());
+ assertEquals(direct.getSpanId(), log.getParentId());
+ assertEquals(newDirectTo.getSpanId(), newDirectFrom.getParentId());
+ assertEquals(newDirectFrom.getSpanId(), newLog.getParentId());
+ assertEquals(newDirectFrom.getSpanId(), newMock.getParentId());
+
+ // Validate message logging
+ assertEquals("message=A direct message", direct.getEvents().iterator().next().getValue());
+ assertEquals("message=A new message", newDirectFrom.getEvents().iterator().next().getValue());
+
+ if (expectedBody == null) {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ log.getEvents().iterator().next().getValue());
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ newLog.getEvents().iterator().next().getValue());
+ } else {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ log.getEvents().iterator().next().getValue());
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ newLog.getEvents().iterator().next().getValue());
+ }
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .routeId("start")
+ .to("direct:new")
+ .log("A direct message")
+ .to("log:info");
+
+ from("direct:new")
+ .delay(2000)
+ .routeId("new")
+ .log("A new message")
+ .to("log:new")
+ .to("mock:end");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncWiretapTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncWiretapTest.java
new file mode 100644
index 00000000000..3458d30ff06
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncWiretapTest.java
@@ -0,0 +1,149 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.telemetry.Op;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+/*
+ * WiretappedRouteTest tests the execution of a new spin off component which would create a new exchange,
+ * for example, using the wiretap component.
+ */
+public class AsyncWiretapTest extends MicrometerObservabilityTracerTestSupport {
+
+ @Test
+ void testRouteMultipleRequests() throws InterruptedException, IOException {
+ int j = 10;
+ MockEndpoint mock = getMockEndpoint("mock:end");
+ mock.expectedMessageCount(j);
+ mock.setAssertPeriod(5000);
+ for (int i = 0; i < j; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ mock.assertIsSatisfied(1000);
+ Map<String, MicrometerObservabilityTrace> traces = traces();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(j, traces.size());
+ // Each trace should have the same structure
+ for (MicrometerObservabilityTrace trace : traces.values()) {
+ checkTrace(trace, "Hello!");
+ }
+
+ }
+
+ private void checkTrace(MicrometerObservabilityTrace trace, String expectedBody) {
+ List<SimpleSpan> spans = trace.getSpans();
+ assertEquals(7, spans.size());
+ SimpleSpan testProducer = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_SENT);
+ SimpleSpan direct = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ SimpleSpan wiretapDirectTo = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://tap", Op.EVENT_SENT);
+ SimpleSpan wiretapDirectFrom
+ = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://tap", Op.EVENT_RECEIVED);
+ SimpleSpan log = MicrometerObservabilityTracerTestSupport.getSpan(spans, "log://info", Op.EVENT_SENT);
+ SimpleSpan wiretapLog = MicrometerObservabilityTracerTestSupport.getSpan(spans, "log://tapped", Op.EVENT_SENT);
+ SimpleSpan wiretapMock = MicrometerObservabilityTracerTestSupport.getSpan(spans, "mock://end", Op.EVENT_SENT);
+
+ // Validate span completion
+ assertNotEquals("", testProducer.getEndTimestamp());
+ assertNotEquals("", direct.getEndTimestamp());
+ assertNotEquals("", wiretapDirectTo.getEndTimestamp());
+ assertNotEquals("", log.getEndTimestamp());
+ assertNotEquals("", wiretapDirectFrom.getEndTimestamp());
+ assertNotEquals("", wiretapLog.getEndTimestamp());
+ assertNotEquals("", wiretapMock.getEndTimestamp());
+
+ // Validate same trace
+ assertEquals(testProducer.getTraceId(), direct.getTraceId());
+ assertEquals(testProducer.getTraceId(), wiretapDirectTo.getTraceId());
+ assertEquals(testProducer.getTraceId(), log.getTraceId());
+ assertEquals(testProducer.getTraceId(), wiretapDirectFrom.getTraceId());
+ assertEquals(testProducer.getTraceId(), wiretapLog.getTraceId());
+ assertEquals(testProducer.getTraceId(), wiretapMock.getTraceId());
+
+ // Validate different Exchange ID
+ assertNotEquals(testProducer.getTags().get("exchangeId"), wiretapDirectTo.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), direct.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), log.getTags().get("exchangeId"));
+ assertEquals(wiretapDirectTo.getTags().get("exchangeId"), wiretapDirectFrom.getTags().get("exchangeId"));
+ assertEquals(wiretapDirectTo.getTags().get("exchangeId"), wiretapLog.getTags().get("exchangeId"));
+ assertEquals(wiretapDirectTo.getTags().get("exchangeId"), wiretapMock.getTags().get("exchangeId"));
+
+ // Validate hierarchy
+ assertEquals("", testProducer.getParentId());
+ assertEquals(testProducer.getSpanId(), direct.getParentId());
+ assertEquals(direct.getSpanId(), wiretapDirectTo.getParentId());
+ assertEquals(direct.getSpanId(), log.getParentId());
+ assertEquals(wiretapDirectTo.getSpanId(), wiretapDirectFrom.getParentId());
+ assertEquals(wiretapDirectFrom.getSpanId(), wiretapLog.getParentId());
+ assertEquals(wiretapDirectFrom.getSpanId(), wiretapMock.getParentId());
+
+ // Validate message logging
+ assertEquals("message=A direct message", direct.getEvents().iterator().next().getValue());
+ assertEquals("message=A tapped message", wiretapDirectFrom.getEvents().iterator().next().getValue());
+
+ if (expectedBody == null) {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ log.getEvents().iterator().next().getValue());
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ wiretapLog.getEvents().iterator().next().getValue());
+ } else {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ log.getEvents().iterator().next().getValue());
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ wiretapLog.getEvents().iterator().next().getValue());
+ }
+
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .routeId("start")
+ .wireTap("direct:tap")
+ .log("A direct message")
+ .to("log:info");
+
+ from("direct:tap")
+ .delay(2000)
+ .routeId("wiretapped")
+ .log("A tapped message")
+ .to("log:tapped")
+ .to("mock:end");
+ }
+ };
+ }
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/CamelOpenTelemetryExtension.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/CamelOpenTelemetryExtension.java
new file mode 100644
index 00000000000..a5e8a199b77
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/CamelOpenTelemetryExtension.java
@@ -0,0 +1,257 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+//import io.opentelemetry.sdk.extension.incubator.trace.LeakDetectingSpanProcessor;
+import io.opentelemetry.sdk.logs.SdkLoggerProvider;
+import io.opentelemetry.sdk.logs.data.LogRecordData;
+import io.opentelemetry.sdk.logs.export.SimpleLogRecordProcessor;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.data.AggregationTemporality;
+import io.opentelemetry.sdk.metrics.data.MetricData;
+import io.opentelemetry.sdk.metrics.internal.SdkMeterProviderUtil;
+import io.opentelemetry.sdk.testing.exporter.InMemoryLogRecordExporter;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.ReadWriteSpan;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.SpanProcessor;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.Marker;
+import org.slf4j.MarkerFactory;
+
+/**
+ * Adapted from
+ * https://github.com/open-telemetry/opentelemetry-java/blob/main/sdk/testing/src/main/java/io/opentelemetry/sdk/testing/junit5/OpenTelemetryExtension.java
+ */
+final class CamelOpenTelemetryExtension implements BeforeEachCallback, AfterEachCallback {
+
+ /**
+ * Returns an extension with a default SDK initialized with an in-memory span exporter and W3C trace context
+ * propagation.
+ */
+ static CamelOpenTelemetryExtension create() {
+ InMemorySpanExporter spanExporter = InMemorySpanExporter.create();
+ SdkTracerProvider tracerProvider = SdkTracerProvider.builder()
+ //.addSpanProcessor(LeakDetectingSpanProcessor.create())
+ .addSpanProcessor(new LoggingSpanProcessor())
+ .addSpanProcessor(SimpleSpanProcessor.create(spanExporter))
+ .build();
+
+ InMemoryMetricReader metricReader = InMemoryMetricReader.create();
+ SdkMeterProvider meterProvider = SdkMeterProvider.builder().registerMetricReader(metricReader).build();
+
+ InMemoryLogRecordExporter logRecordExporter = InMemoryLogRecordExporter.create();
+ SdkLoggerProvider loggerProvider = SdkLoggerProvider.builder()
+ .addLogRecordProcessor(SimpleLogRecordProcessor.create(logRecordExporter))
+ .build();
+ ContextPropagators propagators = ContextPropagators.create(W3CTraceContextPropagator.getInstance());
+ OpenTelemetrySdk openTelemetry = OpenTelemetrySdk.builder()
+ .setPropagators(propagators)
+ .setTracerProvider(tracerProvider)
+ .setMeterProvider(meterProvider)
+ .setLoggerProvider(loggerProvider)
+ .build();
+
+ return new CamelOpenTelemetryExtension(
+ openTelemetry, spanExporter, metricReader, logRecordExporter, propagators);
+ }
+
+ private final OpenTelemetrySdk openTelemetry;
+ private final InMemorySpanExporter spanExporter;
+ private final InMemoryMetricReader metricReader;
+ private final InMemoryLogRecordExporter logRecordExporter;
+ private final ContextPropagators propagators;
+
+ private CamelOpenTelemetryExtension(
+ OpenTelemetrySdk openTelemetry,
+ InMemorySpanExporter spanExporter,
+ InMemoryMetricReader metricReader,
+ InMemoryLogRecordExporter logRecordExporter,
+ ContextPropagators propagators) {
+ this.openTelemetry = openTelemetry;
+ this.spanExporter = spanExporter;
+ this.metricReader = metricReader;
+ this.logRecordExporter = logRecordExporter;
+ this.propagators = propagators;
+ }
+
+ /**
+ * Returns the {@link OpenTelemetrySdk} created by this extension.
+ */
+ public OpenTelemetry getOpenTelemetry() {
+ return openTelemetry;
+ }
+
+ public ContextPropagators getPropagators() {
+ return this.propagators;
+ }
+
+ /**
+ * Returns all the exported {@link SpanData} so far.
+ */
+ public List<SpanData> getSpans() {
+ return spanExporter.getFinishedSpanItems();
+ }
+
+ /**
+ * Returns the current {@link MetricData} in {@link AggregationTemporality#CUMULATIVE} format.
+ */
+ public List<MetricData> getMetrics() {
+ return new ArrayList<>(metricReader.collectAllMetrics());
+ }
+
+ /**
+ * Returns all the exported {@link LogRecordData} so far.
+ */
+ public List<LogRecordData> getLogRecords() {
+ return new ArrayList<>(logRecordExporter.getFinishedLogRecordItems());
+ }
+
+ /**
+ * Clears the collected exported {@link SpanData}. Consider making your test smaller instead of manually clearing
+ * state using this method.
+ */
+ public void clearSpans() {
+ spanExporter.reset();
+ }
+
+ /**
+ * Clears all registered metric instruments, such that {@link #getMetrics()} is empty.
+ */
+ public void clearMetrics() {
+ SdkMeterProviderUtil.resetForTest(openTelemetry.getSdkMeterProvider());
+ }
+
+ /**
+ * Clears the collected exported {@link LogRecordData}. Consider making your test smaller instead of manually
+ * clearing state using this method.
+ */
+ public void clearLogRecords() {
+ logRecordExporter.reset();
+ }
+
+ @Override
+ public void beforeEach(ExtensionContext context) {
+ GlobalOpenTelemetry.resetForTest();
+ GlobalOpenTelemetry.set(openTelemetry);
+ }
+
+ @Override
+ public void afterEach(ExtensionContext context) {
+ GlobalOpenTelemetry.resetForTest();
+ openTelemetry.close();
+ }
+
+ static class LoggingSpanProcessor implements SpanProcessor {
+ private static final Logger LOG = LoggerFactory.getLogger(LoggingSpanProcessor.class);
+ private static final Marker OTEL_MARKER = MarkerFactory.getMarker("OTEL");
+
+ @Override
+ public void onStart(Context context, ReadWriteSpan readWriteSpan) {
+ LOG.info(OTEL_MARKER, "Span started: name - '{}', kind - '{}', id - '{}-{}", readWriteSpan.getName(),
+ readWriteSpan.getKind(),
+ readWriteSpan.getSpanContext().getTraceId(), readWriteSpan.getSpanContext().getSpanId());
+ }
+
+ @Override
+ public boolean isStartRequired() {
+ return true;
+ }
+
+ @Override
+ public void onEnd(ReadableSpan readableSpan) {
+ LOG.info(OTEL_MARKER, "Span ended: name - '{}', kind - '{}', id - '{}-{}", readableSpan.getName(),
+ readableSpan.getKind(),
+ readableSpan.getSpanContext().getTraceId(), readableSpan.getSpanContext().getSpanId());
+ }
+
+ @Override
+ public boolean isEndRequired() {
+ return true;
+ }
+ }
+
+ Map<String, OtelTrace> getTraces() {
+ Map<String, OtelTrace> answer = new HashMap<>();
+ for (SpanData span : this.getSpans()) {
+ String traceId = span.getTraceId();
+ OtelTrace trace = answer.get(traceId);
+ if (trace == null) {
+ trace = new OtelTrace(traceId);
+ answer.put(traceId, trace);
+ }
+ trace.addSpan(span);
+ }
+
+ // Sort the spans for all traces
+ answer.forEach((id, trace) -> Collections.sort(trace.getSpans(), new SpanComparator()));
+
+ return answer;
+ }
+
+ class OtelTrace {
+ String traceId;
+ List<SpanData> spans;
+
+ OtelTrace(String traceId) {
+ this.traceId = traceId;
+ this.spans = new ArrayList<>();
+ }
+
+ void addSpan(SpanData span) {
+ this.spans.add(span);
+ }
+
+ List<SpanData> getSpans() {
+ return this.spans;
+ }
+
+ @Override
+ public String toString() {
+ return traceId + " " + spans;
+ }
+ }
+
+ class SpanComparator implements java.util.Comparator<SpanData> {
+ @Override
+ public int compare(SpanData a, SpanData b) {
+ Long nanosA = a.getStartEpochNanos();
+ Long nanosB = b.getStartEpochNanos();
+ return (int) (nanosA - nanosB);
+ }
+ }
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/DisableEndpointTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/DisableEndpointTest.java
new file mode 100644
index 00000000000..5be3c7a361d
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/DisableEndpointTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+import org.apache.camel.CamelContext;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class DisableEndpointTest extends MicrometerObservabilityTracerTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ tst.setTraceProcessors(true);
+ tst.setExcludePatterns("log*,to*");
+ return super.createCamelContext();
+ }
+
+ @Test
+ void testProcessorsTraceRequest() throws IOException {
+ template.sendBody("direct:start", "my-body");
+ Map<String, MicrometerObservabilityTrace> traces = traces();
+ assertEquals(1, traces.size());
+ checkTrace(traces.values().iterator().next());
+ }
+
+ private void checkTrace(MicrometerObservabilityTrace trace) {
+ List<SimpleSpan> spans = trace.getSpans();
+ assertEquals(2, spans.size());
+ SimpleSpan testProducer = spans.get(0);
+ SimpleSpan direct = spans.get(1);
+
+ // Validate span completion
+ assertNotEquals("", testProducer.getEndTimestamp());
+ assertNotEquals("", direct.getEndTimestamp());
+
+ // Validate same trace
+ assertEquals(testProducer.getTraceId(), direct.getTraceId());
+
+ // Validate hierarchy
+ assertEquals("", testProducer.getParentId());
+ assertEquals(testProducer.getSpanId(), direct.getParentId());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .routeId("start")
+ .log("A message")
+ .to("log:info");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/EnableProcessorsTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/EnableProcessorsTest.java
new file mode 100644
index 00000000000..33a1325b8d1
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/EnableProcessorsTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.telemetry.Op;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class EnableProcessorsTest extends MicrometerObservabilityTracerTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ tst.setTraceProcessors(true);
+ return super.createCamelContext();
+ }
+
+ @Test
+ void testProcessorsTraceRequest() throws IOException {
+ template.sendBody("direct:start", "my-body");
+ Map<String, MicrometerObservabilityTrace> traces = traces();
+ assertEquals(1, traces.size());
+ checkTrace(traces.values().iterator().next());
+ }
+
+ private void checkTrace(MicrometerObservabilityTrace trace) {
+ List<SimpleSpan> spans = trace.getSpans();
+ assertEquals(6, spans.size());
+
+ SimpleSpan testProducer = spans.get(0);
+ SimpleSpan direct = spans.get(1);
+ SimpleSpan innerLog = spans.get(2);
+ SimpleSpan innerProcessor = spans.get(3);
+ SimpleSpan log = spans.get(4);
+ SimpleSpan innerToLog = spans.get(5);
+
+ // Validate span completion
+ assertNotEquals("", testProducer.getEndTimestamp());
+ assertNotEquals("", direct.getEndTimestamp());
+ assertNotEquals("", innerLog.getEndTimestamp());
+ assertNotEquals("", innerProcessor.getEndTimestamp());
+ assertNotEquals("", log.getEndTimestamp());
+ assertNotEquals("", innerToLog.getEndTimestamp());
+
+ // Validate same trace
+ assertEquals(testProducer.getTraceId(), direct.getTraceId());
+ assertEquals(testProducer.getTraceId(), innerLog.getTraceId());
+ assertEquals(testProducer.getTraceId(), innerProcessor.getTraceId());
+ assertEquals(testProducer.getTraceId(), log.getTraceId());
+ assertEquals(testProducer.getTraceId(), innerToLog.getTraceId());
+
+ // Validate op
+ assertEquals(Op.EVENT_RECEIVED.toString(), direct.getTags().get("op"));
+ assertEquals(Op.EVENT_PROCESS.toString(), innerProcessor.getTags().get("op"));
+
+ // Validate hierarchy
+ assertEquals("", testProducer.getParentId());
+ assertEquals(testProducer.getSpanId(), direct.getParentId());
+ assertEquals(direct.getSpanId(), innerLog.getParentId());
+ assertEquals(direct.getSpanId(), innerProcessor.getParentId());
+ assertEquals(direct.getSpanId(), log.getParentId());
+ assertEquals(log.getSpanId(), innerToLog.getParentId());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .routeId("start")
+ .log("A message")
+ .process(new Processor() {
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ exchange.getIn().setHeader("operation", "fake");
+ }
+ })
+ .to("log:info");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTrace.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTrace.java
new file mode 100644
index 00000000000..07e00335bc1
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTrace.java
@@ -0,0 +1,53 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+
+/*
+ * This class is used for testing purposes only. It groups an array of Spans belonging to the same Trace.
+ */
+public class MicrometerObservabilityTrace {
+
+ private String traceId;
+ private List<SimpleSpan> spans;
+
+ public MicrometerObservabilityTrace(String traceId) {
+ this.traceId = traceId;
+ this.spans = new ArrayList<>();
+ }
+
+ public void add(SimpleSpan span) {
+ this.spans.add(span);
+ }
+
+ public List<SimpleSpan> getSpans() {
+ return this.spans;
+ }
+
+ public String getTraceId() {
+ return this.traceId;
+ }
+
+ public String toString() {
+ return "Trace Id: " + traceId + " " + spans;
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerPropagationTestSupport.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerPropagationTestSupport.java
new file mode 100644
index 00000000000..915b6fc27d8
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerPropagationTestSupport.java
@@ -0,0 +1,77 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.util.List;
+
+import io.micrometer.tracing.otel.bridge.OtelCurrentTraceContext;
+import io.micrometer.tracing.otel.bridge.OtelPropagator;
+import io.micrometer.tracing.otel.bridge.OtelTracer;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.telemetry.Op;
+import org.apache.camel.telemetry.TagConstants;
+import org.apache.camel.test.junit5.ExchangeTestSupport;
+
+/**
+ * This test is special as it requires a different setting to inherit the Opentelemetry propagation mechanism.
+ */
+public class MicrometerObservabilityTracerPropagationTestSupport extends ExchangeTestSupport {
+
+ protected CamelOpenTelemetryExtension otelExtension = CamelOpenTelemetryExtension.create();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ MicrometerObservabilityTracer tst = new MicrometerObservabilityTracer();
+ CamelContext context = super.createCamelContext();
+
+ ContextPropagators propagators = otelExtension.getPropagators();
+ io.opentelemetry.api.trace.Tracer otelTracer = otelExtension.getOpenTelemetry().getTracer("traceTest");
+
+ OtelPropagator otelPropagator = new OtelPropagator(propagators, otelTracer);
+ OtelCurrentTraceContext currentTraceContext = new OtelCurrentTraceContext();
+ // We must convert the Otel Tracer into a micrometer Tracer
+ io.micrometer.tracing.Tracer micrometerTracer = new OtelTracer(
+ otelTracer,
+ currentTraceContext,
+ null);
+
+ context.getRegistry().bind("MicrometerObservabilityTracer", micrometerTracer);
+ context.getRegistry().bind("OpentelemetryPropagators", otelPropagator);
+
+ CamelContextAware.trySetCamelContext(tst, context);
+ tst.init(context);
+ return context;
+ }
+
+ protected SpanData getSpan(List<SpanData> trace, String uri, Op op) {
+ for (SpanData span : trace) {
+ String camelURI = span.getAttributes().get(AttributeKey.stringKey("camel.uri"));
+ if (camelURI != null && camelURI.equals(uri)) {
+ String operation = span.getAttributes().get(AttributeKey.stringKey(TagConstants.OP));
+ if (operation != null && operation.equals(op.toString())) {
+ return span;
+ }
+ }
+ }
+ throw new IllegalArgumentException("Trying to get a non existing span!");
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerTest.java
new file mode 100644
index 00000000000..7203edf70bb
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.telemetry.Op;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
+public class MicrometerObservabilityTracerTest extends MicrometerObservabilityTracerTestSupport {
+
+ @Test
+ void testRouteSingleRequest() throws IOException {
+ template.request("direct:start", null);
+ Map<String, MicrometerObservabilityTrace> traces = traces();
+ assertEquals(1, traces.size());
+ checkTrace(traces.values().iterator().next(), null);
+ }
+
+ @Test
+ void testRouteMultipleRequests() throws IOException {
+ for (int i = 1; i <= 10; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ Map<String, MicrometerObservabilityTrace> traces = traces();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(10, traces.size());
+ // Each trace should have the same structure
+ for (MicrometerObservabilityTrace trace : traces.values()) {
+ checkTrace(trace, "Hello!");
+ }
+ }
+
+ private void checkTrace(MicrometerObservabilityTrace trace, String expectedBody) {
+ List<SimpleSpan> spans = trace.getSpans();
+ assertEquals(3, spans.size());
+ SimpleSpan testProducer = spans.get(0);
+ SimpleSpan direct = spans.get(1);
+ SimpleSpan log = spans.get(2);
+
+ // Validate span completion
+ assertNotEquals("", testProducer.getEndTimestamp());
+ assertNotEquals("", direct.getEndTimestamp());
+ assertNotEquals("", log.getEndTimestamp());
+
+ // Validate same trace
+ assertEquals(testProducer.getTraceId(), direct.getTraceId());
+ assertEquals(direct.getTraceId(), log.getTraceId());
+
+ // Validate hierarchy
+ assertEquals("", testProducer.getParentId());
+ assertEquals(testProducer.getSpanId(), direct.getParentId());
+ assertEquals(direct.getSpanId(), log.getParentId());
+
+ // Validate operations
+ assertEquals(Op.EVENT_SENT.toString(), testProducer.getTags().get("op"));
+ assertEquals(Op.EVENT_RECEIVED.toString(), direct.getTags().get("op"));
+
+ // Validate message logging
+ assertEquals("message=A message", direct.getEvents().iterator().next().getValue());
+ if (expectedBody == null) {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ log.getEvents().iterator().next().getValue());
+ } else {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ log.getEvents().iterator().next().getValue());
+ }
+
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .routeId("start")
+ .log("A message")
+ .to("log:info");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerTestSupport.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerTestSupport.java
new file mode 100644
index 00000000000..74c509308fa
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerTestSupport.java
@@ -0,0 +1,70 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+import io.micrometer.tracing.test.simple.SimpleTracer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.telemetry.Op;
+import org.apache.camel.telemetry.TagConstants;
+import org.apache.camel.test.junit5.ExchangeTestSupport;
+
+public class MicrometerObservabilityTracerTestSupport extends ExchangeTestSupport {
+
+ protected SimpleTracer tracer = new SimpleTracer();
+ protected MicrometerObservabilityTracer tst = new MicrometerObservabilityTracer();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("MicrometerObservabilityTracer", tracer);
+ CamelContextAware.trySetCamelContext(tst, context);
+ tst.init(context);
+ return context;
+ }
+
+ protected Map<String, MicrometerObservabilityTrace> traces() {
+ HashMap<String, MicrometerObservabilityTrace> map = new HashMap<>();
+ for (SimpleSpan span : tracer.getSpans()) {
+ String traceId = span.getTraceId();
+ MicrometerObservabilityTrace trace = map.get(traceId);
+ if (trace == null) {
+ trace = new MicrometerObservabilityTrace(traceId);
+ map.put(traceId, trace);
+ }
+ trace.add(span);
+ }
+ return map;
+ }
+
+ protected static SimpleSpan getSpan(List<SimpleSpan> trace, String uri, Op op) {
+ for (SimpleSpan span : trace) {
+ if (span.getTags().get("camel.uri") != null && span.getTags().get("camel.uri").equals(uri)) {
+ if (span.getTags().get(TagConstants.OP).equals(op.toString())) {
+ return span;
+ }
+ }
+ }
+ throw new IllegalArgumentException("Trying to get a non existing span!");
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/SpanPropagationTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/SpanPropagationTest.java
new file mode 100644
index 00000000000..70c4c719187
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/SpanPropagationTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.camel.micrometer.observability;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.micrometer.observability.CamelOpenTelemetryExtension.OtelTrace;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * This test is special as it requires a different setting to inherit the Opentelemetry propagation mechanism.
+ */
+public class SpanPropagationTest extends MicrometerObservabilityTracerPropagationTestSupport {
+
+ @Test
+ void testPropagateUpstreamTraceRequest() throws IOException {
+ template.requestBodyAndHeader("direct:start", "sample body",
+ "traceparent", "00-0af044aea5c127fd5ab5f839de2b8ae2-d362a8a943c2b289-01");
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ assertEquals(1, traces.size());
+ checkTrace(traces.values().iterator().next());
+ }
+
+ private void checkTrace(OtelTrace trace) {
+ List<SpanData> spans = trace.getSpans();
+ assertEquals(3, spans.size());
+ SpanData testProducer = spans.get(0);
+ SpanData direct = spans.get(1);
+ SpanData log = spans.get(2);
+
+ // Validate span completion
+ assertTrue(testProducer.hasEnded());
+ assertTrue(direct.hasEnded());
+ assertTrue(log.hasEnded());
+
+ // Validate same trace
+ assertEquals("0af044aea5c127fd5ab5f839de2b8ae2", testProducer.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), direct.getSpanContext().getTraceId());
+ assertEquals(direct.getSpanContext().getTraceId(), log.getSpanContext().getTraceId());
+
+ // Validate hierarchy
+ assertEquals("d362a8a943c2b289", testProducer.getParentSpanContext().getSpanId());
+ assertEquals(testProducer.getSpanContext().getSpanId(), direct.getParentSpanContext().getSpanId());
+ assertEquals(direct.getSpanContext().getSpanId(), log.getParentSpanContext().getSpanId());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .routeId("start")
+ .log("A message")
+ .to("log:info");
+ }
+ };
+ }
+
+}
diff --git a/components/camel-micrometer-observability/src/test/resources/log4j2.properties b/components/camel-micrometer-observability/src/test/resources/log4j2.properties
new file mode 100644
index 00000000000..80f108cf6fc
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/resources/log4j2.properties
@@ -0,0 +1,30 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+appender.console.type = Console
+appender.console.name = console
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-micrometer-observability-test.log
+appender.file.append = true
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
diff --git a/parent/pom.xml b/parent/pom.xml
index 76eb6db9c31..c5ff0290cd6 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -1922,6 +1922,11 @@
<artifactId>camel-micrometer</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-micrometer-observability</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-micrometer-prometheus</artifactId>
@@ -3287,6 +3292,16 @@
<artifactId>micrometer-tracing</artifactId>
<version>${micrometer-tracing-version}</version>
</dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-tracing-bridge-otel</artifactId>
+ <version>${micrometer-tracing-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.micrometer</groupId>
+ <artifactId>micrometer-tracing-test</artifactId>
+ <version>${micrometer-tracing-version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>