davsclaus commented on code in PR #24363: URL: https://github.com/apache/camel/pull/24363#discussion_r3773230804
########## components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/VertxWebsocketSpanDecorator.java: ########## @@ -0,0 +1,185 @@ +/* + * 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.telemetry.decorators; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.telemetry.SpanContextPropagationExtractor; + +public class VertxWebsocketSpanDecorator extends AbstractSpanDecorator { + + // Constants from VertxWebsocketConstants - duplicated to avoid compile-time dependency + private static final String HANDSHAKE_SPAN_CONTEXT_KEY = "CamelVertxWebsocketHandshakeSpanContext"; + private static final String SEND_TO_ALL = "CamelVertxWebsocket.sendToAll"; + private static final String CONNECTION_KEY = "CamelVertxWebsocket.connectionKey"; + + @Override + public String getComponent() { + return "vertx-websocket"; + } + + @Override + public String getComponentClassName() { + return "org.apache.camel.component.vertx.websocket.VertxWebsocketComponent"; + } + + @Override + public SpanContextPropagationExtractor getExtractor(Exchange exchange) { + return new VertxWebsocketSpanContextPropagationExtractor(exchange); + } + + /** + * Custom extractor for vertx-websocket that ensures the handshake span context header is visible to OpenTelemetry + * when creating span links, even for producer spans. + */ + private static class VertxWebsocketSpanContextPropagationExtractor implements SpanContextPropagationExtractor { + private final Map<String, Object> headers; + private final Exchange exchange; + + VertxWebsocketSpanContextPropagationExtractor(Exchange exchange) { + this.exchange = exchange; + this.headers = exchange.getIn().getHeaders(); + // Proactively collect span context for producers + collectProducerSpanContext(); + } + + /** + * For producer scenarios, collect the handshake span contexts from target WebSocket peers and store them in + * headers so they're available when OpenTelemetryTracer creates the span. + * <p> + * Supports multiple span links for scenarios like sendToAll to multiple connections from different HTTP + * requests. + * <p> + * Only collects span contexts for peers that will actually receive the message based on sendToAll and + * connectionKey settings. + */ + private void collectProducerSpanContext() { + if (headers.containsKey(HANDSHAKE_SPAN_CONTEXT_KEY)) { + // Already set by Consumer, nothing to do + return; + } + + // This is a Producer - collect span contexts from target peers using reflection + try { + // Get the endpoint from exchange context + Object endpoint = exchange.getContext().hasEndpoint(exchange.getProperty(Exchange.TO_ENDPOINT, String.class)); + if (endpoint == null) { + return; + } + + // Get all peers + Method findPeersMethod = endpoint.getClass().getMethod("findPeerObjectsForHostPort"); + @SuppressWarnings("unchecked") + List<Object> allPeers = (List<Object>) findPeersMethod.invoke(endpoint); + + if (allPeers == null || allPeers.isEmpty()) { Review Comment: `exchange.getContext().hasEndpoint(...)` is unreliable for endpoint lookup — it only returns an endpoint if it happens to be registered in the `EndpointRegistry` by that exact URI. This may return `null` in many valid scenarios (e.g., if the URI doesn't match exactly, or the registry has been cleaned up). ########## components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetryTracer.java: ########## @@ -180,6 +183,11 @@ protected void doShutdown() { private class OpentelemetrySpanLifecycleManager implements SpanLifecycleManager { private final static String BAGGAGE_VAR_PREFIX = "OTEL_BAGGAGE_"; + /** + * Exchange property name for storing a span context to create span links. This is used by components like + * vertx-websocket to link WebSocket message spans back to the original HTTP upgrade request span. Review Comment: This hardcodes a vertx-websocket-specific constant (`CamelVertxWebsocketHandshakeSpanContext`) inside the generic OpenTelemetry tracer. The `camel-opentelemetry2` module should not know about individual component internals. The span link concept should be added to the `camel-telemetry` abstraction layer (e.g., `SpanLifecycleManager.create()`) so any component can benefit from it, not just vertx-websocket. ########## components/camel-opentelemetry2/src/main/java/org/apache/camel/opentelemetry2/OpenTelemetryTracer.java: ########## @@ -231,6 +247,49 @@ public String get(SpanContextPropagationExtractor carrier, String key) { return new OpenTelemetrySpanAdapter(builder.startSpan(), baggage); } + /** + * Extracts span contexts from the exchange properties to create span links. This allows linking spans across + * asynchronous boundaries, such as WebSocket messages back to the HTTP upgrade request. + * <p> + * Supports multiple span links for scenarios like broadcasting to multiple WebSocket connections from different + * HTTP requests. + * + * @param extractor the span context propagation extractor (usually the Exchange) + * @return list of span contexts to link to (empty if none present) + */ + private List<SpanContext> extractSpanLinkContexts(SpanContextPropagationExtractor extractor) { + List<SpanContext> result = new ArrayList<>(); + if (extractor == null) { + return result; + } + + Object value = extractor.get(SPAN_LINK_CONTEXT_PROPERTY); + + if (value instanceof SpanContext) { + result.add((SpanContext) value); + } else if (value instanceof String) { + try { + String str = (String) value; + // Format: "traceId1:spanId1,traceId2:spanId2,..." + // Split by comma to support multiple span links + for (String part : str.split(",")) { Review Comment: The custom `traceId:spanId` serialization format is fragile and non-standard. If span links become a first-class feature in the telemetry abstraction, consider using W3C Trace Context (`traceparent` header format) for serialization, which is already the standard used by the context propagation layer. ########## components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketHost.java: ########## @@ -85,6 +86,10 @@ public void connect(VertxWebsocketConsumer consumer) { } route.handler(routingContext -> { + // Capture HTTP request span context BEFORE WebSocket upgrade Review Comment: Capturing span context in the HTTP upgrade handler is the right idea — this is the correct point in the lifecycle where the HTTP span is still active. However, the reflection approach here (while avoiding compile-time dependency) means this code won't produce compile errors if the OTel API changes. Consider whether a conditional import / optional dependency pattern would be more maintainable. ########## components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketEndpoint.java: ########## @@ -250,9 +251,10 @@ protected ServerWebSocket findPeerForConnectionKey(String connectionKey) { } /** - * Finds all WebSockets associated with a host matching this endpoint configured port and resource path + * Finds all VertxWebsocketPeer objects associated with a host matching this endpoint configured port and resource + * path Review Comment: This changes API visibility from `protected` to `public` and changes the return type from `Map<String, ServerWebSocket>` to `List<VertxWebsocketPeer>`. The existing `findPeersForHostPort()` is kept but now delegates to this new method. Per project standards, public API changes require justification. If the only consumer is the reflection-based span decorator, consider whether a different design would avoid the need for this public API. ########## components/camel-telemetry/src/main/java/org/apache/camel/telemetry/decorators/VertxWebsocketSpanDecorator.java: ########## @@ -0,0 +1,185 @@ +/* + * 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.telemetry.decorators; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.telemetry.SpanContextPropagationExtractor; + +public class VertxWebsocketSpanDecorator extends AbstractSpanDecorator { + + // Constants from VertxWebsocketConstants - duplicated to avoid compile-time dependency + private static final String HANDSHAKE_SPAN_CONTEXT_KEY = "CamelVertxWebsocketHandshakeSpanContext"; + private static final String SEND_TO_ALL = "CamelVertxWebsocket.sendToAll"; + private static final String CONNECTION_KEY = "CamelVertxWebsocket.connectionKey"; + + @Override + public String getComponent() { + return "vertx-websocket"; + } + + @Override + public String getComponentClassName() { + return "org.apache.camel.component.vertx.websocket.VertxWebsocketComponent"; + } + + @Override + public SpanContextPropagationExtractor getExtractor(Exchange exchange) { + return new VertxWebsocketSpanContextPropagationExtractor(exchange); + } + + /** + * Custom extractor for vertx-websocket that ensures the handshake span context header is visible to OpenTelemetry + * when creating span links, even for producer spans. + */ + private static class VertxWebsocketSpanContextPropagationExtractor implements SpanContextPropagationExtractor { + private final Map<String, Object> headers; + private final Exchange exchange; + + VertxWebsocketSpanContextPropagationExtractor(Exchange exchange) { + this.exchange = exchange; + this.headers = exchange.getIn().getHeaders(); + // Proactively collect span context for producers + collectProducerSpanContext(); + } + + /** + * For producer scenarios, collect the handshake span contexts from target WebSocket peers and store them in + * headers so they're available when OpenTelemetryTracer creates the span. + * <p> + * Supports multiple span links for scenarios like sendToAll to multiple connections from different HTTP + * requests. + * <p> + * Only collects span contexts for peers that will actually receive the message based on sendToAll and + * connectionKey settings. + */ + private void collectProducerSpanContext() { + if (headers.containsKey(HANDSHAKE_SPAN_CONTEXT_KEY)) { + // Already set by Consumer, nothing to do + return; + } + + // This is a Producer - collect span contexts from target peers using reflection + try { + // Get the endpoint from exchange context + Object endpoint = exchange.getContext().hasEndpoint(exchange.getProperty(Exchange.TO_ENDPOINT, String.class)); + if (endpoint == null) { + return; + } + + // Get all peers + Method findPeersMethod = endpoint.getClass().getMethod("findPeerObjectsForHostPort"); + @SuppressWarnings("unchecked") + List<Object> allPeers = (List<Object>) findPeersMethod.invoke(endpoint); + + if (allPeers == null || allPeers.isEmpty()) { + return; + } + + // Determine which peers will actually receive the message Review Comment: This reflection call assumes the endpoint has a method `findPeerObjectsForHostPort()`. The vertx-websocket PR adds this as a `public` method, but the decorator in `camel-telemetry` cannot compile-time verify it exists. If the method is ever renamed, this will silently fail (caught by the empty `catch` on line 170). Consider whether a different integration approach would be more robust. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
