Adding the client side integration of otel

Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>

Author: ehsavoie

client/base/src/main/java/io/a2a/client/ClientBuilder.java

@@ -5,26 +5,32 @@
 import io.a2a.client.transport.spi.ClientTransportConfig;
 import io.a2a.client.transport.spi.ClientTransportConfigBuilder;
 import io.a2a.client.transport.spi.ClientTransportProvider;
+import io.a2a.client.transport.spi.ClientTransportWrapper;
 import io.a2a.spec.A2AClientException;
 import io.a2a.spec.AgentCard;
 import io.a2a.spec.AgentInterface;
 import io.a2a.spec.TransportProtocol;
+import org.jspecify.annotations.NonNull;
+import org.jspecify.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
+import java.util.ServiceLoader.Provider;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
-import org.jspecify.annotations.NonNull;
-import org.jspecify.annotations.Nullable;
+import java.util.stream.Collectors;
 
 public class ClientBuilder {
 
     private static final Map<String, ClientTransportProvider<? extends ClientTransport, ? extends ClientTransportConfig<?>>> transportProviderRegistry = new HashMap<>();
     private static final Map<Class<? extends ClientTransport>, String> transportProtocolMapping = new HashMap<>();
+    private static final Logger LOGGER = LoggerFactory.getLogger(ClientBuilder.class);
 
     static {
         ServiceLoader<ClientTransportProvider> loader = ServiceLoader.load(ClientTransportProvider.class);
@@ -37,7 +43,8 @@ public class ClientBuilder {
     private final AgentCard agentCard;
 
     private final List<BiConsumer<ClientEvent, AgentCard>> consumers = new ArrayList<>();
-    private @Nullable Consumer<Throwable> streamErrorHandler;
+    private @Nullable
+    Consumer<Throwable> streamErrorHandler;
     private ClientConfig clientConfig = new ClientConfig.Builder().build();
 
     private final Map<Class<? extends ClientTransport>, ClientTransportConfig<? extends ClientTransport>> clientTransports = new LinkedHashMap<>();
@@ -105,7 +112,7 @@ private ClientTransport buildClientTransport() throws A2AClientException {
             throw new A2AClientException("Missing required TransportConfig for " + agentInterface.transport());
         }
 
-        return clientTransportProvider.create(clientTransportConfig, agentCard, agentInterface.url());
+        return wrap(clientTransportProvider.create(clientTransportConfig, agentCard, agentInterface.url()), clientTransportConfig);
     }
 
     private Map<String, String> getServerPreferredTransports() {
@@ -160,10 +167,50 @@ private AgentInterface findBestClientTransport() throws A2AClientException {
         if (transportProtocol == null || transportUrl == null) {
             throw new A2AClientException("No compatible transport found");
         }
-        if (! transportProviderRegistry.containsKey(transportProtocol)) {
+        if (!transportProviderRegistry.containsKey(transportProtocol)) {
             throw new A2AClientException("No client available for " + transportProtocol);
         }
 
         return new AgentInterface(transportProtocol, transportUrl);
     }
+
+    /**
+     * Wraps the transport with all available transport wrappers discovered via ServiceLoader.
+     * Wrappers are applied in priority order (highest priority first).
+     *
+     * @param transport the base transport to wrap
+     * @param clientTransportConfig the transport configuration
+     * @return the wrapped transport (or original if no wrappers are available/applicable)
+     */
+    private ClientTransport wrap(ClientTransport transport, ClientTransportConfig<? extends ClientTransport> clientTransportConfig) {
+        ServiceLoader<ClientTransportWrapper> wrapperLoader = ServiceLoader.load(ClientTransportWrapper.class);
+
+        // Collect all wrappers and sort by natural order (uses Comparable implementation)
+        List<ClientTransportWrapper> wrappers = wrapperLoader.stream().map(Provider::get)
+                .sorted()
+                .collect(Collectors.toList());
+
+        if (wrappers.isEmpty()) {
+            LOGGER.debug("No client transport wrappers found via ServiceLoader");
+            return transport;
+        }
+
+        // Apply wrappers in priority order
+        ClientTransport wrapped = transport;
+        for (ClientTransportWrapper wrapper : wrappers) {
+            try {
+                ClientTransport newWrapped = wrapper.wrap(wrapped, clientTransportConfig);
+                if (newWrapped != wrapped) {
+                    LOGGER.debug("Applied transport wrapper: {} (priority: {})",
+                            wrapper.getClass().getName(), wrapper.priority());
+                }
+                wrapped = newWrapped;
+            } catch (Exception e) {
+                LOGGER.warn("Failed to apply transport wrapper {}: {}",
+                        wrapper.getClass().getName(), e.getMessage(), e);
+            }
+        }
+
+        return wrapped;
+    }
 }

client/transport/rest/src/main/java/io/a2a/client/transport/rest/RestTransportConfig.java

@@ -4,7 +4,7 @@
 import io.a2a.client.transport.spi.ClientTransportConfig;
 import org.jspecify.annotations.Nullable;
 
-public class RestTransportConfig extends ClientTransportConfig<RestTransport>  {
+public class RestTransportConfig extends ClientTransportConfig<RestTransport> {
 
     private final @Nullable A2AHttpClient httpClient;
 

client/transport/spi/src/main/java/io/a2a/client/transport/spi/ClientTransportConfig.java

@@ -2,15 +2,18 @@
 
 import io.a2a.client.transport.spi.interceptors.ClientCallInterceptor;
 import java.util.ArrayList;
+import java.util.HashMap;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * Configuration for an A2A client transport.
  */
 public abstract class ClientTransportConfig<T extends ClientTransport> {
 
     protected List<ClientCallInterceptor> interceptors = new ArrayList<>();
+    protected Map<String, ? extends Object > parameters = new HashMap<>();
 
     public void setInterceptors(List<ClientCallInterceptor> interceptors) {
         this.interceptors = new ArrayList<>(interceptors);
@@ -19,4 +22,12 @@ public void setInterceptors(List<ClientCallInterceptor> interceptors) {
     public List<ClientCallInterceptor> getInterceptors() {
         return interceptors;
     }
+
+    public void setParameters(Map<String, ? extends Object > parameters) {
+        this.parameters = new HashMap<>(parameters);
+    }
+
+    public Map<String, ? extends Object > getParameters() {
+        return parameters;
+    }
 }

client/transport/spi/src/main/java/io/a2a/client/transport/spi/ClientTransportWrapper.java

@@ -0,0 +1,81 @@
+package io.a2a.client.transport.spi;
+
+/**
+ * Service provider interface for wrapping client transports with additional functionality.
+ * Implementations can add cross-cutting concerns like tracing, metrics, logging, etc.
+ *
+ * <p>Wrappers are discovered via Java's ServiceLoader mechanism. To register a wrapper,
+ * create a file {@code META-INF/services/io.a2a.client.transport.spi.ClientTransportWrapper}
+ * containing the fully qualified class name of your implementation.
+ *
+ * <p>Wrappers are sorted by priority in descending order (highest priority first).
+ * This interface implements {@link Comparable} to enable natural sorting.
+ *
+ * <p>Example implementation:
+ * <pre>{@code
+ * public class TracingWrapper implements ClientTransportWrapper {
+ *     @Override
+ *     public ClientTransport wrap(ClientTransport transport, ClientTransportConfig<?> config) {
+ *         if (config.getParameters().containsKey("tracer")) {
+ *             return new TracingTransport(transport, (Tracer) config.getParameters().get("tracer"));
+ *         }
+ *         return transport;
+ *     }
+ *
+ *     @Override
+ *     public int priority() {
+ *         return 100; // Higher priority = wraps earlier (outermost)
+ *     }
+ * }
+ * }</pre>
+ */
+public interface ClientTransportWrapper extends Comparable<ClientTransportWrapper> {
+
+    /**
+     * Wraps the given transport with additional functionality.
+     *
+     * <p>Implementations should check the configuration to determine if they should
+     * actually wrap the transport. If the wrapper is not applicable (e.g., required
+     * configuration is missing), return the original transport unchanged.
+     *
+     * @param transport the transport to wrap
+     * @param config the transport configuration, may contain wrapper-specific parameters
+     * @return the wrapped transport, or the original if wrapping is not applicable
+     */
+    ClientTransport wrap(ClientTransport transport, ClientTransportConfig<?> config);
+
+    /**
+     * Returns the priority of this wrapper. Higher priority wrappers are applied first
+     * (wrap the transport earlier, resulting in being the outermost wrapper).
+     *
+     * <p>Default priority is 0. Suggested ranges:
+     * <ul>
+     *   <li>1000+ : Critical infrastructure (security, authentication)
+     *   <li>500-999: Observability (tracing, metrics, logging)
+     *   <li>100-499: Enhancement (caching, retry logic)
+     *   <li>0-99: Optional features
+     * </ul>
+     *
+     * @return the priority value, higher values = higher priority
+     */
+    default int priority() {
+        return 0;
+    }
+
+    /**
+     * Compares this wrapper with another based on priority.
+     * Returns a negative integer, zero, or a positive integer as this wrapper
+     * has higher priority than, equal to, or lower priority than the specified wrapper.
+     *
+     * <p>Note: This comparison is reversed (higher priority comes first) to enable
+     * natural sorting in descending priority order.
+     *
+     * @param other the wrapper to compare to
+     * @return negative if this has higher priority, positive if lower, zero if equal
+     */
+    @Override
+    default int compareTo(ClientTransportWrapper other) {
+        // Reverse comparison: higher priority should come first
+        return Integer.compare(other.priority(), this.priority());
+    }
+}

examples/helloworld/client/pom.xml

@@ -20,6 +20,22 @@
             <groupId>io.github.a2asdk</groupId>
             <artifactId>a2a-java-sdk-client</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.github.a2asdk</groupId>
+            <artifactId>a2a-java-sdk-opentelemetry</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-sdk</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-otlp</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.opentelemetry</groupId>
+            <artifactId>opentelemetry-exporter-logging</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

examples/helloworld/client/src/main/java/io/a2a/examples/helloworld/HelloWorldClient.java

@@ -12,7 +12,6 @@
 import io.a2a.A2A;
 
 import io.a2a.client.Client;
-import io.a2a.client.ClientBuilder;
 import io.a2a.client.ClientEvent;
 import io.a2a.client.MessageEvent;
 import io.a2a.client.http.A2ACardResolver;
@@ -22,6 +21,12 @@
 import io.a2a.spec.Message;
 import io.a2a.spec.Part;
 import io.a2a.spec.TextPart;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
 
 /**
  * A simple example of using the A2A Java SDK to communicate with an A2A server.
@@ -34,7 +39,13 @@ public class HelloWorldClient {
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
     public static void main(String[] args) {
+        
         try {
+            String endpoint = "http://localhost:5317";
+            if(args != null && args.length > 0) {
+                endpoint = args[0];
+            }
+            OpenTelemetry openTelemetrySdk = initOpenTelemetry(endpoint);
             AgentCard finalAgentCard = null;
             AgentCard publicAgentCard = new A2ACardResolver("http://localhost:9999").getAgentCard();
             System.out.println("Successfully fetched public agent card:");
@@ -82,16 +93,18 @@ public static void main(String[] args) {
                 error.printStackTrace();
                 messageResponse.completeExceptionally(error);
             };
-
+            JSONRPCTransportConfig transportConfig = new JSONRPCTransportConfig();
+            transportConfig.setParameters(Map.of("io.a2a.extras.opentelemetry.Tracer",
+                    openTelemetrySdk.getTracer("helloworld-client")));
             Client client = Client
                     .builder(finalAgentCard)
                     .addConsumers(consumers)
                     .streamingErrorHandler(streamingErrorHandler)
-                    .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfig())
+                    .withTransport(JSONRPCTransport.class, transportConfig)
                     .build();
 
             Message message = A2A.toUserMessage(MESSAGE_TEXT); // the message ID will be automatically generated for you
-            
+
             System.out.println("Sending message: " + MESSAGE_TEXT);
             client.sendMessage(message);
             System.out.println("Message sent successfully. Responses will be handled by the configured consumers.");
@@ -108,4 +121,17 @@ public static void main(String[] args) {
         }
     }
 
-} 
+    static OpenTelemetry initOpenTelemetry(String endpoint) {
+        SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
+                .addSpanProcessor(SimpleSpanProcessor.create(OtlpGrpcSpanExporter.builder().setEndpoint(endpoint).build()))
+                .setResource(Resource.getDefault().toBuilder().put("service.version", "1.0").put("service.name", "helloworld-client").build())
+                .build();
+
+        OpenTelemetrySdk openTelemetrySdk = OpenTelemetrySdk.builder()
+                .setTracerProvider(sdkTracerProvider)
+                .build();
+
+        Runtime.getRuntime().addShutdownHook(new Thread(sdkTracerProvider::close));
+        return openTelemetrySdk;
+    }
+}

examples/helloworld/pom.xml

@@ -34,25 +34,6 @@
         </dependencies>
     </dependencyManagement>
 
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>io.quarkus</groupId>
-                <artifactId>quarkus-maven-plugin</artifactId>
-                <extensions>true</extensions>
-                <executions>
-                    <execution>
-                        <goals>
-                            <goal>build</goal>
-                            <goal>generate-code</goal>
-                            <goal>generate-code-tests</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
-        </plugins>
-    </build>
-
     <modules>
         <module>client</module>
         <module>server</module>

examples/helloworld/server/pom.xml

@@ -37,7 +37,6 @@
         <dependency>
             <groupId>io.github.a2asdk</groupId>
             <artifactId>a2a-java-sdk-opentelemetry</artifactId>
-            <version>0.4.0.Alpha1-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>jakarta.enterprise</groupId>
@@ -48,6 +47,11 @@
             <groupId>jakarta.ws.rs</groupId>
             <artifactId>jakarta.ws.rs-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-observability-devservices-lgtm</artifactId>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -65,6 +69,9 @@
                         </goals>
                     </execution>
                 </executions>
+                <configuration>
+                    <jvmArgs>--add-opens=java.base/java.lang=ALL-UNNAMED</jvmArgs>
+                </configuration>
             </plugin>
         </plugins>
     </build>

examples/helloworld/server/src/main/resources/application.properties

@@ -2,5 +2,8 @@
 
 # OpenTelemetry configuration
 quarkus.otel.sdk.disabled=false
-quarkus.otel.exporter.otlp.endpoint=http://localhost:4317
-quarkus.log.console.format=%d{HH:mm:ss} %-5p traceId=%X{traceId}, parentId=%X{parentId}, spanId=%X{spanId}, sampled=%X{sampled} [%c{2.}] (%t) %s%e%n
+quarkus.observability.lgtm.grafana-port=3001
+quarkus.observability.lgtm.otel-grpc-port=5317
+quarkus.observability.lgtm.otel-http-port=5318
+#quarkus.otel.exporter.otlp.endpoint=http://localhost:4317
+#quarkus.log.console.format=%d{HH:mm:ss} %-5p traceId=%X{traceId}, parentId=%X{parentId}, spanId=%X{spanId}, sampled=%X{sampled} [%c{2.}] (%t) %s%e%n