improving vertx websocket for end to end tracing in camel quarkus - POC

extensions/vertx-websocket/runtime/pom.xml

@@ -48,6 +48,12 @@
             <groupId>org.apache.camel</groupId>
             <artifactId>camel-vertx-websocket</artifactId>
         </dependency>
+        <!-- OpenTelemetry API for span context capture (optional) -->
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-api</artifactId>
+            <optional>true</optional>
+        </dependency>
     </dependencies>
 
     <build>

extensions/vertx-websocket/runtime/src/main/java/org/apache/camel/quarkus/component/vertx/websocket/QuarkusVertxWebsocketConsumer.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.quarkus.component.vertx.websocket;
+
+import io.opentelemetry.api.trace.SpanContext;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.vertx.websocket.VertxWebsocketConsumer;
+import org.apache.camel.component.vertx.websocket.VertxWebsocketEndpoint;
+import org.apache.camel.component.vertx.websocket.VertxWebsocketEvent;
+
+/**
+ * Custom Vert.x WebSocket consumer for Quarkus that captures the HTTP upgrade span context
+ * and makes it available for OpenTelemetry tracing with span links.
+ */
+public class QuarkusVertxWebsocketConsumer extends VertxWebsocketConsumer {
+
+    /**
+     * Exchange header name for storing the HTTP upgrade span context.
+     * This allows the OpenTelemetry tracer to create span links back to the original HTTP request.
+     * Format: "traceId:spanId" (hex strings separated by colon)
+     */
+    public static final String PROPERTY_HANDSHAKE_SPAN_CONTEXT = "CamelVertxWebsocketHandshakeSpanContext";
+
+    public QuarkusVertxWebsocketConsumer(VertxWebsocketEndpoint endpoint, Processor processor) {
+        super(endpoint, processor);
+    }
+
+    @Override
+    protected void populateExchangeHeaders(
+            Exchange exchange, String connectionKey, SocketAddress remote, RoutingContext routingContext,
+            VertxWebsocketEvent event) {
+        super.populateExchangeHeaders(exchange, connectionKey, remote, routingContext, event);
+
+        SpanContext handshakeSpanContext = routingContext.get(PROPERTY_HANDSHAKE_SPAN_CONTEXT);
+        if (handshakeSpanContext != null && handshakeSpanContext.isValid()) {
+            String serialized = handshakeSpanContext.getTraceId() + ":" + handshakeSpanContext.getSpanId();
+            exchange.getIn().setHeader(PROPERTY_HANDSHAKE_SPAN_CONTEXT, serialized);
+        }
+    }
+}

extensions/vertx-websocket/runtime/src/main/java/org/apache/camel/quarkus/component/vertx/websocket/VertxWebsocketRecorder.java

@@ -34,6 +34,8 @@
 import io.vertx.core.http.WebSocketConnectOptions;
 import io.vertx.ext.web.Router;
 import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
 import org.apache.camel.component.vertx.websocket.VertxWebsocketComponent;
 import org.apache.camel.component.vertx.websocket.VertxWebsocketConfiguration;
 import org.apache.camel.component.vertx.websocket.VertxWebsocketConstants;
@@ -137,6 +139,12 @@ public QuarkusVertxWebsocketEndpoint(String uri, VertxWebsocketComponent compone
             super(uri, component, configuration);
         }
 
+        @Override
+        public Consumer createConsumer(Processor processor) throws Exception {
+            // Use our custom consumer that captures OpenTelemetry span context
+            return new QuarkusVertxWebsocketConsumer(this, processor);
+        }
+
         @Override
         public WebSocketConnectOptions getWebSocketConnectOptions(HttpClientOptions options) {
             WebSocketConnectOptions connectOptions = super.getWebSocketConnectOptions(options);

integration-tests/vertx-websocket/pom.xml

@@ -51,6 +51,22 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-websockets</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-opentelemetry2</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-opentelemetry</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-timer</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-logging</artifactId>
+        </dependency>
 
     	<!-- test dependencies -->
         <dependency>

integration-tests/vertx-websocket/src/main/java/org/apache/camel/quarkus/component/vertx/websocket/it/VertxWebsocketSpanLinkResource.java

@@ -0,0 +1,72 @@
+/*
+ * 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.quarkus.component.vertx.websocket.it;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.MediaType;
+import org.apache.camel.ConsumerTemplate;
+import org.apache.camel.ProducerTemplate;
+
+@Path("/vertx-websocket/span-link")
+@ApplicationScoped
+public class VertxWebsocketSpanLinkResource {
+
+    @Inject
+    ProducerTemplate producerTemplate;
+
+    @Inject
+    ConsumerTemplate consumerTemplate;
+
+    /**
+     * Send a message to the WebSocket endpoint.
+     * This simulates a client or timer sending a message.
+     */
+    @POST
+    @Path("/send")
+    @Produces(MediaType.TEXT_PLAIN)
+    public String sendMessage(String message) {
+        producerTemplate.sendBody("vertx-websocket:/span-link-test?sendToAll=true", message);
+        return "Message sent: " + message;
+    }
+
+    /**
+     * Receive messages that were processed by the WebSocket consumer.
+     * This allows tests to verify that messages were received and processed.
+     */
+    @GET
+    @Path("/receive")
+    @Produces(MediaType.TEXT_PLAIN)
+    public String receiveMessage() {
+        String message = consumerTemplate.receiveBody("seda:websocket-messages", 5000, String.class);
+        return message != null ? message : "No message received";
+    }
+
+    /**
+     * Health check endpoint to verify routes are started.
+     */
+    @GET
+    @Path("/health")
+    @Produces(MediaType.TEXT_PLAIN)
+    public String health() {
+        return "OK";
+    }
+}

