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
The following commit(s) were added to refs/heads/main by this push:
new 3a751cf60be7 feat(components): otel camel flag to control external
root span
3a751cf60be7 is described below
commit 3a751cf60be7056d7338400545f2b5312daac232
Author: Pasquale Congiusti <[email protected]>
AuthorDate: Thu Jan 8 11:10:45 2026 +0100
feat(components): otel camel flag to control external root span
We have introduced a flag which we use to control wether a span was
generated from Camel or not. This is useful when we reuse "dirty" threads which
were previously asynchronously setting otel context via thread locals. By
checking the flag we can decide to continue propagation of a trace from an
existing span (for example, when vertx or otel agents are injecting a span
before the Camel process kicks in).
Ref CAMEL-22648
---
.../opentelemetry2/OpenTelemetrySpanAdapter.java | 10 +-
.../camel/opentelemetry2/OpenTelemetryTracer.java | 11 +-
.../apache/camel/opentelemetry2/SpanInjection.java | 155 +++++++++++++++++++++
3 files changed, 174 insertions(+), 2 deletions(-)
diff --git
a/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetrySpanAdapter.java
b/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetrySpanAdapter.java
index df97a73ecf0b..fab7a069b0c3 100644
---
a/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetrySpanAdapter.java
+++
b/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetrySpanAdapter.java
@@ -29,14 +29,18 @@ import org.apache.camel.telemetry.TagConstants;
public class OpenTelemetrySpanAdapter implements
org.apache.camel.telemetry.Span {
private static final String DEFAULT_EVENT_NAME = "log";
+ static final String BAGGAGE_CAMEL_FLAG = "camelScope";
private final Span otelSpan;
private final Baggage baggage;
private Scope scope;
+ private Scope baggageScope;
protected OpenTelemetrySpanAdapter(Span otelSpan, Baggage baggage) {
this.otelSpan = otelSpan;
- this.baggage = baggage;
+ // We store an important flag in the baggage in order to verify if the
+ // root span was generated internally or from a third party dependency.
+ this.baggage = baggage.toBuilder().put(BAGGAGE_CAMEL_FLAG,
"true").build();
}
protected Span getSpan() {
@@ -45,6 +49,7 @@ public class OpenTelemetrySpanAdapter implements
org.apache.camel.telemetry.Span
protected void makeCurrent() {
this.scope = this.otelSpan.makeCurrent();
+ this.baggageScope = this.baggage.makeCurrent();
}
protected void end() {
@@ -52,6 +57,9 @@ public class OpenTelemetrySpanAdapter implements
org.apache.camel.telemetry.Span
}
protected void close() {
+ if (baggageScope != null) {
+ this.baggageScope.close();
+ }
if (scope != null) {
this.scope.close();
}
diff --git
a/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetryTracer.java
b/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetryTracer.java
index 071425324c6a..381488b6e300 100644
---
a/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetryTracer.java
+++
b/components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetryTracer.java
@@ -105,8 +105,17 @@ public class OpenTelemetryTracer extends
org.apache.camel.telemetry.Tracer {
builder =
builder.setParent(Context.current().with(otelParentSpan.getSpan()));
baggage = otelParentSpan.getBaggage();
} else {
+ Context current = Context.root();
+ // If the current span was generated by Camel, then, this is a
"dirty" context.
+ // A "dirty" context happens when the Camel thread local is
reused and
+ // due to the way Camel async works, can't reliably clean its
context before reusing it.
+ if
(Baggage.current().getEntryValue(OpenTelemetrySpanAdapter.BAGGAGE_CAMEL_FLAG)
== null) {
+ // Not "dirty" context. In this case a Span exists and the
current span was generated by some third party dependency (ie, vertx)
+ // therefore we need to consider this span as the root on
such a trace.
+ current = Context.current();
+ }
// Try to get parent from context propagation (upstream traces)
- Context ctx =
contextPropagators.getTextMapPropagator().extract(Context.root(), extractor,
+ Context ctx =
contextPropagators.getTextMapPropagator().extract(current, extractor,
new TextMapGetter<SpanContextPropagationExtractor>() {
@Override
public Iterable<String>
keys(SpanContextPropagationExtractor carrier) {
diff --git
a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/SpanInjection.java
b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/SpanInjection.java
new file mode 100644
index 000000000000..917aaba3b8b9
--- /dev/null
+++
b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/SpanInjection.java
@@ -0,0 +1,155 @@
+/*
+ * 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.opentelemetry2;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+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.opentelemetry2.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.assertTrue;
+
+public class SpanInjection extends OpenTelemetryTracerTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ OpenTelemetryTracer tst = new OpenTelemetryTracer();
+ tst.setTracer(otelExtension.getOpenTelemetry().getTracer("traceTest"));
+
tst.setContextPropagators(otelExtension.getOpenTelemetry().getPropagators());
+ tst.setTraceProcessors(true);
+ CamelContext context = super.createCamelContext();
+ CamelContextAware.trySetCamelContext(tst, context);
+ tst.init(context);
+ return context;
+ }
+
+ @Test
+ void testRouteSingleRequest() throws IOException {
+ // NOTE: we simulate that any external third party is the root parent,
as we want Camel traces to depend on it.
+ Span span =
otelExtension.getOpenTelemetry().getTracer("traceTest").spanBuilder("mySpan").startSpan();
+ String expectedTrace = span.getSpanContext().getTraceId();
+ String expectedSpan = span.getSpanContext().getSpanId();
+ try (Scope scope = span.makeCurrent()) {
+ template.sendBody("direct:start", "my-body");
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ assertEquals(1, traces.size());
+ checkTrace(traces.values().iterator().next(), expectedTrace,
expectedSpan);
+ }
+ }
+
+ @Test
+ void testRouteMultipleRequests() throws IOException {
+ int i = 10;
+ Map<String, String> tracesRef = new HashMap<>();
+ for (int j = 0; j < i; j++) {
+ // NOTE: we simulate that any external third party is the root
parent, as we want Camel traces to depend on it.
+ Span span =
otelExtension.getOpenTelemetry().getTracer("traceTest").spanBuilder("mySpan").startSpan();
+ // We hold the reference of each parent span for each trace
+ tracesRef.put(span.getSpanContext().getTraceId(),
span.getSpanContext().getSpanId());
+ try (Scope scope = span.makeCurrent()) {
+ context.createProducerTemplate().sendBody("direct:start",
"Hello!");
+ }
+ }
+ 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(i, traces.size());
+ // Each trace should have the same structure
+ for (OtelTrace trace : traces.values()) {
+ String expectedTrace = trace.traceId;
+ String expectedSpan = tracesRef.get(expectedTrace);
+ checkTrace(trace, expectedTrace, expectedSpan);
+ }
+ }
+
+ private void checkTrace(OtelTrace trace, String parentTrace, String
parentSpan) {
+ List<SpanData> spans = trace.getSpans();
+ assertEquals(6, spans.size());
+ SpanData testProducer = spans.get(0);
+ SpanData direct = spans.get(1);
+ SpanData innerLog = spans.get(2);
+ SpanData innerProcessor = spans.get(3);
+ SpanData log = spans.get(4);
+ SpanData innerToLog = spans.get(5);
+
+ // Validate span completion
+ assertTrue(testProducer.hasEnded());
+ assertTrue(direct.hasEnded());
+ assertTrue(innerLog.hasEnded());
+ assertTrue(innerProcessor.hasEnded());
+ assertTrue(log.hasEnded());
+ assertTrue(innerToLog.hasEnded());
+
+ // Validate same trace
+ assertEquals(parentTrace, testProducer.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(),
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
+ // The parent now must be a valid trace as it was generated by a third
party (our test in this case).
+ assertTrue(testProducer.getParentSpanContext().isValid());
+ assertEquals(parentSpan,
testProducer.getParentSpanContext().getSpanId());
+
+ assertEquals(testProducer.getSpanContext().getSpanId(),
direct.getParentSpanContext().getSpanId());
+ assertEquals(direct.getSpanContext().getSpanId(),
innerLog.getParentSpanContext().getSpanId());
+ assertEquals(direct.getSpanContext().getSpanId(),
innerProcessor.getParentSpanContext().getSpanId());
+ assertEquals(direct.getSpanContext().getSpanId(),
log.getParentSpanContext().getSpanId());
+ assertEquals(log.getSpanContext().getSpanId(),
innerToLog.getParentSpanContext().getSpanId());
+ }
+
+ @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");
+ }
+ };
+ }
+}
