[Backport 4.18.x] fix(components): telemetry span customization

components/camel-micrometer-observability/src/main/docs/micrometer-observability.adoc

@@ -77,6 +77,7 @@ Here it follows an example to configure the component to work with Opentelemetry
 
 ```java
 import org.apache.camel.micrometer.observability.MicrometerObservabilityTracer;
+import io.micrometer.tracing.otel.bridge.OtelBaggageManager;
 import io.micrometer.tracing.otel.bridge.OtelCurrentTraceContext;
 import io.micrometer.tracing.otel.bridge.OtelPropagator;
 import io.micrometer.tracing.otel.bridge.OtelTracer;
@@ -102,8 +103,10 @@ import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
             // which requires a Zipkin server listening to that port
             // )
             .build();
-        ContextPropagators propagators =  ContextPropagators.create(
-                    W3CTraceContextPropagator.getInstance());
+        ContextPropagators propagators = ContextPropagators.create(
+                TextMapPropagator.composite(
+                        W3CTraceContextPropagator.getInstance(),
+                        W3CBaggagePropagator.getInstance()));
         OpenTelemetrySdk openTelemetry = OpenTelemetrySdk.builder()
             .setTracerProvider(sdkTracerProvider)
             .setPropagators(propagators)
@@ -112,8 +115,11 @@ import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
             openTelemetry.getTracer("camel-app");
         io.micrometer.tracing.Tracer micrometerTracer = new OtelTracer(
                 otelTracer,
-                new OtelCurrentTraceContext(),
-                null);
+                currentTraceContext,
+                event -> {
+                },
+                new OtelBaggageManager(
+                        currentTraceContext, List.of(MicrometerObservabilitySpanAdapter.BAGGAGE_CAMEL_FLAG), List.of()));
         OtelPropagator otelPropagator = new OtelPropagator(propagators, otelTracer);
         getContext().getRegistry().bind("MicrometerObservabilityTracer", micrometerTracer);
         getContext().getRegistry().bind("OpentelemetryPropagators", otelPropagator);
@@ -127,6 +133,8 @@ NOTE: this is an example that can be used as a reference. It may not work exactl
 
 You can see that the configuration of this component may get a bit difficult, unless you are already familiar with the tracing technology you're going to implement.
 
+NOTE: an important thing to do is to include a `BaggageManager` specifying the baggage properties to include during context propagation. The field provided in the example, `MicrometerObservabilitySpanAdapter.BAGGAGE_CAMEL_FLAG`, is necessary to handle consistency across asynchronous threads.
+
 === How to trace
 
 Once the application is instrumented and configured, you can observe the traces produced with the tooling compatible to the concrete implementation you have in place. You are invited to follow the specific documentation of each technology.
@@ -139,4 +147,29 @@ Your application may require a Java agent in order to get the traces generated b
 
 You can leverage the `traceHeadersInclusion` to include the generated `CAMEL_TRACE_ID` and `CAMEL_SPAN_ID` into the Camel Exchange and together with `camel-mdc` you can make those headers available in the MDC context (via `camel.mdc.customHeaders=CAMEL_TRACE_ID,CAMEL_SPAN_ID` configuration). This is the idiomatic way in Camel.
 
-As an alternative, you can add Mapped Diagnostic Context tracing information (ie, `trace_id` and `span_id`) adding the specific https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/docs/logger-mdc-instrumentation.md[Opentelemetry Logger MDC auto instrumentation]. This would be available if you configure the Opentelemetry. The logging configuration depends on the logging framework you're using.
+As an alternative, you can add Mapped Diagnostic Context tracing information (ie, `trace_id` and `span_id`) adding the specific https://github.com/open-telemetry/opentelemetry-java-instrumentation/blob/main/docs/logger-mdc-instrumentation.md[Opentelemetry Logger MDC auto instrumentation]. This would be available if you configure the Opentelemetry. The logging configuration depends on the logging framework you're using.
+
+=== Span customization
+
+When you're working at a very low level, you may need to tweak your metrics and add some in-process custom `span` in order to trace some specific measure of your application. If you need this advanced use case, you can create it during your process by configuring a Micrometer Tracer object and share it to your route. For example, in Java DSL:
+
+[source,java]
+----
+protected io.micrometer.tracing.Tracer tracer = new OtelTracer(
+    otelTracer,
+    currentTraceContext,
+    event -> {
+    },
+    new OtelBaggageManager(
+            currentTraceContext, List.of(MicrometerObservabilitySpanAdapter.BAGGAGE_CAMEL_FLAG), List.of()));
+...
+public void process(Exchange exchange) throws Exception {
+    exchange.getIn().setHeader("operation", "fake");
+    // We add a span during the processing. We need to verify this span is correctly
+    // created and belong to the proper hierarchy. Important: the user has to know which is the
+    // tracer, likely, setting it on the camel-telemetry Tracer component explicitly.
+    Span mySpan = tracer.nextSpan().name("mySpan").start();
+    // Do the work here
+    mySpan.end();
+}
+----

components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilitySpanAdapter.java

@@ -18,16 +18,24 @@
 
 import java.util.Map;
 
+import io.micrometer.tracing.BaggageInScope;
+import io.micrometer.tracing.Tracer;
 import org.apache.camel.telemetry.Span;
 
 public class MicrometerObservabilitySpanAdapter implements Span {
 
     private static final String DEFAULT_EVENT_NAME = "log";
+    static final String BAGGAGE_CAMEL_FLAG = "camelScope";
 
     private final io.micrometer.tracing.Span span;
+    private final Tracer.SpanInScope scope;
+    private final BaggageInScope baggage;
 
-    public MicrometerObservabilitySpanAdapter(io.micrometer.tracing.Span span) {
+    public MicrometerObservabilitySpanAdapter(io.micrometer.tracing.Span span, Tracer.SpanInScope scope,
+                                              BaggageInScope baggage) {
         this.span = span;
+        this.scope = scope;
+        this.baggage = baggage;
     }
 
     @Override
@@ -71,7 +79,12 @@ protected void close() {
     }
 
     protected void deactivate() {
-
+        if (baggage != null) {
+            baggage.close();
+        }
+        if (scope != null) {
+            scope.close();
+        }
     }
 
     @Override

components/camel-micrometer-observability/src/main/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracer.java

@@ -21,6 +21,7 @@
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 import io.micrometer.observation.ObservationHandler;
 import io.micrometer.observation.ObservationRegistry;
+import io.micrometer.tracing.BaggageInScope;
 import io.micrometer.tracing.Span.Builder;
 import io.micrometer.tracing.Tracer;
 import io.micrometer.tracing.handler.DefaultTracingObservationHandler;
@@ -136,11 +137,35 @@ public Span create(String spanName, Span parent, SpanContextPropagationExtractor
                     return extractor.get(key) == null ? null : (String) extractor.get(key);
                 });
 
+                /*
+                * 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).
+                * We also need to verify that the span generating this is coming from a Camel "dirty" context.
+                */
+                boolean hasUpstreamTrace = extractor.get("traceparent") != null || extractor.get("X-B3-TraceId") != null;
+                boolean dirtyContext = tracer.getBaggage(MicrometerObservabilitySpanAdapter.BAGGAGE_CAMEL_FLAG).get() != null;
+                if (!hasUpstreamTrace && dirtyContext) {
+                    builder.setNoParent();
+                }
+
+                // Dirty context
+
                 span = builder.start();
             }
+
             span.name(spanName);
 
-            return new MicrometerObservabilitySpanAdapter(span);
+            // We need to enable scope now, in order to store it correctly in the thread context.
+            // NOTE: the scope will be eventually closed by the framework.
+            Tracer.SpanInScope scope = tracer.withSpan(span);
+            // We need to enable baggage as it holds important context propagation properties required to recognize
+            // what we call "dirty" context, i.e., a thread
+            // NOTE: the baggage will be eventually closed by the framework.
+            BaggageInScope baggageInScope
+                    = tracer.createBaggageInScope(MicrometerObservabilitySpanAdapter.BAGGAGE_CAMEL_FLAG, "true");
+            return new MicrometerObservabilitySpanAdapter(span, scope, baggageInScope);
         }
 
         @Override

components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/CamelOpenTelemetryExtension.java

@@ -24,8 +24,10 @@
 
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.baggage.propagation.W3CBaggagePropagator;
 import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
 import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 //import io.opentelemetry.sdk.extension.incubator.trace.LeakDetectingSpanProcessor;
 import io.opentelemetry.sdk.logs.SdkLoggerProvider;
@@ -67,7 +69,12 @@ static CamelOpenTelemetryExtension create() {
         SdkLoggerProvider loggerProvider = SdkLoggerProvider.builder()
                 .addLogRecordProcessor(SimpleLogRecordProcessor.create(logRecordExporter))
                 .build();
-        ContextPropagators propagators = ContextPropagators.create(W3CTraceContextPropagator.getInstance());
+        // ContextPropagators propagators = ContextPropagators.create(W3CTraceContextPropagator.getInstance());
+        // NOTE: BaggagePropagator is required to detect the presence of a possible "dirty" context
+        ContextPropagators propagators = ContextPropagators.create(
+                TextMapPropagator.composite(
+                        W3CTraceContextPropagator.getInstance(),
+                        W3CBaggagePropagator.getInstance()));
         OpenTelemetrySdk openTelemetry = OpenTelemetrySdk.builder()
                 .setPropagators(propagators)
                 .setTracerProvider(tracerProvider)
@@ -209,9 +216,9 @@ public String toString() {
     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);
+            long nanosA = a.getStartEpochNanos();
+            long nanosB = b.getStartEpochNanos();
+            return Long.compare(nanosA, nanosB);
         }
     }
 }

components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MicrometerObservabilityTracerPropagationTestSupport.java

@@ -18,11 +18,15 @@
 
 import java.util.List;
 
+import io.micrometer.tracing.otel.bridge.OtelBaggageManager;
 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.baggage.propagation.W3CBaggagePropagator;
 import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
 import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapPropagator;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import org.apache.camel.CamelContext;
 import org.apache.camel.CamelContextAware;
@@ -37,23 +41,31 @@ public class MicrometerObservabilityTracerPropagationTestSupport extends Exchang
 
     protected CamelOpenTelemetryExtension otelExtension = CamelOpenTelemetryExtension.create();
     protected MicrometerObservabilityTracer tst = new MicrometerObservabilityTracer();
+    protected io.micrometer.tracing.Tracer tracer;
+    io.opentelemetry.api.trace.Tracer otelTracer;
 
     @Override
     protected CamelContext createCamelContext() throws Exception {
         CamelContext context = super.createCamelContext();
 
-        ContextPropagators propagators = otelExtension.getPropagators();
-        io.opentelemetry.api.trace.Tracer otelTracer = otelExtension.getOpenTelemetry().getTracer("traceTest");
+        ContextPropagators propagators = ContextPropagators.create(
+                TextMapPropagator.composite(
+                        W3CTraceContextPropagator.getInstance(),
+                        W3CBaggagePropagator.getInstance()));
+        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(
+        tracer = new OtelTracer(
                 otelTracer,
                 currentTraceContext,
-                null);
+                event -> {
+                },
+                new OtelBaggageManager(
+                        currentTraceContext, List.of(MicrometerObservabilitySpanAdapter.BAGGAGE_CAMEL_FLAG), List.of()));
 
-        context.getRegistry().bind("MicrometerObservabilityTracer", micrometerTracer);
+        context.getRegistry().bind("MicrometerObservabilityTracer", tracer);
         context.getRegistry().bind("OpentelemetryPropagators", otelPropagator);
 
         CamelContextAware.trySetCamelContext(tst, context);

components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/SpanCustomizationTest.java

@@ -0,0 +1,122 @@
+/*
+ * 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.Span;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.sdk.trace.data.SpanData;
+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.micrometer.observability.CamelOpenTelemetryExtension.OtelTrace;
+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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class SpanCustomizationTest extends MicrometerObservabilityTracerPropagationTestSupport {
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        tst.setTraceProcessors(true);
+        return super.createCamelContext();
+    }
+
+    @Test
+    void testRouteSingleRequest() throws IOException {
+        template.request("direct:start", null);
+        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(7, spans.size());
+        SpanData testProducer = spans.get(0);
+        SpanData direct = spans.get(1);
+        SpanData innerLog = spans.get(2);
+        SpanData innerProcessor = spans.get(3);
+        SpanData customSpan = spans.get(4);
+        SpanData log = spans.get(5);
+        SpanData innerToLog = spans.get(6);
+
+        // Validate span completion
+        assertTrue(testProducer.hasEnded());
+        assertTrue(direct.hasEnded());
+        assertTrue(innerLog.hasEnded());
+        assertTrue(innerProcessor.hasEnded());
+        assertTrue(customSpan.hasEnded());
+        assertTrue(log.hasEnded());
+        assertTrue(innerToLog.hasEnded());
+
+        // Validate same trace
+        assertEquals(testProducer.getSpanContext().getTraceId(), direct.getSpanContext().getTraceId());
+        assertEquals(testProducer.getSpanContext().getTraceId(), direct.getSpanContext().getTraceId());
+        assertEquals(testProducer.getSpanContext().getTraceId(), innerLog.getSpanContext().getTraceId());
+        assertEquals(testProducer.getSpanContext().getTraceId(), innerProcessor.getSpanContext().getTraceId());
+        assertEquals(testProducer.getSpanContext().getTraceId(), customSpan.getSpanContext().getTraceId());
+        assertEquals(testProducer.getSpanContext().getTraceId(), log.getSpanContext().getTraceId());
+        assertEquals(testProducer.getSpanContext().getTraceId(), innerToLog.getSpanContext().getTraceId());
+
+        // Validate operations
+        assertEquals(Op.EVENT_RECEIVED.toString(), direct.getAttributes().get(AttributeKey.stringKey("op")));
+        assertEquals(Op.EVENT_PROCESS.toString(), innerProcessor.getAttributes().get(AttributeKey.stringKey("op")));
+
+        // Validate hierarchy
+        assertFalse(testProducer.getParentSpanContext().isValid());
+        assertEquals(testProducer.getSpanContext().getSpanId(), direct.getParentSpanContext().getSpanId());
+        assertEquals(direct.getSpanContext().getSpanId(), innerProcessor.getParentSpanContext().getSpanId());
+        assertEquals(innerProcessor.getSpanContext().getSpanId(), customSpan.getParentSpanContext().getSpanId());
+
+        // Validate custom span
+        assertEquals("mySpan", customSpan.getName());
+    }
+
+    @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");
+                                // We add a span during the processing. We need to verify this span is correctly
+                                // created and belong to the proper hierarchy. Important: the user has to know which is the
+                                // tracer, likely, setting it on the camel-telemetry Tracer component explicitly.
+                                Span mySpan = tracer.nextSpan().name("mySpan").start();
+                                // Do the work here
+                                mySpan.end();
+                            }
+                        })
+                        .to("log:info");
+            }
+        };
+    }
+}