integration-tests/vertx-websocket/src/main/java/org/apache/camel/quarkus/component/vertx/websocket/it/VertxWebsocketSpanLinkRoutes.java

@@ -0,0 +1,46 @@
+/*
+ * 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.quarkus.component.vertx.websocket.it;
+
+import org.apache.camel.builder.RouteBuilder;
+
+/**
+ * Routes for testing OpenTelemetry span links with WebSocket connections.
+ * This test scenario validates that WebSocket message spans are linked
+ * back to the HTTP upgrade request span.
+ */
+public class VertxWebsocketSpanLinkRoutes extends RouteBuilder {
+
+    @Override
+    public void configure() throws Exception {
+        // Route 1: WebSocket Consumer - receives messages from client
+        from("vertx-websocket:/span-link-test")
+                .routeId("websocket-consumer")
+                .to("direct:logMessage");
+
+        // Route 2: Log the message
+        from("direct:logMessage")
+                .routeId("log-message")
+                .log("Greeting: ${body}");
+
+        // Route 3: Timer Producer - sends messages to WebSocket clients (with sendToAll=true)
+        from("timer:websocket-timer?period=5000&repeatCount=1")
+                .routeId("timer-producer")
+                .setBody().constant("Hello World")
+                .to("vertx-websocket:/span-link-test?sendToAll=true");
+    }
+}

integration-tests/vertx-websocket/src/test/java/org/apache/camel/quarkus/component/vertx/websocket/it/VertxWebsocketMultiSpanLinkTest.java

@@ -0,0 +1,103 @@
+/*
+ * 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.quarkus.component.vertx.websocket.it;
+
+import java.net.URI;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.vertx.core.Vertx;
+import io.vertx.core.http.WebSocket;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test to verify that multiple WebSocket connections from different HTTP requests
+ * result in multiple FOLLOWS_FROM span links when using sendToAll.
+ */
+@QuarkusTest
+public class VertxWebsocketMultiSpanLinkTest {
+
+    @TestHTTPResource("/span-link-test")
+    URI websocketUri;
+
+    @Test
+    public void testMultipleConnectionsGenerateMultipleSpanLinks() throws Exception {
+        Vertx vertx = Vertx.vertx();
+
+        try {
+            // Create first WebSocket connection (HTTP request 1)
+            CompletableFuture<WebSocket> ws1Future = new CompletableFuture<>();
+            vertx.createHttpClient().webSocket(websocketUri.getPort(), websocketUri.getHost(), websocketUri.getPath(),
+                    result -> {
+                        if (result.succeeded()) {
+                            ws1Future.complete(result.result());
+                        } else {
+                            ws1Future.completeExceptionally(result.cause());
+                        }
+                    });
+            WebSocket ws1 = ws1Future.get(10, TimeUnit.SECONDS);
+
+            // Small delay to ensure connections are established separately
+            Thread.sleep(100);
+
+            // Create second WebSocket connection (HTTP request 2)
+            CompletableFuture<WebSocket> ws2Future = new CompletableFuture<>();
+            vertx.createHttpClient().webSocket(websocketUri.getPort(), websocketUri.getHost(), websocketUri.getPath(),
+                    result -> {
+                        if (result.succeeded()) {
+                            ws2Future.complete(result.result());
+                        } else {
+                            ws2Future.completeExceptionally(result.cause());
+                        }
+                    });
+            WebSocket ws2 = ws2Future.get(10, TimeUnit.SECONDS);
+
+            // Collect messages from both connections
+            CompletableFuture<String> msg1 = new CompletableFuture<>();
+            CompletableFuture<String> msg2 = new CompletableFuture<>();
+
+            ws1.textMessageHandler(msg1::complete);
+            ws2.textMessageHandler(msg2::complete);
+
+            // Send a client message to trigger Consumer span link
+            ws1.writeTextMessage("Client message from connection 1");
+
+            // Wait for Timer to trigger and send to all connections (this will create Producer span links)
+            String message1 = msg1.get(15, TimeUnit.SECONDS);
+            String message2 = msg2.get(15, TimeUnit.SECONDS);
+
+            // Both connections should receive the broadcast message
+            assertEquals("Hello World", message1);
+            assertEquals("Hello World", message2);
+
+            ws1.close();
+            ws2.close();
+
+            // Note: The actual verification of multiple FOLLOWS_FROM references needs to be done
+            // by inspecting the Jaeger API output, as we have 2 different HTTP upgrade requests
+            // that should both be linked to the Timer → WebSocket producer span
+            System.out.println("✅ Multiple connections established and received broadcast messages");
+            System.out.println("📊 Check Jaeger for Producer span with 2 FOLLOWS_FROM references");
+        } finally {
+            vertx.close();
+        }
+    }
+}

integration-tests/vertx-websocket/src/test/java/org/apache/camel/quarkus/component/vertx/websocket/it/VertxWebsocketSpanLinkIT.java

@@ -0,0 +1,23 @@
+/*
+ * 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.quarkus.component.vertx.websocket.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+
+@QuarkusIntegrationTest
+public class VertxWebsocketSpanLinkIT extends VertxWebsocketSpanLinkTest {
+}