CAMEL-15260 - opentracing: fixes header extraction regression

components/camel-opentelemetry/src/main/java/org/apache/camel/opentelemetry/OpenTelemetryTracer.java

@@ -22,12 +22,18 @@
 import io.opentelemetry.OpenTelemetry;
 import io.opentelemetry.trace.DefaultTracer;
 import io.opentelemetry.trace.Span;
+import io.opentelemetry.trace.SpanContext;
 import io.opentelemetry.trace.Tracer;
+import io.opentelemetry.trace.TracingContextUtils;
+import org.apache.camel.Exchange;
 import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.opentelemetry.propagators.OpenTelemetryGetter;
 import org.apache.camel.opentelemetry.propagators.OpenTelemetrySetter;
 import org.apache.camel.tracing.InjectAdapter;
 import org.apache.camel.tracing.SpanAdapter;
+import org.apache.camel.tracing.SpanDecorator;
 import org.apache.camel.tracing.SpanKind;
+import org.apache.camel.tracing.decorators.AbstractInternalSpanDecorator;
 
 @ManagedResource(description = "OpenTelemetryTracer")
 public class OpenTelemetryTracer extends org.apache.camel.tracing.Tracer {
@@ -43,10 +49,13 @@ public void setTracer(Tracer tracer) {
     }
 
     private Span.Kind mapToSpanKind(SpanKind kind) {
-        if (kind == SpanKind.SPAN_KIND_CLIENT) {
-            return Span.Kind.CLIENT;
+        switch (kind) {
+            case SPAN_KIND_CLIENT: return Span.Kind.CLIENT;
+            case SPAN_KIND_SERVER: return Span.Kind.SERVER;
+            case CONSUMER: return Span.Kind.CONSUMER;
+            case PRODUCER: return Span.Kind.PRODUCER;
+            default: return Span.Kind.SERVER;
         }
-        return Span.Kind.SERVER;
     }
 
     @Override
@@ -76,11 +85,21 @@ protected SpanAdapter startSendingEventSpan(String operationName, SpanKind kind,
     }
 
     @Override
-    protected SpanAdapter startExchangeBeginSpan(String operationName, SpanKind kind, SpanAdapter parent) {
+    protected SpanAdapter startExchangeBeginSpan(Exchange exchange, SpanDecorator sd, String operationName, SpanKind kind, SpanAdapter parent) {
         Span.Builder builder = tracer.spanBuilder(operationName);
         if (parent != null) {
-            OpenTelemetrySpanAdapter oTelSpanWrapper = (OpenTelemetrySpanAdapter) parent;
-            builder = builder.setParent(((OpenTelemetrySpanAdapter) parent).getOpenTelemetrySpan());
+            OpenTelemetrySpanAdapter spanFromExchange = (OpenTelemetrySpanAdapter) parent;
+            builder = builder.setParent(spanFromExchange.getOpenTelemetrySpan());
+        } else {
+            Context ctx = OpenTelemetry.getPropagators().getHttpTextFormat().extract(Context.current(), sd.getExtractAdapter(exchange.getIn().getHeaders(), encoding), new OpenTelemetryGetter());
+            Span span = TracingContextUtils.getSpan(ctx);
+            SpanContext parentFromHeaders = span.getContext();
+
+            if (parentFromHeaders != null && parentFromHeaders.isValid()) {
+                builder.setParent(parentFromHeaders).setSpanKind(mapToSpanKind(sd.getReceiverSpanKind()));
+            } else if (!(sd instanceof AbstractInternalSpanDecorator)) {
+                builder.setSpanKind(mapToSpanKind(sd.getReceiverSpanKind()));
+            }
         }
 
         return new OpenTelemetrySpanAdapter(builder.startSpan());

components/camel-opentracing/pom.xml

@@ -94,6 +94,28 @@
             <scope>test</scope>
             <type>test-jar</type>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-http</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-jetty</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.glassfish.grizzly</groupId>
+            <artifactId>grizzly-http-client</artifactId>
+            <version>1.16</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.opentracing.contrib</groupId>
+            <artifactId>opentracing-grizzly-ahc</artifactId>
+            <version>0.1.3</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.camel</groupId>
             <artifactId>camel-test-spring-junit5</artifactId>

components/camel-opentracing/src/main/java/org/apache/camel/opentracing/OpenTracingTracer.java

@@ -18,6 +18,8 @@
 
 import java.util.Set;
 
+import io.opentracing.Span;
+import io.opentracing.SpanContext;
 import io.opentracing.Tracer;
 import io.opentracing.Tracer.SpanBuilder;
 import io.opentracing.contrib.tracerresolver.TracerResolver;
@@ -29,7 +31,9 @@
 import org.apache.camel.spi.RoutePolicy;
 import org.apache.camel.tracing.InjectAdapter;
 import org.apache.camel.tracing.SpanAdapter;
+import org.apache.camel.tracing.SpanDecorator;
 import org.apache.camel.tracing.SpanKind;
+import org.apache.camel.tracing.decorators.AbstractInternalSpanDecorator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,10 +59,13 @@ public OpenTracingTracer() {
     }
 
     private String mapToSpanKind(SpanKind kind) {
-        if (kind == SpanKind.SPAN_KIND_CLIENT) {
-            return Tags.SPAN_KIND_CLIENT;
+        switch (kind) {
+            case SPAN_KIND_CLIENT: return Tags.SPAN_KIND_CLIENT;
+            case SPAN_KIND_SERVER: return Tags.SPAN_KIND_SERVER;
+            case CONSUMER: return Tags.SPAN_KIND_CONSUMER;
+            case PRODUCER: return Tags.SPAN_KIND_PRODUCER;
+            default: return null;
         }
-        return Tags.SPAN_KIND_SERVER;
     }
 
 
@@ -97,12 +104,27 @@ protected void initTracer() {
         return new OpenTracingSpanAdapter(spanBuilder.start());
     }
 
-    @Override protected SpanAdapter startExchangeBeginSpan(String operationName, SpanKind kind, SpanAdapter parent) {
-        SpanBuilder spanBuilder = tracer.buildSpan(operationName);
+    @Override protected SpanAdapter startExchangeBeginSpan(Exchange exchange, SpanDecorator sd, String operationName, SpanKind kind, SpanAdapter parent) {
+        SpanBuilder builder = tracer.buildSpan(operationName);
         if (parent != null) {
-            spanBuilder.asChildOf(((OpenTracingSpanAdapter) parent).getOpenTracingSpan());
+            // we found a Span already associated with this exchange, use it as parent
+            Span parentFromExchange = ((OpenTracingSpanAdapter) parent).getOpenTracingSpan();
+            builder.asChildOf(parentFromExchange);
+        } else {
+            SpanContext parentFromHeaders = tracer.extract(Format.Builtin.TEXT_MAP, new OpenTracingExtractAdapter(sd.getExtractAdapter(exchange.getIn().getHeaders(), encoding)));
+
+            if (parentFromHeaders != null) {
+                // this means it's an inter-process request or the context was manually injected into the headers
+                // we add the server tag
+                builder.asChildOf(parentFromHeaders).withTag(Tags.SPAN_KIND.getKey(), mapToSpanKind(sd.getReceiverSpanKind()));
+            } else if (!(sd instanceof AbstractInternalSpanDecorator)) {
+                // no parent found anywhere and it's not an internal endpoint
+                // we add the server tag
+                builder.withTag(Tags.SPAN_KIND.getKey(), mapToSpanKind(sd.getReceiverSpanKind()));
+            }
         }
-        return new OpenTracingSpanAdapter(spanBuilder.start());
+
+        return new OpenTracingSpanAdapter(builder.start());
     }
 
     public Tracer getTracer() {

components/camel-opentracing/src/test/java/org/apache/camel/opentracing/CamelOpenTracingTestSupport.java

@@ -162,7 +162,7 @@ protected void verifySpan(int index, SpanTestData[] testdata, List<MockSpan> spa
 
         if (!td.getTags().isEmpty()) {
             for (Map.Entry<String, String> entry : td.getTags().entrySet()) {
-                assertEquals(entry.getValue(), span.tags().get(entry.getKey()));
+                assertEquals(entry.getValue(), String.valueOf(span.tags().get(entry.getKey())));
             }
         }
 

components/camel-opentracing/src/test/java/org/apache/camel/opentracing/EIPTracingActiveSpanTest.java

@@ -24,8 +24,8 @@
 import org.apache.camel.RoutesBuilder;
 import org.apache.camel.builder.RouteBuilder;
 import org.apache.camel.spi.InterceptStrategy;
-import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtensionContext;
 
 public class EIPTracingActiveSpanTest extends CamelOpenTracingTestSupport {
 
@@ -41,8 +41,8 @@ public EIPTracingActiveSpanTest() {
         super(testdata);
     }
 
-    @BeforeAll
-    public void setUp() throws Exception {
+    @Override
+    public void doPreSetup() {
         GlobalTracerTestUtil.resetGlobalTracer();
     }
 

components/camel-opentracing/src/test/java/org/apache/camel/opentracing/OpentracingSpanCollectorInRegistryTest.java

@@ -22,8 +22,8 @@
 import org.apache.camel.BindToRegistry;
 import org.apache.camel.CamelContext;
 import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtensionContext;
 
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -34,8 +34,8 @@ public class OpentracingSpanCollectorInRegistryTest extends CamelTestSupport {
     @BindToRegistry("tracer")
     private NoopTracer c = NoopTracerFactory.create();
 
-    @BeforeAll
-    public void setUp() throws Exception {
+    @Override
+    public void doPreSetup() {
         GlobalTracerTestUtil.resetGlobalTracer();
     }
 

components/camel-opentracing/src/test/java/org/apache/camel/opentracing/integration/InterprocessTest.java

@@ -0,0 +1,91 @@
+/*
+ * 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.opentracing.integration;
+
+
+import com.ning.http.client.AsyncHttpClient;
+import com.ning.http.client.AsyncHttpClientConfig;
+import io.opentracing.contrib.grizzly.ahc.TracingRequestFilter;
+import io.opentracing.tag.Tags;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.opentracing.CamelOpenTracingTestSupport;
+import org.apache.camel.opentracing.SpanTestData;
+import org.apache.camel.test.AvailablePortFinder;
+import org.junit.jupiter.api.Test;
+
+
+public class InterprocessTest extends CamelOpenTracingTestSupport {
+    private static final String URL = "http://localhost:" + AvailablePortFinder.getNextAvailable() + "/test";
+    private static final String URI = "jetty:" + URL;
+
+    private static SpanTestData[] testdata = {
+            new SpanTestData().setLabel("jetty:start server").setUri(URI).setOperation("GET")
+                    .setKind("server")
+                    .addTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_SERVER)
+                    .addTag(Tags.COMPONENT.getKey(), "camel-jetty")
+                    .addTag(Tags.HTTP_URL.getKey(), URL)
+                    .addTag(Tags.HTTP_METHOD.getKey(), "GET")
+                    // jetty does not set the http response code header
+                    // this is where instrumentation has value
+                    // .addTag(Tags.HTTP_STATUS.getKey(), "200")
+                    .setParentId(1),
+            new SpanTestData().setLabel("ahc: client").setUri(null).setOperation("HTTP::GET")
+                    .setKind("client")
+                    .addTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CLIENT)
+                    .addTag(Tags.COMPONENT.getKey(), "java-grizzly-ahc")
+                    .addTag(Tags.HTTP_URL.getKey(), URL)
+                    .addTag(Tags.HTTP_METHOD.getKey(), "GET")
+                    .addTag(Tags.HTTP_STATUS.getKey(), "200")
+    };
+
+    public InterprocessTest() {
+        super(testdata);
+    }
+
+    @Test
+    public void testBridgeEndpoint() throws Exception {
+        AsyncHttpClientConfig config = new AsyncHttpClientConfig.Builder()
+                .addRequestFilter(new TracingRequestFilter(getTracer()))
+                .build();
+
+        try (AsyncHttpClient client = new AsyncHttpClient(config)) {
+            client.prepareGet(URL).execute().get();
+        }
+
+        verify();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                Processor serviceProc = new Processor() {
+                    public void process(Exchange exchange) throws Exception {
+                        // get the request URL and copy it to the request body
+                        String uri = exchange.getIn().getHeader(Exchange.HTTP_URI, String.class);
+                        exchange.getMessage().setBody(uri);
+                    }
+                };
+
+                from(URI).process(serviceProc);
+            }
+        };
+    }
+}

components/camel-tracing/src/main/java/org/apache/camel/tracing/Tracer.java

@@ -71,14 +71,14 @@ public abstract class Tracer extends ServiceSupport implements RoutePolicyFactor
     private final TracingEventNotifier eventNotifier = new TracingEventNotifier();
     private Set<String> excludePatterns = new HashSet<>(0);
     private InterceptStrategy tracingStrategy;
-    private boolean encoding;
+    protected boolean encoding;
     private CamelContext camelContext;
 
     protected abstract void initTracer();
 
     protected abstract SpanAdapter startSendingEventSpan(String operationName, SpanKind kind, SpanAdapter parent);
 
-    protected abstract SpanAdapter startExchangeBeginSpan(String operationName, SpanKind kind, SpanAdapter parent);
+    protected abstract SpanAdapter startExchangeBeginSpan(Exchange exchange, SpanDecorator sd, String operationName, SpanKind kind, SpanAdapter parent);
 
     protected abstract void finishSpan(SpanAdapter span);
 
@@ -283,11 +283,7 @@ public void onExchangeBegin(Route route, Exchange exchange) {
                 SpanDecorator sd = getSpanDecorator(route.getEndpoint());
                 SpanAdapter parent = ActiveSpanManager.getSpan(exchange);
                 SpanAdapter span;
-                if (parent == null && !(sd instanceof AbstractInternalSpanDecorator)) {
-                    span = startExchangeBeginSpan(sd.getOperationName(exchange, route.getEndpoint()), sd.getReceiverSpanKind(), null);
-                } else {
-                    span = startExchangeBeginSpan(sd.getOperationName(exchange, route.getEndpoint()), null, parent);
-                }
+                span = startExchangeBeginSpan(exchange, sd, sd.getOperationName(exchange, route.getEndpoint()), sd.getReceiverSpanKind(), parent);
                 sd.pre(span, exchange, route.getEndpoint());
                 ActiveSpanManager.activate(exchange, span);
                 if (LOG.isTraceEnabled()) {