From 28362738f79ca00eabcdf18f05f0819d403f3651 Mon Sep 17 00:00:00 2001 From: Emmanuel Hugonnet Date: Wed, 21 Oct 2026 15:34:33 +0200 Subject: [PATCH] feat(opentelemetry): add streaming span lifecycle and operation metrics - Per-event child spans and terminal closing span for streaming methods - Duration histograms for client (gen_ai.client.operation.duration) and server (gen_ai.agent.a2a.streaming.duration) - Signal REST stream completion to terminal consumer for proper span closing Signed-off-by: Emmanuel Hugonnet --- .../transport/grpc/EventStreamObserver.java | 13 +- .../grpc/EventStreamObserverTest.java | 37 ++ .../client/transport/rest/RestTransport.java | 6 +- .../transport/rest/sse/SSEEventListener.java | 13 +- .../transport/rest/RestTransportTest.java | 84 +++ .../client/transport/spi/ClientTransport.java | 6 +- .../grpc/EventStreamObserver_v0_3.java | 13 +- .../grpc/EventStreamObserver_v0_3_Test.java | 37 ++ docs/content/dev/extra/opentelemetry.md | 95 ++++ examples/helloworld/client/README.md | 4 +- .../helloworld/client/HelloWorldClient.java | 25 +- .../src/main/resources/application.properties | 1 + extras/opentelemetry/client/pom.xml | 6 + .../client/OpenTelemetryClientTransport.java | 184 +++++-- .../OpenTelemetryClientTransportWrapper.java | 14 +- .../OpenTelemetryClientTransportTest.java | 186 +++++-- extras/opentelemetry/common/pom.xml | 34 ++ .../opentelemetry/A2AObservabilityNames.java | 26 +- .../StreamingEventAttributes.java | 87 +++ .../extras/opentelemetry/OtelSpanMocks.java | 54 ++ .../StreamingEventAttributesTest.java | 60 ++ .../opentelemetry/integration-tests/pom.xml | 10 + .../opentelemetry/it/A2ATestRoutes.java | 97 +++- .../src/main/resources/application.properties | 3 +- .../opentelemetry/it/A2ATestRoutesTest.java | 98 ++++ .../sdk/extras/opentelemetry/it/BaseTest.java | 5 + .../it/OpenTelemetryA2ABaseTest.java | 91 +++ .../src/test/resources/application.properties | 3 +- extras/opentelemetry/server/pom.xml | 6 +- .../OpenTelemetryRequestHandlerDecorator.java | 267 ++++++++- .../src/main/resources/META-INF/beans.xml | 2 +- ...nTelemetryRequestHandlerDecoratorTest.java | 519 +++++++++++++++++- pom.xml | 6 + 33 files changed, 1939 insertions(+), 153 deletions(-) create mode 100644 client/transport/grpc/src/test/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserverTest.java create mode 100644 compat-0.3/client/transport/grpc/src/test/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3_Test.java create mode 100644 extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributes.java create mode 100644 extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/OtelSpanMocks.java create mode 100644 extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributesTest.java create mode 100644 extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutesTest.java diff --git a/client/transport/grpc/src/main/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserver.java b/client/transport/grpc/src/main/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserver.java index 2b7c36ffa..efa28fd72 100644 --- a/client/transport/grpc/src/main/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserver.java +++ b/client/transport/grpc/src/main/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserver.java @@ -9,14 +9,15 @@ import org.a2aproject.sdk.grpc.StreamResponse; import org.a2aproject.sdk.spec.StreamingEventKind; import io.grpc.stub.StreamObserver; +import org.jspecify.annotations.Nullable; public class EventStreamObserver implements StreamObserver { private static final Logger LOGGER = Logger.getLogger(EventStreamObserver.class.getName()); private final Consumer eventHandler; - private final Consumer errorHandler; + @Nullable private final Consumer errorHandler; - public EventStreamObserver(Consumer eventHandler, Consumer errorHandler) { + public EventStreamObserver(Consumer eventHandler, @Nullable Consumer errorHandler) { this.eventHandler = eventHandler; this.errorHandler = errorHandler; } @@ -39,7 +40,9 @@ public void onNext(StreamResponse response) { break; default: LOGGER.warning("Invalid stream response " + response.getPayloadCase()); - errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase())); + if (errorHandler != null) { + errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase())); + } return; } eventHandler.accept(event); @@ -59,6 +62,8 @@ public void onError(Throwable t) { @Override public void onCompleted() { - // done + if (errorHandler != null) { + errorHandler.accept(null); + } } } diff --git a/client/transport/grpc/src/test/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserverTest.java b/client/transport/grpc/src/test/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserverTest.java new file mode 100644 index 000000000..5b5cfe673 --- /dev/null +++ b/client/transport/grpc/src/test/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserverTest.java @@ -0,0 +1,37 @@ +package org.a2aproject.sdk.client.transport.grpc; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; +import org.a2aproject.sdk.spec.StreamingEventKind; +import org.junit.jupiter.api.Test; + +class EventStreamObserverTest { + + @Test + void onCompleted_callsErrorHandlerWithNull() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + + Consumer eventHandler = events::add; + Consumer errorHandler = errors::add; + EventStreamObserver observer = new EventStreamObserver(eventHandler, errorHandler); + + observer.onCompleted(); + + assertEquals(1, errors.size(), "errorHandler should have been called once"); + assertNull(errors.get(0), "errorHandler should be called with null to signal clean completion"); + assertEquals(0, events.size(), "eventHandler should not be called"); + } + + @Test + void onCompleted_withNullErrorHandler_doesNotThrow() { + Consumer eventHandler = e -> {}; + EventStreamObserver observer = new EventStreamObserver(eventHandler, null); + + observer.onCompleted(); + } +} diff --git a/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/RestTransport.java b/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/RestTransport.java index 1f6535250..68449dcbb 100644 --- a/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/RestTransport.java +++ b/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/RestTransport.java @@ -121,7 +121,8 @@ public void sendMessageStreaming(MessageSendParams messageSendParams, Consumer sseEventListener.onMessage(event, ref.get()), throwable -> sseEventListener.onError(throwable, ref.get()), () -> { - // We don't need to do anything special on completion + // Signal normal stream completion to error handler (null error means success) + sseEventListener.onComplete(); })); } catch (IOException e) { throw new A2AClientException("Failed to send streaming message request: " + e, e); @@ -381,7 +382,8 @@ public void subscribeToTask(TaskIdParams request, Consumer e event -> sseEventListener.onMessage(event, ref.get()), throwable -> sseEventListener.onError(throwable, ref.get()), () -> { - // We don't need to do anything special on completion + // Signal normal stream completion to error handler (null error means success) + sseEventListener.onComplete(); })); } catch (IOException e) { throw new A2AClientException("Failed to send streaming message request: " + e, e); diff --git a/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/sse/SSEEventListener.java b/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/sse/SSEEventListener.java index d2507c64e..0efe85f62 100644 --- a/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/sse/SSEEventListener.java +++ b/client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/sse/SSEEventListener.java @@ -27,6 +27,11 @@ public SSEEventListener(Consumer eventHandler, super(eventHandler, errorHandler); } + public void onComplete() { + LOGGER.fine("SSEEventListener.onComplete() called - signaling successful stream completion"); + signalTerminal(null); + } + @Override public void onMessage(ServerSentEvent event, @Nullable Future completableFuture) { try { @@ -35,9 +40,7 @@ public void onMessage(ServerSentEvent event, @Nullable Future completableF JsonFormat.parser().merge(event.data(), builder); parseAndHandleMessage(builder.build(), completableFuture); } catch (InvalidProtocolBufferException e) { - if (getErrorHandler() != null) { - getErrorHandler().accept(RestErrorMapper.mapRestError(event.data(), 500)); - } + signalTerminal(RestErrorMapper.mapRestError(event.data(), 500)); } } @@ -60,9 +63,7 @@ private void parseAndHandleMessage(StreamResponse response, @Nullable Future { LOGGER.warning("Invalid stream response " + response.getPayloadCase()); - if (getErrorHandler() != null) { - getErrorHandler().accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase())); - } + signalTerminal(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase())); return; } } diff --git a/client/transport/rest/src/test/java/org/a2aproject/sdk/client/transport/rest/RestTransportTest.java b/client/transport/rest/src/test/java/org/a2aproject/sdk/client/transport/rest/RestTransportTest.java index 05d1b7b12..bed763f8e 100644 --- a/client/transport/rest/src/test/java/org/a2aproject/sdk/client/transport/rest/RestTransportTest.java +++ b/client/transport/rest/src/test/java/org/a2aproject/sdk/client/transport/rest/RestTransportTest.java @@ -69,6 +69,7 @@ public class RestTransportTest { private static final Logger LOGGER = Logger.getLogger(RestTransportTest.class.getName()); + private static final Throwable SENTINEL = new RuntimeException("not yet called"); private ClientAndServer server; private static final AgentCard CARD = AgentCard.builder() .name("Hello World Agent") @@ -474,6 +475,89 @@ public void testSubscribe() throws Exception { assertEquals("Why did the chicken cross the road? To get to the other side!", ((TextPart) part).text()); } + /** + * Test that the error handler is called with null on normal stream completion for sendMessageStreaming. + */ + @Test + public void testSendMessageStreamingCompletion() throws Exception { + this.server.when( + request() + .withMethod("POST") + .withPath("/message:stream") + .withBody(JsonBody.json(SEND_MESSAGE_STREAMING_TEST_REQUEST, MatchType.ONLY_MATCHING_FIELDS)) + ) + .respond( + response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody(SEND_MESSAGE_STREAMING_TEST_RESPONSE) + ); + + RestTransport client = new RestTransport(CARD); + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(Collections.singletonList(new TextPart("tell me some jokes"))) + .contextId("context-1234") + .messageId("message-1234") + .build(); + MessageSendConfiguration configuration = MessageSendConfiguration.builder() + .acceptedOutputModes(List.of("text")) + .returnImmediately(true) + .build(); + MessageSendParams params = MessageSendParams.builder() + .message(message) + .configuration(configuration) + .build(); + CountDownLatch eventLatch = new CountDownLatch(1); + CountDownLatch completionLatch = new CountDownLatch(1); + AtomicReference completionValue = new AtomicReference<>(SENTINEL); + Consumer eventHandler = event -> eventLatch.countDown(); + Consumer errorHandler = error -> { + completionValue.set(error); + completionLatch.countDown(); + }; + client.sendMessageStreaming(params, eventHandler, errorHandler, null); + + assertTrue(eventLatch.await(10, TimeUnit.SECONDS), "Should receive streaming event"); + assertTrue(completionLatch.await(10, TimeUnit.SECONDS), "Error handler should be called on stream completion"); + assertNull(completionValue.get(), "Error handler should be called with null for successful completion"); + } + + /** + * Test that the error handler is called with null on normal stream completion for subscribeToTask. + */ + @Test + public void testSubscribeToTaskCompletion() throws Exception { + this.server.when( + request() + .withMethod("POST") + .withPath("/tasks/task-1234:subscribe") + ) + .respond( + response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody(TASK_RESUBSCRIPTION_REQUEST_TEST_RESPONSE) + ); + + RestTransport client = new RestTransport(CARD); + TaskIdParams taskIdParams = new TaskIdParams("task-1234"); + + CountDownLatch eventLatch = new CountDownLatch(1); + CountDownLatch completionLatch = new CountDownLatch(1); + AtomicReference completionValue = new AtomicReference<>(SENTINEL); + Consumer eventHandler = event -> eventLatch.countDown(); + Consumer errorHandler = error -> { + completionValue.set(error); + completionLatch.countDown(); + }; + client.subscribeToTask(taskIdParams, eventHandler, errorHandler, null); + + assertTrue(eventLatch.await(10, TimeUnit.SECONDS), "Should receive streaming event"); + assertTrue(completionLatch.await(10, TimeUnit.SECONDS), "Error handler should be called on stream completion"); + assertNull(completionValue.get(), "Error handler should be called with null for successful completion"); + } + /** * Test that ExtensionSupportRequiredError is properly unmarshalled from REST error response. */ diff --git a/client/transport/spi/src/main/java/org/a2aproject/sdk/client/transport/spi/ClientTransport.java b/client/transport/spi/src/main/java/org/a2aproject/sdk/client/transport/spi/ClientTransport.java index 0e3d9bf82..4eb1393a3 100644 --- a/client/transport/spi/src/main/java/org/a2aproject/sdk/client/transport/spi/ClientTransport.java +++ b/client/transport/spi/src/main/java/org/a2aproject/sdk/client/transport/spi/ClientTransport.java @@ -43,7 +43,8 @@ EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext con * * @param request the message send parameters * @param eventConsumer consumer that will receive streaming events as they arrive - * @param errorConsumer consumer that will be called if an error occurs during streaming + * @param errorConsumer consumer called when the stream ends; called with {@code null} to signal + * clean completion, or with a non-null {@link Throwable} if an error occurs * @param context optional client call context for the request (may be {@code null}) * @throws A2AClientException if setting up the streaming connection fails */ @@ -131,7 +132,8 @@ void deleteTaskPushNotificationConfigurations( * * @param request the task ID parameters specifying which task to subscribe to * @param eventConsumer consumer that will receive streaming events as they arrive - * @param errorConsumer consumer that will be called if an error occurs during streaming + * @param errorConsumer consumer called when the stream ends; called with {@code null} to signal + * clean completion, or with a non-null {@link Throwable} if an error occurs * @param context optional client call context for the request (may be {@code null}) * @throws A2AClientException if subscribing to the task fails for any reason */ diff --git a/compat-0.3/client/transport/grpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3.java b/compat-0.3/client/transport/grpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3.java index 55458a93f..fc7e8a93c 100644 --- a/compat-0.3/client/transport/grpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3.java +++ b/compat-0.3/client/transport/grpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3.java @@ -8,15 +8,16 @@ import java.util.function.Consumer; import java.util.logging.Logger; +import org.jspecify.annotations.Nullable; import static org.a2aproject.sdk.compat03.grpc.utils.ProtoUtils_v0_3.FromProto; public class EventStreamObserver_v0_3 implements StreamObserver { private static final Logger LOGGER = Logger.getLogger(EventStreamObserver_v0_3.class.getName()); private final Consumer eventHandler; - private final Consumer errorHandler; + @Nullable private final Consumer errorHandler; - public EventStreamObserver_v0_3(Consumer eventHandler, Consumer errorHandler) { + public EventStreamObserver_v0_3(Consumer eventHandler, @Nullable Consumer errorHandler) { this.eventHandler = eventHandler; this.errorHandler = errorHandler; } @@ -39,7 +40,9 @@ public void onNext(StreamResponse response) { break; default: LOGGER.warning("Invalid stream response " + response.getPayloadCase()); - errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase())); + if (errorHandler != null) { + errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase())); + } return; } eventHandler.accept(event); @@ -59,6 +62,8 @@ public void onError(Throwable t) { @Override public void onCompleted() { - // done + if (errorHandler != null) { + errorHandler.accept(null); + } } } diff --git a/compat-0.3/client/transport/grpc/src/test/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3_Test.java b/compat-0.3/client/transport/grpc/src/test/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3_Test.java new file mode 100644 index 000000000..d8f26d14c --- /dev/null +++ b/compat-0.3/client/transport/grpc/src/test/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3_Test.java @@ -0,0 +1,37 @@ +package org.a2aproject.sdk.compat03.client.transport.grpc; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; +import org.a2aproject.sdk.compat03.spec.StreamingEventKind_v0_3; +import org.junit.jupiter.api.Test; + +class EventStreamObserver_v0_3_Test { + + @Test + void onCompleted_callsErrorHandlerWithNull() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + + Consumer eventHandler = events::add; + Consumer errorHandler = errors::add; + EventStreamObserver_v0_3 observer = new EventStreamObserver_v0_3(eventHandler, errorHandler); + + observer.onCompleted(); + + assertEquals(1, errors.size(), "errorHandler should have been called once"); + assertNull(errors.get(0), "errorHandler should be called with null to signal clean completion"); + assertEquals(0, events.size(), "eventHandler should not be called"); + } + + @Test + void onCompleted_withNullErrorHandler_doesNotThrow() { + Consumer eventHandler = e -> {}; + EventStreamObserver_v0_3 observer = new EventStreamObserver_v0_3(eventHandler, null); + + observer.onCompleted(); + } +} diff --git a/docs/content/dev/extra/opentelemetry.md b/docs/content/dev/extra/opentelemetry.md index a29e1b681..f084eeeed 100644 --- a/docs/content/dev/extra/opentelemetry.md +++ b/docs/content/dev/extra/opentelemetry.md @@ -14,6 +14,7 @@ Adds distributed tracing and context propagation to A2A servers and clients usin - **Context Propagation**: OpenTelemetry trace context propagation across async operations - **Request/Response Logging**: Optional extraction of request and response data into spans - **Error Tracking**: Automatic error status and error type attributes on failures +- **Metrics**: `gen_ai.client.operation.duration` histogram on the client side; `gen_ai.agent.a2a.streaming.duration` histogram for full streaming lifecycle on the server side ## Modules @@ -65,6 +66,76 @@ Async agent execution (maintains trace context) Response (with trace headers) ``` +### Streaming Span Lifecycle + +Streaming methods (`message/stream` and `tasks/resubscribe`) follow a two-span model to correctly represent the full duration of a streaming response. + +**Span 1 — method span** is created when the request arrives and ended as soon as the publisher is returned to the transport layer. It captures request attributes and the initial outcome (success or synchronous error). + +**Per-event child spans** (`-event`) are created for each streaming event published via `onNext()`. Each child span carries a [link](https://opentelemetry.io/docs/concepts/signals/traces/#span-links) back to Span 1 and is made the current span during delegate execution, so any nested telemetry is correctly parented. + +**Span 2 — end span** (`-end`) is created when the stream terminates — either on completion or error. It also carries a link to Span 1 for correlation. + +
+sequenceDiagram
+    participant C as Client
+    participant D as OTel Decorator
+    participant H as RequestHandler
+    participant W as WrappedPublisher
+
+    C->>D: onMessageSendStream(params)
+    D->>D: create Span₁ (SERVER)
+    D->>H: onMessageSendStream(params)
+    H-->>D: publisher
+    D->>D: capture Span₁.context
+    D->>D: Span₁.setStatus(OK) · Span₁.end()
+    D-->>C: WrappedPublisher
+
+    note over D: Span₁ closed immediately
+
+    loop for each streaming event
+        W->>W: create child span (link→Span₁)
+        W-->>C: onNext(event)
+        W->>W: child span.end()
+    end
+
+    alt stream completes normally
+        W->>W: create Span₂ (SERVER, link→Span₁)
+        W->>W: Span₂.setStatus(OK) · Span₂.end()
+        W-->>C: onComplete()
+    else stream errors
+        W->>W: create Span₂ (SERVER, link→Span₁)
+        W->>W: Span₂.setStatus(ERROR) · Span₂.end()
+        W-->>C: onError(throwable)
+    end
+
+ +Each per-event child span carries: + +| Attribute | Value | +|-----------|-------| +| `gen_ai.agent.a2a.streaming-event` | Event kind identifier (`"message"`, `"task"`, `"statusUpdate"`, `"artifactUpdate"`) | +| `gen_ai.agent.a2a.task.status.state` | Task status state (for `TaskStatusUpdateEvent` and `Task` events) | +| `gen_ai.agent.a2a.task_id` | Task identifier (when available from the event) | + +### Metrics + +The server decorator records the following histogram instruments: + +| Instrument | Type | Unit | Description | +|------------|------|------|-------------| +| `gen_ai.agent.a2a.streaming.duration` | Histogram | `s` | Duration from stream initiation to last event, for `message/stream` and `tasks/resubscribe` | + +Attributes on `gen_ai.agent.a2a.streaming.duration`: + +| Attribute | Value | +|-----------|-------| +| `gen_ai.agent.a2a.operation.name` | A2A method name (`message/stream` or `tasks/resubscribe`) | +| `gen_ai.system` | `a2a` | +| `error.type` | `error` (only on stream failure) | + +Meter injection uses CDI `Instance`, so the histogram is silently skipped if no `Meter` bean is available in the container. + ### Context-Aware Async Executor > **Note:** The `AsyncManagedExecutorProducer` is provided by the **Quarkus reference server** ([`reference/common`](https://github.com/a2aproject/a2a-java/blob/main/reference/common/src/main/java/org/a2aproject/sdk/server/common/quarkus/AsyncManagedExecutorProducer.java)), not the OpenTelemetry module. It is documented here because it enables context propagation (including trace context) across async boundaries. @@ -133,6 +204,30 @@ Adds OpenTelemetry spans to A2A client operations: ``` +### Metrics + +The client transport records the following histogram: + +| Instrument | Type | Unit | Description | +|------------|------|------|-------------| +| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of A2A client operations from invocation to completion or error | + +Attributes on `gen_ai.client.operation.duration`: + +| Attribute | Value | +|-----------|-------| +| `gen_ai.agent.a2a.operation.name` | A2A method name (e.g., `message/send`, `tasks/get`) | +| `gen_ai.system` | `a2a` | +| `error.type` | `error` (only on failure) | + +Metrics require passing a `Meter` to the constructor: + +```java +OpenTelemetryClientTransport transport = new OpenTelemetryClientTransport(delegate, tracer, meter); +``` + +If `meter` is `null`, the histogram is silently skipped. + ### Context Propagation For trace context propagation across async client boundaries: diff --git a/examples/helloworld/client/README.md b/examples/helloworld/client/README.md index b186f7ec7..d0673779f 100644 --- a/examples/helloworld/client/README.md +++ b/examples/helloworld/client/README.md @@ -161,11 +161,11 @@ The Java client (`HelloWorldClient.java`) performs the following actions: ## Enable OpenTelemetry (Optional) -The client includes support for distributed tracing with OpenTelemetry. To enable it: +The client includes support for distributed tracing and metrics with OpenTelemetry. To enable it: ### Prerequisites -**IMPORTANT**: The client expects an OpenTelemetry collector to be ready and accepting traces. You have two options: +**IMPORTANT**: The client expects an OpenTelemetry collector to be ready and accepting traces and metrics. You have two options: #### Option 1: Use the Java Server Example (Recommended) diff --git a/examples/helloworld/client/src/main/java/org/a2aproject/sdk/examples/helloworld/client/HelloWorldClient.java b/examples/helloworld/client/src/main/java/org/a2aproject/sdk/examples/helloworld/client/HelloWorldClient.java index 0c24d08c2..46a80ecb2 100644 --- a/examples/helloworld/client/src/main/java/org/a2aproject/sdk/examples/helloworld/client/HelloWorldClient.java +++ b/examples/helloworld/client/src/main/java/org/a2aproject/sdk/examples/helloworld/client/HelloWorldClient.java @@ -1,5 +1,6 @@ package org.a2aproject.sdk.examples.helloworld.client; +import static org.a2aproject.sdk.extras.opentelemetry.client.OpenTelemetryClientTransportWrapper.OTEL_METER_KEY; import static org.a2aproject.sdk.extras.opentelemetry.client.OpenTelemetryClientTransportWrapper.OTEL_TRACER_KEY; import static org.a2aproject.sdk.extras.opentelemetry.client.propagation.OpenTelemetryClientPropagatorTransportWrapper.OTEL_OPEN_TELEMETRY_KEY; @@ -34,8 +35,11 @@ import io.grpc.ManagedChannelBuilder; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter; import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader; import io.opentelemetry.sdk.resources.Resource; import io.opentelemetry.sdk.trace.SdkTracerProvider; import io.opentelemetry.sdk.trace.export.BatchSpanProcessor; @@ -136,20 +140,32 @@ public static void main(String[] args) { } static OpenTelemetrySdk initOpenTelemetry() { + Resource resource = Resource.getDefault().toBuilder() + .put("service.version", "1.0") + .put("service.name", "helloworld-client") + .build(); + SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder() .addSpanProcessor(BatchSpanProcessor.builder( OtlpGrpcSpanExporter.builder() .setEndpoint("http://localhost:5317") .build() ).build()) - .setResource(Resource.getDefault().toBuilder() - .put("service.version", "1.0") - .put("service.name", "helloworld-client") - .build()) + .setResource(resource) + .build(); + + SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder() + .registerMetricReader(PeriodicMetricReader.builder( + OtlpGrpcMetricExporter.builder() + .setEndpoint("http://localhost:5317") + .build() + ).build()) + .setResource(resource) .build(); return OpenTelemetrySdk.builder() .setTracerProvider(sdkTracerProvider) + .setMeterProvider(sdkMeterProvider) .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance())) .build(); } @@ -186,6 +202,7 @@ private static void updateTransportConfig(ClientTransportConfig transportConfig, if (openTelemetrySdk != null) { Map parameters = new HashMap<>(transportConfig.getParameters()); parameters.put(OTEL_TRACER_KEY, openTelemetrySdk.getTracer("helloworld-client")); + parameters.put(OTEL_METER_KEY, openTelemetrySdk.getMeter("helloworld-client")); parameters.put(OTEL_OPEN_TELEMETRY_KEY, openTelemetrySdk); transportConfig.setParameters(parameters); } diff --git a/examples/helloworld/server/src/main/resources/application.properties b/examples/helloworld/server/src/main/resources/application.properties index 4edbf121b..8e406688a 100644 --- a/examples/helloworld/server/src/main/resources/application.properties +++ b/examples/helloworld/server/src/main/resources/application.properties @@ -5,6 +5,7 @@ quarkus.agentcard.protocol=JSONRPC # OpenTelemetry configuration quarkus.otel.sdk.disabled=false +quarkus.otel.metrics.enabled=true quarkus.observability.lgtm.grafana-port=3001 quarkus.observability.lgtm.otel-grpc-port=5317 quarkus.observability.lgtm.otel-http-port=5318 diff --git a/extras/opentelemetry/client/pom.xml b/extras/opentelemetry/client/pom.xml index b29a5324e..447577e2d 100644 --- a/extras/opentelemetry/client/pom.xml +++ b/extras/opentelemetry/client/pom.xml @@ -28,6 +28,12 @@ org.slf4j slf4j-api + + ${project.groupId} + a2a-java-sdk-opentelemetry-common + tests + test + org.junit.jupiter junit-jupiter diff --git a/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java b/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java index 574253978..638fa3ba2 100644 --- a/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java +++ b/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransport.java @@ -1,7 +1,9 @@ package org.a2aproject.sdk.extras.opentelemetry.client; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_REQUEST_SYS_PROPERTY; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CLIENT_OPERATION_DURATION; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CONFIG_ID; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CONTEXT_ID; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_EXTENSIONS; @@ -11,10 +13,28 @@ import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_REQUEST; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_ROLE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_STREAMING_EVENT; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_END_SUFFIX; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_EVENT_SUFFIX; +import io.opentelemetry.api.metrics.DoubleHistogram; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.stream.Collectors; import org.a2aproject.sdk.client.transport.spi.ClientTransport; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.a2aproject.sdk.extras.opentelemetry.StreamingEventAttributes; import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.A2AClientException; import org.a2aproject.sdk.spec.A2AMethods; @@ -33,27 +53,29 @@ import org.a2aproject.sdk.spec.TaskIdParams; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; import org.a2aproject.sdk.spec.TaskQueryParams; -import io.opentelemetry.api.trace.Span; -import io.opentelemetry.api.trace.SpanBuilder; -import io.opentelemetry.api.trace.SpanContext; -import io.opentelemetry.api.trace.SpanKind; -import io.opentelemetry.api.trace.StatusCode; -import io.opentelemetry.api.trace.Tracer; -import io.opentelemetry.context.Scope; -import java.util.HashMap; -import java.util.Map; -import java.util.function.Consumer; -import java.util.stream.Collectors; import org.jspecify.annotations.Nullable; public class OpenTelemetryClientTransport implements ClientTransport { private final Tracer tracer; private final ClientTransport delegate; + @Nullable private final DoubleHistogram operationDurationHistogram; public OpenTelemetryClientTransport(ClientTransport delegate, Tracer tracer) { + this(delegate, tracer, null); + } + + /** + * @param meter optional meter for recording {@code gen_ai.client.operation.duration}; null disables metrics + */ + public OpenTelemetryClientTransport(ClientTransport delegate, Tracer tracer, @Nullable Meter meter) { this.delegate = delegate; this.tracer = tracer; + this.operationDurationHistogram = meter == null ? null + : meter.histogramBuilder(GENAI_CLIENT_OPERATION_DURATION) + .setUnit("s") + .setDescription("Duration of A2A client operations") + .build(); } private boolean extractRequest() { @@ -67,6 +89,7 @@ private boolean extractResponse() { @Override public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext context) throws A2AClientException { ClientCallContext clientContext = createContext(context); + long startNanos = System.nanoTime(); SpanBuilder spanBuilder = tracer.spanBuilder(A2AMethods.SEND_MESSAGE_METHOD).setSpanKind(SpanKind.CLIENT); spanBuilder.setAttribute(GENAI_OPERATION_NAME, A2AMethods.SEND_MESSAGE_METHOD); if (request.message() != null) { @@ -91,6 +114,7 @@ public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallCont spanBuilder.setAttribute(GENAI_REQUEST, request.toString()); } Span span = spanBuilder.startSpan(); + boolean success = false; try (Scope scope = span.makeCurrent()) { EventKind result = delegate.sendMessage(request, clientContext); if (result != null && extractResponse()) { @@ -99,12 +123,14 @@ public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallCont if (result != null) { span.setStatus(StatusCode.OK); } + success = true; return result; } catch (Exception ex) { span.setStatus(StatusCode.ERROR, ex.getMessage()); throw ex; } finally { span.end(); + recordOperationDuration(startNanos, A2AMethods.SEND_MESSAGE_METHOD, success); } } @@ -112,6 +138,7 @@ public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallCont public void sendMessageStreaming(MessageSendParams request, Consumer eventConsumer, Consumer errorConsumer, @Nullable ClientCallContext context) throws A2AClientException { ClientCallContext clientContext = createContext(context); + long startNanos = System.nanoTime(); SpanBuilder spanBuilder = tracer.spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD).setSpanKind(SpanKind.CLIENT); spanBuilder.setAttribute(GENAI_OPERATION_NAME, A2AMethods.SEND_STREAMING_MESSAGE_METHOD); if (request.message() != null) { @@ -135,17 +162,26 @@ public void sendMessageStreaming(MessageSendParams request, Consumer eventConsumer, Consumer errorConsumer, @Nullable ClientCallContext context) throws A2AClientException { ClientCallContext clientContext = createContext(context); + long startNanos = System.nanoTime(); SpanBuilder spanBuilder = tracer.spanBuilder(A2AMethods.SUBSCRIBE_TO_TASK_METHOD).setSpanKind(SpanKind.CLIENT); spanBuilder.setAttribute(GENAI_OPERATION_NAME, A2AMethods.SUBSCRIBE_TO_TASK_METHOD); if (request.id() != null) { @@ -377,17 +442,26 @@ public void subscribeToTask(TaskIdParams request, Consumer e if (extractRequest()) { spanBuilder.setAttribute(GENAI_REQUEST, request.toString()); } + AtomicBoolean terminated = new AtomicBoolean(false); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { + SpanContext spanContext = span.getSpanContext(); delegate.subscribeToTask( request, - new OpenTelemetryEventConsumer(A2AMethods.SUBSCRIBE_TO_TASK_METHOD + "-event", eventConsumer, tracer, span.getSpanContext()), - new OpenTelemetryErrorConsumer(A2AMethods.SUBSCRIBE_TO_TASK_METHOD + "-error", errorConsumer, tracer, span.getSpanContext()), + new OpenTelemetryEventConsumer(A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_EVENT_SUFFIX, + eventConsumer, tracer, spanContext), + new OpenTelemetryTerminalConsumer(errorConsumer, + new TerminalConsumerConfig(A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_END_SUFFIX, + tracer, spanContext, startNanos, A2AMethods.SUBSCRIBE_TO_TASK_METHOD, + operationDurationHistogram, terminated)), clientContext ); span.setStatus(StatusCode.OK); } catch (Exception ex) { span.setStatus(StatusCode.ERROR, ex.getMessage()); + if (!terminated.get()) { + recordOperationDuration(startNanos, A2AMethods.SUBSCRIBE_TO_TASK_METHOD, false); + } throw ex; } finally { span.end(); @@ -397,9 +471,11 @@ public void subscribeToTask(TaskIdParams request, Consumer e @Override public AgentCard getExtendedAgentCard(GetExtendedAgentCardParams params, @Nullable ClientCallContext context) throws A2AClientException { ClientCallContext clientContext = createContext(context); + long startNanos = System.nanoTime(); SpanBuilder spanBuilder = tracer.spanBuilder(A2AMethods.GET_EXTENDED_AGENT_CARD_METHOD).setSpanKind(SpanKind.CLIENT); spanBuilder.setAttribute(GENAI_OPERATION_NAME, A2AMethods.GET_EXTENDED_AGENT_CARD_METHOD); Span span = spanBuilder.startSpan(); + boolean success = false; try (Scope scope = span.makeCurrent()) { AgentCard result = delegate.getExtendedAgentCard(params, clientContext); if (result != null && extractResponse()) { @@ -408,15 +484,21 @@ public AgentCard getExtendedAgentCard(GetExtendedAgentCardParams params, @Nullab if (result != null) { span.setStatus(StatusCode.OK); } + success = true; return result; } catch (Exception ex) { span.setStatus(StatusCode.ERROR, ex.getMessage()); throw ex; } finally { span.end(); + recordOperationDuration(startNanos, A2AMethods.GET_EXTENDED_AGENT_CARD_METHOD, success); } } + private void recordOperationDuration(long startNanos, String operationName, boolean success) { + StreamingEventAttributes.recordDuration(operationDurationHistogram, startNanos, operationName, success); + } + private ClientCallContext createContext(@Nullable ClientCallContext context) { if (context == null) { return new ClientCallContext(Map.of(), new HashMap<>()); @@ -433,60 +515,74 @@ private static class OpenTelemetryEventConsumer implements Consumer delegate; private final Tracer tracer; - private final SpanContext context; + private final SpanContext parentSpanContext; private final String name; - public OpenTelemetryEventConsumer(String name, Consumer delegate, Tracer tracer, SpanContext context) { + OpenTelemetryEventConsumer(String name, Consumer delegate, + Tracer tracer, SpanContext parentSpanContext) { this.delegate = delegate; this.tracer = tracer; - this.context = context; + this.parentSpanContext = parentSpanContext; this.name = name; } @Override public void accept(StreamingEventKind t) { - SpanBuilder spanBuilder = tracer.spanBuilder(name) - .setSpanKind(SpanKind.CLIENT); - spanBuilder.setAttribute("gen_ai.agent.a2a.streaming-event", t.toString()); - spanBuilder.addLink(context); - Span span = spanBuilder.startSpan(); - try { + SpanBuilder builder = tracer.spanBuilder(name) + .setSpanKind(SpanKind.CLIENT) + .addLink(parentSpanContext) + .setAttribute(GENAI_STREAMING_EVENT, t.kind()); + StreamingEventAttributes.apply(builder, t); + Span eventSpan = builder.startSpan(); + try (Scope ignored = eventSpan.makeCurrent()) { delegate.accept(t); - span.setStatus(StatusCode.OK); + eventSpan.setStatus(StatusCode.OK); + } catch (RuntimeException | Error e) { + eventSpan.recordException(e); + eventSpan.setStatus(StatusCode.ERROR); + throw e; } finally { - span.end(); + eventSpan.end(); } } } - private static class OpenTelemetryErrorConsumer implements Consumer { + private record TerminalConsumerConfig(String name, Tracer tracer, SpanContext parentSpanContext, + long startNanos, String operationName, @Nullable DoubleHistogram operationDurationHistogram, + AtomicBoolean terminated) {} + + private static class OpenTelemetryTerminalConsumer implements Consumer { private final Consumer delegate; - private final Tracer tracer; - private final SpanContext context; - private final String name; + private final TerminalConsumerConfig config; - public OpenTelemetryErrorConsumer(String name, Consumer delegate, Tracer tracer, SpanContext context) { + OpenTelemetryTerminalConsumer(Consumer delegate, TerminalConsumerConfig config) { this.delegate = delegate; - this.tracer = tracer; - this.context = context; - this.name = name; + this.config = config; } @Override - public void accept(Throwable t) { - if (t == null) { + public void accept(@Nullable Throwable t) { + if (!config.terminated().compareAndSet(false, true)) { return; } - SpanBuilder spanBuilder = tracer.spanBuilder(name) - .setSpanKind(SpanKind.CLIENT); - spanBuilder.addLink(context); - Span span = spanBuilder.startSpan(); - try { - span.setStatus(StatusCode.ERROR, t.getMessage()); - delegate.accept(t); + Span closingSpan = config.tracer().spanBuilder(config.name()) + .setSpanKind(SpanKind.CLIENT) + .addLink(config.parentSpanContext()) + .startSpan(); + try (Scope ignored = closingSpan.makeCurrent()) { + if (t == null) { + closingSpan.setStatus(StatusCode.OK); + delegate.accept(null); + } else { + closingSpan.setAttribute(ERROR_TYPE, t.getClass().getName()); + closingSpan.setStatus(StatusCode.ERROR, t.getMessage()); + delegate.accept(t); + } } finally { - span.end(); + closingSpan.end(); + StreamingEventAttributes.recordDuration(config.operationDurationHistogram(), + config.startNanos(), config.operationName(), t == null); } } } diff --git a/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportWrapper.java b/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportWrapper.java index 5b329c8f1..fce3a8c78 100644 --- a/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportWrapper.java +++ b/extras/opentelemetry/client/src/main/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportWrapper.java @@ -3,6 +3,7 @@ import org.a2aproject.sdk.client.transport.spi.ClientTransport; import org.a2aproject.sdk.client.transport.spi.ClientTransportConfig; import org.a2aproject.sdk.client.transport.spi.ClientTransportWrapper; +import io.opentelemetry.api.metrics.Meter; import io.opentelemetry.api.trace.Tracer; /** @@ -26,11 +27,18 @@ public class OpenTelemetryClientTransportWrapper implements ClientTransportWrapp */ public static final String OTEL_TRACER_KEY = "org.a2aproject.sdk.extras.opentelemetry.Tracer"; + /** + * Configuration key for the OpenTelemetry Meter instance. + * Value must be of type {@link Meter}. + */ + public static final String OTEL_METER_KEY = "org.a2aproject.sdk.extras.opentelemetry.Meter"; + @Override public ClientTransport wrap(ClientTransport transport, ClientTransportConfig config) { - Object tracerObj = config.getParameters().get(OTEL_TRACER_KEY); - if (tracerObj != null && tracerObj instanceof Tracer tracer) { - return new OpenTelemetryClientTransport(transport, tracer); + if (config.getParameters().get(OTEL_TRACER_KEY) instanceof Tracer tracer) { + Object meterObj = config.getParameters().get(OTEL_METER_KEY); + Meter meter = (meterObj instanceof Meter m) ? m : null; + return new OpenTelemetryClientTransport(transport, tracer, meter); } // No tracer configured, return unwrapped transport return transport; diff --git a/extras/opentelemetry/client/src/test/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportTest.java b/extras/opentelemetry/client/src/test/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportTest.java index fd1e7d992..65927735d 100644 --- a/extras/opentelemetry/client/src/test/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportTest.java +++ b/extras/opentelemetry/client/src/test/java/org/a2aproject/sdk/extras/opentelemetry/client/OpenTelemetryClientTransportTest.java @@ -1,15 +1,52 @@ package org.a2aproject.sdk.extras.opentelemetry.client; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_REQUEST_SYS_PROPERTY; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_REQUEST; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE; - +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_STREAMING_EVENT; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_END_SUFFIX; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_EVENT_SUFFIX; +import static org.a2aproject.sdk.extras.opentelemetry.OtelSpanMocks.mockClosingSpanBuilder; +import static org.a2aproject.sdk.extras.opentelemetry.OtelSpanMocks.mockEventSpanBuilder; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyDouble; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.opentelemetry.api.metrics.DoubleHistogram; +import io.opentelemetry.api.metrics.DoubleHistogramBuilder; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import java.util.List; +import java.util.function.Consumer; import org.a2aproject.sdk.client.transport.spi.ClientTransport; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.A2AClientException; +import org.a2aproject.sdk.spec.A2AMethods; import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.CancelTaskParams; import org.a2aproject.sdk.spec.DeleteTaskPushNotificationConfigParams; import org.a2aproject.sdk.spec.EventKind; import org.a2aproject.sdk.spec.GetExtendedAgentCardParams; @@ -20,19 +57,13 @@ import org.a2aproject.sdk.spec.Message; import org.a2aproject.sdk.spec.MessageSendParams; import org.a2aproject.sdk.spec.StreamingEventKind; -import org.a2aproject.sdk.spec.TextPart; import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskIdParams; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; import org.a2aproject.sdk.spec.TaskQueryParams; -import io.opentelemetry.api.trace.Span; -import io.opentelemetry.api.trace.SpanBuilder; -import io.opentelemetry.api.trace.SpanContext; -import io.opentelemetry.api.trace.SpanKind; -import io.opentelemetry.api.trace.StatusCode; -import io.opentelemetry.api.trace.Tracer; -import io.opentelemetry.context.Scope; +import org.a2aproject.sdk.spec.TextPart; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.ArgumentCaptor; @@ -41,16 +72,6 @@ import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; -import java.util.List; -import java.util.function.Consumer; - -import static org.junit.jupiter.api.Assertions.*; -import static org.mockito.ArgumentMatchers.*; -import static org.mockito.Mockito.*; - -import org.a2aproject.sdk.spec.A2AMethods; -import org.a2aproject.sdk.spec.CancelTaskParams; - @ExtendWith(MockitoExtension.class) @MockitoSettings(strictness = Strictness.LENIENT) class OpenTelemetryClientTransportTest { @@ -95,6 +116,7 @@ void setUp() { transport = new OpenTelemetryClientTransport(delegate, tracer); } + @Test void testSendMessage_Success() throws A2AClientException { MessageSendParams request = mock(MessageSendParams.class); @@ -147,6 +169,7 @@ void testSendMessage_ThrowsException() throws A2AClientException { } @Test + @SuppressWarnings("unchecked") void testSendMessageStreaming() throws A2AClientException { MessageSendParams request = mock(MessageSendParams.class); when(request.toString()).thenReturn("request-string"); @@ -298,6 +321,7 @@ void testDeleteTaskPushNotificationConfigurations_Success() throws A2AClientExce } @Test + @SuppressWarnings("unchecked") void testSubscribeToTask() throws A2AClientException { TaskIdParams request = mock(TaskIdParams.class); when(request.toString()).thenReturn("request-string"); @@ -360,17 +384,17 @@ void testClose() { } @Test + @SuppressWarnings("unchecked") void testEventConsumer_ThroughSendMessageStreaming() throws A2AClientException { MessageSendParams request = mock(MessageSendParams.class); when(request.toString()).thenReturn("request-string"); - SpanBuilder eventSpanBuilder = mock(SpanBuilder.class); Span eventSpan = mock(Span.class); - when(tracer.spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + "-event")).thenReturn(eventSpanBuilder); - when(eventSpanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(eventSpanBuilder); - when(eventSpanBuilder.setAttribute(anyString(), anyString())).thenReturn(eventSpanBuilder); - when(eventSpanBuilder.addLink(any(SpanContext.class))).thenReturn(eventSpanBuilder); - when(eventSpanBuilder.startSpan()).thenReturn(eventSpan); + SpanBuilder eventSpanBuilder = mockEventSpanBuilder(tracer, + A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_EVENT_SUFFIX, eventSpan); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); ArgumentCaptor> eventConsumerCaptor = ArgumentCaptor.forClass(Consumer.class); Consumer originalConsumer = mock(Consumer.class); @@ -388,23 +412,21 @@ void testEventConsumer_ThroughSendMessageStreaming() throws A2AClientException { eventConsumerCaptor.getValue().accept(event); - verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + "-event"); - verify(eventSpan).setStatus(StatusCode.OK); + verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_EVENT_SUFFIX); + verify(eventSpanBuilder).addLink(spanContext); + verify(eventSpanBuilder).setAttribute(GENAI_STREAMING_EVENT, event.kind()); verify(eventSpan).end(); verify(originalConsumer).accept(event); } @Test + @SuppressWarnings("unchecked") void testErrorConsumer_ThroughSendMessageStreaming() throws A2AClientException { MessageSendParams request = mock(MessageSendParams.class); when(request.toString()).thenReturn("request-string"); - SpanBuilder errorSpanBuilder = mock(SpanBuilder.class); - Span errorSpan = mock(Span.class); - when(tracer.spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + "-error")).thenReturn(errorSpanBuilder); - when(errorSpanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(errorSpanBuilder); - when(errorSpanBuilder.addLink(any(SpanContext.class))).thenReturn(errorSpanBuilder); - when(errorSpanBuilder.startSpan()).thenReturn(errorSpan); + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); ArgumentCaptor> errorConsumerCaptor = ArgumentCaptor.forClass(Consumer.class); Consumer originalConsumer = mock(Consumer.class); @@ -414,30 +436,35 @@ void testErrorConsumer_ThroughSendMessageStreaming() throws A2AClientException { verify(delegate).sendMessageStreaming(eq(request), any(Consumer.class), errorConsumerCaptor.capture(), any(ClientCallContext.class)); Throwable error = new RuntimeException("Test error"); - errorConsumerCaptor.getValue().accept(error); - verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + "-error"); - verify(errorSpan).setStatus(StatusCode.ERROR, "Test error"); - verify(errorSpan).end(); + verify(closingSpan).setAttribute(ERROR_TYPE, RuntimeException.class.getName()); + verify(closingSpan).setStatus(StatusCode.ERROR, "Test error"); + verify(closingSpan, atLeastOnce()).end(); verify(originalConsumer).accept(error); } @Test - void testErrorConsumer_WithNullThrowable() throws A2AClientException { + @SuppressWarnings("unchecked") + void testTerminalConsumer_WithNullThrowable_CreatesOkClosingSpan() throws A2AClientException { MessageSendParams request = mock(MessageSendParams.class); when(request.toString()).thenReturn("request-string"); - ArgumentCaptor> errorConsumerCaptor = ArgumentCaptor.forClass(Consumer.class); + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + ArgumentCaptor> terminalConsumerCaptor = ArgumentCaptor.forClass(Consumer.class); Consumer originalConsumer = mock(Consumer.class); transport.sendMessageStreaming(request, mock(Consumer.class), originalConsumer, context); - verify(delegate).sendMessageStreaming(eq(request), any(Consumer.class), errorConsumerCaptor.capture(), any(ClientCallContext.class)); + verify(delegate).sendMessageStreaming(eq(request), any(Consumer.class), terminalConsumerCaptor.capture(), any(ClientCallContext.class)); - errorConsumerCaptor.getValue().accept(null); + terminalConsumerCaptor.getValue().accept(null); - verify(originalConsumer, never()).accept(any()); + verify(closingSpan).setStatus(StatusCode.OK); + verify(closingSpan).end(); + verify(originalConsumer).accept(null); } @Test @@ -473,6 +500,81 @@ void testResubscribe_ThrowsException() throws A2AClientException { verify(span).end(); } + @Nested + class MetricsTests { + + @Mock + private Meter meter; + + @Mock + private DoubleHistogramBuilder histogramBuilder; + + @Mock + private DoubleHistogram histogram; + + @BeforeEach + void setUpMeter() { + lenient().when(meter.histogramBuilder(anyString())).thenReturn(histogramBuilder); + lenient().when(histogramBuilder.setUnit(anyString())).thenReturn(histogramBuilder); + lenient().when(histogramBuilder.setDescription(anyString())).thenReturn(histogramBuilder); + lenient().when(histogramBuilder.build()).thenReturn(histogram); + } + + @Test + void sendMessage_recordsOperationDuration() throws A2AClientException { + OpenTelemetryClientTransport transportWithMeter = new OpenTelemetryClientTransport(delegate, tracer, meter); + MessageSendParams request = mock(MessageSendParams.class); + EventKind expectedResult = mock(EventKind.class); + when(request.toString()).thenReturn("request-string"); + when(expectedResult.toString()).thenReturn("response-string"); + when(delegate.sendMessage(eq(request), any(ClientCallContext.class))).thenReturn(expectedResult); + + transportWithMeter.sendMessage(request, context); + + verify(histogram).record(anyDouble(), any(io.opentelemetry.api.common.Attributes.class)); + } + + @Test + void sendMessage_noHistogramWhenNoMeter() throws A2AClientException { + // transport without meter — should not throw, histogram never invoked + MessageSendParams request = mock(MessageSendParams.class); + EventKind expectedResult = mock(EventKind.class); + when(request.toString()).thenReturn("request-string"); + when(expectedResult.toString()).thenReturn("response-string"); + when(delegate.sendMessage(eq(request), any(ClientCallContext.class))).thenReturn(expectedResult); + + transport.sendMessage(request, context); + + verify(histogram, never()).record(anyDouble(), any(io.opentelemetry.api.common.Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void sendMessageStreaming_recordsFailedOperationDuration() throws A2AClientException { + OpenTelemetryClientTransport transportWithMeter = new OpenTelemetryClientTransport(delegate, tracer, meter); + MessageSendParams request = mock(MessageSendParams.class); + A2AClientException expectedException = new A2AClientException("Streaming failed"); + doThrow(expectedException).when(delegate).sendMessageStreaming(any(MessageSendParams.class), any(Consumer.class), + any(Consumer.class), any(ClientCallContext.class)); + + assertThrows(A2AClientException.class, + () -> transportWithMeter.sendMessageStreaming(request, mock(Consumer.class), mock(Consumer.class), context)); + + verify(histogram).record(anyDouble(), any(io.opentelemetry.api.common.Attributes.class)); + } + } + + @Test + @SuppressWarnings("unchecked") + void closingSpan_notCreatedDuringStreamingInitiation() throws A2AClientException { + MessageSendParams request = mock(MessageSendParams.class); + when(request.toString()).thenReturn("request-string"); + + transport.sendMessageStreaming(request, mock(Consumer.class), mock(Consumer.class), context); + + verify(tracer, never()).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX); + } + @Test void testSendMessageStreaming_ThrowsException() throws A2AClientException { MessageSendParams request = mock(MessageSendParams.class); diff --git a/extras/opentelemetry/common/pom.xml b/extras/opentelemetry/common/pom.xml index 6d5c5f8ef..e1b4e7672 100644 --- a/extras/opentelemetry/common/pom.xml +++ b/extras/opentelemetry/common/pom.xml @@ -15,4 +15,38 @@ A2A Java SDK :: Extras :: Opentelemetry :: Common Common OpenTelemetry utilities for A2A Java SDK + + + ${project.groupId} + a2a-java-sdk-spec + + + org.junit.jupiter + junit-jupiter-api + test + + + org.mockito + mockito-core + test + + + + + + + org.apache.maven.plugins + maven-jar-plugin + + + test-jar + + test-jar + + + + + + + diff --git a/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java b/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java index 4bd74e99b..8ea1fa7c6 100644 --- a/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java +++ b/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java @@ -1,11 +1,18 @@ package org.a2aproject.sdk.extras.opentelemetry; public interface A2AObservabilityNames { - + String EXTRACT_REQUEST_SYS_PROPERTY = "org.a2aproject.sdk.server.extract.request"; String EXTRACT_RESPONSE_SYS_PROPERTY = "org.a2aproject.sdk.server.extract.response"; String ERROR_TYPE = "error.type"; + /** + * Fixed error marker recorded in histogram attributes on failure. + * + *

Intentionally low-cardinality: metrics use this constant to avoid unbounded label sets, + * while span attributes carry the actual exception class name for debugging. + */ + String ERROR_TYPE_VALUE = "error"; String GENAI_PREFIX = "gen_ai.agent.a2a"; String GENAI_CONFIG_ID = GENAI_PREFIX + ".config_id"; @@ -14,10 +21,25 @@ public interface A2AObservabilityNames { String GENAI_MESSAGE_ID = GENAI_PREFIX + ".message_id"; String GENAI_OPERATION_NAME = GENAI_PREFIX + ".operation.name"; //gen_ai.agent.operation.name ? String GENAI_PARTS_NUMBER = GENAI_PREFIX + ".parts.number"; - String GENAI_PROTOCOL = GENAI_PREFIX + ".protocol"; + String GENAI_TRANSPORT = GENAI_PREFIX + ".transport"; String GENAI_STATUS = GENAI_PREFIX + ".status"; + String GENAI_PROTOCOL_VERSION = GENAI_PREFIX + ".protocol.version"; String GENAI_REQUEST = GENAI_PREFIX + ".request"; //gen_ai.input.messages ? String GENAI_RESPONSE = GENAI_PREFIX + ".response"; // gen_ai.output.messages ? String GENAI_ROLE = GENAI_PREFIX + ".role"; String GENAI_TASK_ID = GENAI_PREFIX + ".task_id"; + + // Streaming event attributes + String GENAI_STREAMING_EVENT = GENAI_PREFIX + ".streaming-event"; + String GENAI_TASK_STATUS_STATE = GENAI_PREFIX + ".task.status.state"; + + // Streaming span name suffixes + String STREAMING_EVENT_SUFFIX = "-event"; + String STREAMING_END_SUFFIX = "-end"; + + // Metrics + String GENAI_CLIENT_OPERATION_DURATION = "gen_ai.client.operation.duration"; + String GENAI_STREAMING_DURATION = GENAI_PREFIX + ".streaming.duration"; + String GENAI_SYSTEM = "gen_ai.system"; + String GENAI_SYSTEM_VALUE = "a2a"; } diff --git a/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributes.java b/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributes.java new file mode 100644 index 000000000..2d3252b2f --- /dev/null +++ b/extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributes.java @@ -0,0 +1,87 @@ +package org.a2aproject.sdk.extras.opentelemetry; + +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE_VALUE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_SYSTEM; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_SYSTEM_VALUE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_STATUS_STATE; + +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.common.AttributesBuilder; +import io.opentelemetry.api.metrics.DoubleHistogram; +import io.opentelemetry.api.trace.SpanBuilder; +import org.a2aproject.sdk.spec.Message; +import org.a2aproject.sdk.spec.StreamingEventKind; +import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; +import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; +import org.jspecify.annotations.Nullable; + +public final class StreamingEventAttributes { + + private StreamingEventAttributes() { + } + + /** + * Applies task-specific attributes derived from the streaming event to the span builder. + * Populates task ID, task status state, and other event-kind-specific fields. + * + * @param builder the span builder to enrich with event attributes + * @param item the streaming event to extract attributes from + */ + public static void apply(SpanBuilder builder, StreamingEventKind item) { + switch (item.kind()) { + case TaskStatusUpdateEvent.STREAMING_EVENT_ID -> { + TaskStatusUpdateEvent event = (TaskStatusUpdateEvent) item; + builder.setAttribute(GENAI_TASK_STATUS_STATE, event.status().state().name()); + builder.setAttribute(GENAI_TASK_ID, event.taskId()); + } + case Task.STREAMING_EVENT_ID -> { + Task task = (Task) item; + builder.setAttribute(GENAI_TASK_STATUS_STATE, task.status().state().name()); + builder.setAttribute(GENAI_TASK_ID, task.id()); + } + case TaskArtifactUpdateEvent.STREAMING_EVENT_ID -> { + TaskArtifactUpdateEvent event = (TaskArtifactUpdateEvent) item; + builder.setAttribute(GENAI_TASK_ID, event.taskId()); + } + case Message.STREAMING_EVENT_ID -> { + Message message = (Message) item; + if (message.taskId() != null) { + builder.setAttribute(GENAI_TASK_ID, message.taskId()); + } + } + default -> { + } + } + } + + /** + * Records elapsed time to a histogram using the standard A2A operation attributes. + * + *

The {@code error.type} attribute is set to the fixed string {@value A2AObservabilityNames#ERROR_TYPE_VALUE} + * on failures rather than the actual exception class name. This is intentional: metrics use a low-cardinality + * error marker to avoid unbounded label sets, while spans carry the full class name for debugging. + * + * @param histogram the histogram to record to; no-op if {@code null} + * @param startNanos the start time from {@link System#nanoTime()} + * @param operationName the A2A operation name recorded as {@code gen_ai.agent.a2a.operation.name} + * @param success {@code false} adds the {@code error.type} attribute + */ + public static void recordDuration(@Nullable DoubleHistogram histogram, long startNanos, + String operationName, boolean success) { + if (histogram == null) { + return; + } + double seconds = (System.nanoTime() - startNanos) / 1_000_000_000.0; + AttributesBuilder builder = Attributes.builder() + .put(GENAI_OPERATION_NAME, operationName) + .put(GENAI_SYSTEM, GENAI_SYSTEM_VALUE); + if (!success) { + builder.put(ERROR_TYPE, ERROR_TYPE_VALUE); + } + histogram.record(seconds, builder.build()); + } +} diff --git a/extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/OtelSpanMocks.java b/extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/OtelSpanMocks.java new file mode 100644 index 000000000..e1266db24 --- /dev/null +++ b/extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/OtelSpanMocks.java @@ -0,0 +1,54 @@ +package org.a2aproject.sdk.extras.opentelemetry; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; + +/** Shared Mockito helpers for OTel span builder stubs used in client and server tests. */ +public final class OtelSpanMocks { + + private OtelSpanMocks() { + } + + /** + * Stubs {@code tracer.spanBuilder(spanName)} to return a builder that produces {@code closingSpan}, + * with status and scope stubs that cover both success and error paths. + */ + public static SpanBuilder mockClosingSpanBuilder(Tracer tracer, String spanName, Span closingSpan) { + SpanBuilder closingSpanBuilder = mock(SpanBuilder.class); + when(tracer.spanBuilder(spanName)).thenReturn(closingSpanBuilder); + when(closingSpanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(closingSpanBuilder); + when(closingSpanBuilder.addLink(any(SpanContext.class))).thenReturn(closingSpanBuilder); + when(closingSpanBuilder.startSpan()).thenReturn(closingSpan); + // lenient: success/error path coverage varies per test + lenient().when(closingSpan.setStatus(any(StatusCode.class))).thenReturn(closingSpan); + lenient().when(closingSpan.setStatus(any(StatusCode.class), anyString())).thenReturn(closingSpan); + when(closingSpan.makeCurrent()).thenReturn(mock(Scope.class)); + return closingSpanBuilder; + } + + /** + * Stubs {@code tracer.spanBuilder(spanName)} to return a builder that produces {@code eventSpan}, + * with attribute and scope stubs for per-event spans. + */ + public static SpanBuilder mockEventSpanBuilder(Tracer tracer, String spanName, Span eventSpan) { + SpanBuilder eventSpanBuilder = mock(SpanBuilder.class); + when(tracer.spanBuilder(spanName)).thenReturn(eventSpanBuilder); + when(eventSpanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(eventSpanBuilder); + when(eventSpanBuilder.addLink(any(SpanContext.class))).thenReturn(eventSpanBuilder); + when(eventSpanBuilder.setAttribute(anyString(), anyString())).thenReturn(eventSpanBuilder); + when(eventSpanBuilder.startSpan()).thenReturn(eventSpan); + when(eventSpan.makeCurrent()).thenReturn(mock(Scope.class)); + return eventSpanBuilder; + } +} diff --git a/extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributesTest.java b/extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributesTest.java new file mode 100644 index 000000000..53f4c541a --- /dev/null +++ b/extras/opentelemetry/common/src/test/java/org/a2aproject/sdk/extras/opentelemetry/StreamingEventAttributesTest.java @@ -0,0 +1,60 @@ +package org.a2aproject.sdk.extras.opentelemetry; + +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyDouble; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.DoubleHistogram; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class StreamingEventAttributesTest { + + @Test + void recordDuration_noOpWhenHistogramIsNull() { + StreamingEventAttributes.recordDuration(null, System.nanoTime(), "test-op", true); + // must not throw + } + + @Test + void recordDuration_recordsPositiveValueForSubSecondOperation() { + DoubleHistogram histogram = mock(DoubleHistogram.class); + long startNanos = System.nanoTime() - 500_000L; // simulate 500 µs elapsed + + StreamingEventAttributes.recordDuration(histogram, startNanos, "test-op", true); + + ArgumentCaptor valueCaptor = ArgumentCaptor.forClass(Double.class); + verify(histogram).record(valueCaptor.capture(), any(Attributes.class)); + double recorded = valueCaptor.getValue(); + assertTrue(recorded > 0.0, "recorded duration must be positive, got: " + recorded); + assertTrue(recorded < 1.0, "recorded duration should be sub-second, got: " + recorded); + } + + @Test + void recordDuration_setsErrorTypeAttributeOnFailure() { + DoubleHistogram histogram = mock(DoubleHistogram.class); + + StreamingEventAttributes.recordDuration(histogram, System.nanoTime() - 500_000L, "test-op", false); + + ArgumentCaptor attrsCaptor = ArgumentCaptor.forClass(Attributes.class); + verify(histogram).record(anyDouble(), attrsCaptor.capture()); + assertNotNull(attrsCaptor.getValue().get(AttributeKey.stringKey(A2AObservabilityNames.ERROR_TYPE))); + } + + @Test + void recordDuration_omitsErrorTypeAttributeOnSuccess() { + DoubleHistogram histogram = mock(DoubleHistogram.class); + + StreamingEventAttributes.recordDuration(histogram, System.nanoTime() - 500_000L, "test-op", true); + + ArgumentCaptor attrsCaptor = ArgumentCaptor.forClass(Attributes.class); + verify(histogram).record(anyDouble(), attrsCaptor.capture()); + assertNull(attrsCaptor.getValue().get(AttributeKey.stringKey(A2AObservabilityNames.ERROR_TYPE))); + } +} diff --git a/extras/opentelemetry/integration-tests/pom.xml b/extras/opentelemetry/integration-tests/pom.xml index 676136c29..beccbc3ae 100644 --- a/extras/opentelemetry/integration-tests/pom.xml +++ b/extras/opentelemetry/integration-tests/pom.xml @@ -83,6 +83,16 @@ awaitility test + + org.mockito + mockito-core + test + + + org.mockito + mockito-junit-jupiter + test + diff --git a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java index 4c37b5c5d..13ca3a44f 100644 --- a/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java +++ b/extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java @@ -13,9 +13,15 @@ import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; +import io.opentelemetry.api.OpenTelemetry; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.metrics.data.HistogramData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.data.MetricDataType; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricExporter; import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; import io.opentelemetry.sdk.trace.data.SpanData; import io.vertx.ext.web.Router; @@ -27,7 +33,6 @@ import jakarta.inject.Inject; import jakarta.inject.Singleton; import java.util.List; -import java.util.stream.Collectors; /** * Test routes for OpenTelemetry integration testing. @@ -44,6 +49,10 @@ public class A2ATestRoutes { TestUtilsBean testUtilsBean; @Inject InMemorySpanExporter inMemorySpanExporter; + @Inject + InMemoryMetricExporter inMemoryMetricExporter; + @Inject + OpenTelemetry openTelemetry; @Inject Tracer tracer; @@ -84,9 +93,17 @@ void setupRoutes(@Observes Router router) { .produces(APPLICATION_JSON) .handler(ctx -> exportSpans(ctx)); + router.get("/export-metrics") + .produces(APPLICATION_JSON) + .handler(ctx -> exportMetrics(ctx)); + router.get("/reset") .produces(TEXT_PLAIN) .handler(ctx -> reset(ctx)); + + router.get("/reset-metrics") + .produces(TEXT_PLAIN) + .handler(ctx -> resetMetrics(ctx)); } public void saveTask(String body, RoutingContext rc) { @@ -196,7 +213,7 @@ public void exportSpans(RoutingContext rc) { List spans = inMemorySpanExporter.getFinishedSpanItems() .stream() .filter(sd -> !sd.getName().contains("export") && !sd.getName().contains("reset")) - .collect(Collectors.toList()); + .toList(); String json = gson.toJson(serialize(spans)); rc.response() .setStatusCode(200) @@ -228,12 +245,78 @@ private JsonElement serialize(List spanDatas) { spanData.getResource().getAttributes().forEach((k, v) -> { jsonObject.addProperty("resource_" + k.getKey(), v.toString()); }); + + serializeSpanEvents(spanData, jsonObject); + serializeSpanLinks(spanData, jsonObject); + spans.add(jsonObject); } return spans; } + private void serializeSpanEvents(SpanData spanData, JsonObject jsonObject) { + jsonObject.addProperty("events_count", spanData.getEvents().size()); + spanData.getEvents().forEach(event -> { + JsonObject eventObject = new JsonObject(); + eventObject.addProperty("name", event.getName()); + event.getAttributes().forEach((k, v) -> eventObject.addProperty("attr_" + k.getKey(), v.toString())); + jsonObject.add("event_" + event.getName(), eventObject); + }); + } + + private void serializeSpanLinks(SpanData spanData, JsonObject jsonObject) { + jsonObject.addProperty("links_count", spanData.getLinks().size()); + if (!spanData.getLinks().isEmpty()) { + jsonObject.addProperty("link_0_spanId", spanData.getLinks().get(0).getSpanContext().getSpanId()); + jsonObject.addProperty("link_0_traceId", spanData.getLinks().get(0).getSpanContext().getTraceId()); + } + } + + public void exportMetrics(RoutingContext rc) { + if (openTelemetry instanceof OpenTelemetrySdk sdk) { + sdk.getSdkMeterProvider().forceFlush(); + } + List metrics = inMemoryMetricExporter.getFinishedMetricItems().stream() + .filter(m -> m.getName().startsWith("gen_ai.")) + .toList(); + rc.response() + .setStatusCode(200) + .putHeader(CONTENT_TYPE, APPLICATION_JSON) + .end(gson.toJson(serializeMetrics(metrics))); + } + + JsonElement serializeMetrics(List metricDataList) { + JsonArray result = new JsonArray(metricDataList.size()); + for (MetricData m : metricDataList) { + JsonObject obj = new JsonObject(); + obj.addProperty("name", m.getName()); + obj.addProperty("description", m.getDescription()); + obj.addProperty("unit", m.getUnit()); + obj.addProperty("type", m.getType().name()); + if (m.getType() == MetricDataType.HISTOGRAM) { + HistogramData histogram = m.getHistogramData(); + obj.addProperty("data_count", histogram.getPoints().size()); + JsonArray points = new JsonArray(histogram.getPoints().size()); + histogram.getPoints().forEach(point -> { + JsonObject pointObj = new JsonObject(); + pointObj.addProperty("count", point.getCount()); + pointObj.addProperty("sum", point.getSum()); + point.getAttributes().forEach((k, v) -> pointObj.addProperty("attr_" + k.getKey(), v.toString())); + points.add(pointObj); + }); + obj.add("points", points); + } + result.add(obj); + } + return result; + } + + public void resetMetrics(RoutingContext rc) { + inMemoryMetricExporter.reset(); + rc.response().setStatusCode(200).end(); + } + public void reset(RoutingContext rc) { inMemorySpanExporter.reset(); rc.response().setStatusCode(200).end(); @@ -256,4 +339,14 @@ InMemorySpanExporter inMemorySpanExporter() { return InMemorySpanExporter.create(); } } + + @ApplicationScoped + static class InMemoryMetricExporterProducer { + + @Produces + @Singleton + InMemoryMetricExporter inMemoryMetricExporter() { + return InMemoryMetricExporter.create(); + } + } } diff --git a/extras/opentelemetry/integration-tests/src/main/resources/application.properties b/extras/opentelemetry/integration-tests/src/main/resources/application.properties index 052be4a60..aae4890aa 100644 --- a/extras/opentelemetry/integration-tests/src/main/resources/application.properties +++ b/extras/opentelemetry/integration-tests/src/main/resources/application.properties @@ -8,7 +8,8 @@ quarkus.http.test-port=8081 # OpenTelemetry configuration quarkus.otel.sdk.disabled=false quarkus.otel.traces.enabled=true -quarkus.otel.metrics.enabled=false +quarkus.otel.metrics.enabled=true +quarkus.otel.metrics.exporter=cdi quarkus.otel.logs.enabled=false quarkus.otel.instrument.vertx-http=false diff --git a/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutesTest.java b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutesTest.java new file mode 100644 index 000000000..dbc30cd3e --- /dev/null +++ b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutesTest.java @@ -0,0 +1,98 @@ +package org.a2aproject.sdk.extras.opentelemetry.it; + +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.sdk.metrics.data.HistogramData; +import io.opentelemetry.sdk.metrics.data.HistogramPointData; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.data.MetricDataType; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.util.List; + +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CLIENT_OPERATION_DURATION; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_STREAMING_DURATION; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class A2ATestRoutesTest { + + @Mock + private MetricData metricData; + @Mock + private HistogramData histogramData; + @Mock + private HistogramPointData point1; + @Mock + private HistogramPointData point2; + + private final A2ATestRoutes routes = new A2ATestRoutes(); + + @Test + void serializeMetrics_singleHistogramPoint_returnsPointsArray() { + when(metricData.getName()).thenReturn(GENAI_STREAMING_DURATION); + when(metricData.getDescription()).thenReturn("desc"); + when(metricData.getUnit()).thenReturn("s"); + when(metricData.getType()).thenReturn(MetricDataType.HISTOGRAM); + when(metricData.getHistogramData()).thenReturn(histogramData); + when(histogramData.getPoints()).thenReturn(List.of(point1)); + when(point1.getCount()).thenReturn(1L); + when(point1.getSum()).thenReturn(0.5); + when(point1.getAttributes()).thenReturn(Attributes.empty()); + + JsonElement result = routes.serializeMetrics(List.of(metricData)); + + JsonObject metric = result.getAsJsonArray().get(0).getAsJsonObject(); + assertTrue(metric.has("points"), "Should have 'points' JsonArray"); + assertFalse(metric.has("point"), "Should not have single-overwriting 'point' key"); + assertEquals(1, metric.getAsJsonArray("points").size()); + } + + @Test + void serializeMetrics_multipleHistogramPoints_allPointsSerialized() { + when(metricData.getName()).thenReturn(GENAI_STREAMING_DURATION); + when(metricData.getDescription()).thenReturn("desc"); + when(metricData.getUnit()).thenReturn("s"); + when(metricData.getType()).thenReturn(MetricDataType.HISTOGRAM); + when(metricData.getHistogramData()).thenReturn(histogramData); + when(histogramData.getPoints()).thenReturn(List.of(point1, point2)); + when(point1.getCount()).thenReturn(1L); + when(point1.getSum()).thenReturn(0.5); + when(point1.getAttributes()).thenReturn(Attributes.empty()); + when(point2.getCount()).thenReturn(3L); + when(point2.getSum()).thenReturn(2.0); + when(point2.getAttributes()).thenReturn(Attributes.empty()); + + JsonElement result = routes.serializeMetrics(List.of(metricData)); + + JsonObject metric = result.getAsJsonArray().get(0).getAsJsonObject(); + JsonArray points = metric.getAsJsonArray("points"); + assertEquals(2, points.size(), "Both histogram data points must be serialized"); + assertEquals(1L, points.get(0).getAsJsonObject().get("count").getAsLong()); + assertEquals(0.5, points.get(0).getAsJsonObject().get("sum").getAsDouble()); + assertEquals(3L, points.get(1).getAsJsonObject().get("count").getAsLong()); + assertEquals(2.0, points.get(1).getAsJsonObject().get("sum").getAsDouble()); + } + + @Test + void serializeMetrics_nonHistogramMetric_hasNoPointsKey() { + when(metricData.getName()).thenReturn(GENAI_CLIENT_OPERATION_DURATION); + when(metricData.getDescription()).thenReturn("desc"); + when(metricData.getUnit()).thenReturn("s"); + when(metricData.getType()).thenReturn(MetricDataType.LONG_SUM); + + JsonElement result = routes.serializeMetrics(List.of(metricData)); + + JsonObject metric = result.getAsJsonArray().get(0).getAsJsonObject(); + assertFalse(metric.has("points"), "Non-histogram metric should not have a 'points' key"); + assertFalse(metric.has("point"), "Non-histogram metric should not have a 'point' key"); + } +} diff --git a/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/BaseTest.java b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/BaseTest.java index b3c154f44..c15b66ff8 100644 --- a/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/BaseTest.java +++ b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/BaseTest.java @@ -14,6 +14,11 @@ protected List> getSpans() { }); } + protected List> getMetrics() { + return get("/export-metrics").body().as(new TypeRef<>() { + }); + } + protected void buildGlobalTelemetryInstance() { // Do nothing in JVM mode } diff --git a/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java index cbe09435b..93b9d91b7 100644 --- a/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java +++ b/extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java @@ -3,6 +3,9 @@ import static io.restassured.RestAssured.given; import static java.net.HttpURLConnection.HTTP_OK; import static java.util.concurrent.TimeUnit.SECONDS; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_STREAMING_DURATION; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_END_SUFFIX; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_EVENT_SUFFIX; import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.*; @@ -235,6 +238,93 @@ void testSpanAttributes() throws Exception { } } + @Test + void testSendMessageStreamCreatesClosingSpan() throws Exception { + reset(); + given().get("/reset-metrics").then().statusCode(HTTP_OK); + + Client streamingClient = Client.builder(A2A.getAgentCard("http://localhost:" + serverPort)) + .clientConfig(new ClientConfig.Builder().setStreaming(true).build()) + .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfigBuilder()) + .build(); + + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("stream test"))) + .messageId("stream-msg-1") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + + streamingClient.sendMessage(params, List.of(), null, null); + + await().atMost(10, SECONDS).until(() -> { + List> spans = getSpans(); + return spans.stream().anyMatch(s -> (A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX).equals(s.get("name"))); + }); + + List> spans = getSpans(); + + Map initialSpan = spans.stream() + .filter(s -> A2AMethods.SEND_STREAMING_MESSAGE_METHOD.equals(s.get("name"))) + .findFirst() + .orElseThrow(() -> new AssertionError("No initial SendStreamingMessage span found")); + + Map closingSpan = spans.stream() + .filter(s -> (A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX).equals(s.get("name"))) + .findFirst() + .orElseThrow(() -> new AssertionError("No SendStreamingMessage-stream closing span found")); + + assertEquals(Boolean.TRUE, initialSpan.get("ended"), "Initial span should be ended"); + assertEquals("SERVER", initialSpan.get("kind"), "Initial span should be SERVER"); + + assertEquals(Boolean.TRUE, closingSpan.get("ended"), "Closing span should be ended"); + assertEquals("SERVER", closingSpan.get("kind"), "Closing span should be SERVER"); + + assertTrue( + spans.stream().anyMatch(s -> (A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_EVENT_SUFFIX).equals(s.get("name"))), + "Should have at least one streaming event span"); + assertEquals(1, ((Number) closingSpan.get("links_count")).intValue(), + "Closing span should have exactly one link"); + assertEquals(initialSpan.get("spanId"), closingSpan.get("link_0_spanId"), + "Closing span should link to the initial span"); + } + + @Test + void testSendMessageStreamRecordsStreamingDuration() throws Exception { + reset(); + given().get("/reset-metrics").then().statusCode(HTTP_OK); + + Client streamingClient = Client.builder(A2A.getAgentCard("http://localhost:" + serverPort)) + .clientConfig(new ClientConfig.Builder().setStreaming(true).build()) + .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfigBuilder()) + .build(); + + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("metric stream test"))) + .messageId("metric-stream-msg-1") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + + streamingClient.sendMessage(params, List.of(), null, null); + + await().atMost(10, SECONDS).until(() -> { + List> metrics = getMetrics(); + return metrics.stream().anyMatch(m -> GENAI_STREAMING_DURATION.equals(m.get("name"))); + }); + + List> metrics = getMetrics(); + Map streamingMetric = metrics.stream() + .filter(m -> GENAI_STREAMING_DURATION.equals(m.get("name"))) + .findFirst() + .orElseThrow(() -> new AssertionError("No streaming duration metric found")); + + assertEquals("HISTOGRAM", streamingMetric.get("type"), + "Streaming duration metric should be a histogram"); + assertTrue(((Number) streamingMetric.get("data_count")).intValue() > 0, + "Histogram should have at least one data point"); + } + protected void saveTaskInTaskStore(Task task) throws Exception { HttpClient httpClient = HttpClient.newBuilder() .version(HttpClient.Version.HTTP_2) @@ -297,4 +387,5 @@ protected void ensureQueueForTask(String taskId) throws Exception { throw new RuntimeException(String.format("Ensuring queue failed! Status: %d, Body: %s", response.statusCode(), response.body())); } } + } diff --git a/extras/opentelemetry/integration-tests/src/test/resources/application.properties b/extras/opentelemetry/integration-tests/src/test/resources/application.properties index 932acdd58..b5eb37db9 100644 --- a/extras/opentelemetry/integration-tests/src/test/resources/application.properties +++ b/extras/opentelemetry/integration-tests/src/test/resources/application.properties @@ -7,7 +7,8 @@ quarkus.http.test-port=8081 # OpenTelemetry configuration for testing quarkus.otel.sdk.disabled=false quarkus.otel.traces.enabled=true -quarkus.otel.metrics.enabled=false +quarkus.otel.metrics.enabled=true +quarkus.otel.metrics.exporter=cdi quarkus.otel.logs.enabled=false quarkus.otel.instrument.vertx-http=false diff --git a/extras/opentelemetry/server/pom.xml b/extras/opentelemetry/server/pom.xml index cd23f8cab..c07a23353 100644 --- a/extras/opentelemetry/server/pom.xml +++ b/extras/opentelemetry/server/pom.xml @@ -48,8 +48,10 @@ jakarta.inject-api - org.slf4j - slf4j-api + ${project.groupId} + a2a-java-sdk-opentelemetry-common + tests + test org.junit.jupiter diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java index 2166b06b9..ba1f7a615 100644 --- a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java +++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecorator.java @@ -9,13 +9,40 @@ import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_MESSAGE_ID; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_PARTS_NUMBER; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_PROTOCOL_VERSION; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_REQUEST; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_ROLE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_STREAMING_DURATION; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_STREAMING_EVENT; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TRANSPORT; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_END_SUFFIX; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_EVENT_SUFFIX; +import io.opentelemetry.api.metrics.DoubleHistogram; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import jakarta.annotation.Priority; +import jakarta.decorator.Decorator; +import jakarta.decorator.Delegate; +import jakarta.enterprise.inject.Any; +import jakarta.enterprise.inject.Instance; +import jakarta.inject.Inject; +import java.util.Map; +import java.util.concurrent.Flow; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.logging.Logger; +import org.a2aproject.sdk.extras.opentelemetry.StreamingEventAttributes; import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.server.auth.TaskOperation; import org.a2aproject.sdk.server.requesthandlers.RequestHandler; import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.A2AMethods; @@ -27,26 +54,13 @@ import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; import org.a2aproject.sdk.spec.ListTasksParams; import org.a2aproject.sdk.spec.MessageSendParams; -import org.a2aproject.sdk.server.auth.TaskOperation; import org.a2aproject.sdk.spec.StreamingEventKind; import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskIdParams; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; import org.a2aproject.sdk.spec.TaskQueryParams; -import io.opentelemetry.api.trace.Span; -import io.opentelemetry.api.trace.SpanKind; -import io.opentelemetry.api.trace.StatusCode; -import io.opentelemetry.api.trace.Tracer; -import io.opentelemetry.context.Scope; -import jakarta.annotation.Priority; -import jakarta.decorator.Decorator; -import jakarta.decorator.Delegate; -import jakarta.enterprise.inject.Any; -import jakarta.inject.Inject; -import java.util.concurrent.Flow; +import org.a2aproject.sdk.spec.TransportProtocol; import org.jspecify.annotations.Nullable; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * OpenTelemetry CDI Decorator for {@link RequestHandler}. @@ -54,10 +68,11 @@ * This decorator adds distributed tracing to A2A server request handlers. * It creates spans for each request handler method invocation, capturing: *

    - *
  • Request parameters as span attributes
    • - *
  • Response data as span attributes
    • - *
  • Errors and exceptions with proper status codes
    • - *
  • Timing information for performance monitoring
    • + *
  • Request parameters as span attributes
    • + *
  • Response data as span attributes
    • + *
  • Errors and exceptions with proper status codes
    • + *
  • Timing information for performance monitoring
    • + *
  • Streaming operation duration via {@code gen_ai.agent.a2a.streaming.duration} histogram
    • *
*

* To enable this decorator, add it to your beans.xml: @@ -71,8 +86,6 @@ @Priority(100) public abstract class OpenTelemetryRequestHandlerDecorator implements RequestHandler { - private static final Logger LOGGER = LoggerFactory.getLogger(OpenTelemetryRequestHandlerDecorator.class); - @Inject @Delegate @Any @@ -81,6 +94,12 @@ public abstract class OpenTelemetryRequestHandlerDecorator implements RequestHan @Inject private Tracer tracer; + @Inject + private Instance meterInstance; + + @Nullable + private volatile DoubleHistogram streamingDurationHistogram; + /** * Default constructor for CDI. */ @@ -88,14 +107,49 @@ public OpenTelemetryRequestHandlerDecorator() { } /** - * Constructor for testing. + * Constructor for testing without metrics. * * @param delegate the delegate request handler * @param tracer the tracer to use */ public OpenTelemetryRequestHandlerDecorator(RequestHandler delegate, Tracer tracer) { + this(delegate, tracer, null); + } + + /** + * Constructor for testing with metrics. + * + * @param delegate the delegate request handler + * @param tracer the tracer to use + * @param meter optional meter for streaming duration histogram; null disables metrics + */ + public OpenTelemetryRequestHandlerDecorator(RequestHandler delegate, Tracer tracer, @Nullable Meter meter) { this.delegate = delegate; this.tracer = tracer; + this.streamingDurationHistogram = buildStreamingDurationHistogram(meter); + } + + @Nullable + private DoubleHistogram streamingDurationHistogram() { + if (streamingDurationHistogram == null && meterInstance != null && meterInstance.isResolvable()) { + synchronized (this) { + if (streamingDurationHistogram == null) { + streamingDurationHistogram = buildStreamingDurationHistogram(meterInstance.get()); + } + } + } + return streamingDurationHistogram; + } + + @Nullable + private static DoubleHistogram buildStreamingDurationHistogram(@Nullable Meter meter) { + if (meter == null) { + return null; + } + return meter.histogramBuilder(GENAI_STREAMING_DURATION) + .setUnit("s") + .setDescription("Duration of A2A streaming operations from initiation to last event") + .build(); } @Override @@ -111,6 +165,7 @@ public Task onGetTask(TaskQueryParams params, ServerCallContext context) throws spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -143,6 +198,7 @@ public ListTasksResult onListTasks(ListTasksParams params, ServerCallContext con if (params.contextId() != null) { spanBuilder.setAttribute(GENAI_CONTEXT_ID, params.contextId()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -176,6 +232,7 @@ public Task onCancelTask(CancelTaskParams params, ServerCallContext context) thr spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -224,6 +281,7 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -272,7 +330,9 @@ public Flow.Publisher onMessageSendStream(MessageSendParams spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); + long startNanos = System.nanoTime(); try (Scope scope = span.makeCurrent()) { Flow.Publisher result = delegate.onMessageSendStream(params, context); @@ -282,8 +342,13 @@ public Flow.Publisher onMessageSendStream(MessageSendParams } span.setStatus(StatusCode.OK); - return result; + SpanContext spanContext = span.getSpanContext(); + return new OpenTelemetryStreamPublisher(result, + new StreamingSpanConfig(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD, + spanContext, streamingDurationHistogram(), startNanos)); } catch (A2AError error) { + StreamingEventAttributes.recordDuration(streamingDurationHistogram(), + startNanos, A2AMethods.SEND_STREAMING_MESSAGE_METHOD, false); span.setAttribute(ERROR_TYPE, error.getMessage()); span.setStatus(StatusCode.ERROR, error.getMessage()); throw error; @@ -308,6 +373,7 @@ public TaskPushNotificationConfig onCreateTaskPushNotificationConfig(TaskPushNot spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -344,6 +410,7 @@ public TaskPushNotificationConfig onGetTaskPushNotificationConfig(GetTaskPushNot spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -377,7 +444,9 @@ public Flow.Publisher onSubscribeToTask(TaskIdParams params, spanBuilder.setAttribute(GENAI_REQUEST, params.toString()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); + long startNanos = System.nanoTime(); try (Scope scope = span.makeCurrent()) { Flow.Publisher result = delegate.onSubscribeToTask(params, context); @@ -387,8 +456,13 @@ public Flow.Publisher onSubscribeToTask(TaskIdParams params, } span.setStatus(StatusCode.OK); - return result; + SpanContext spanContext = span.getSpanContext(); + return new OpenTelemetryStreamPublisher(result, + new StreamingSpanConfig(tracer, A2AMethods.SUBSCRIBE_TO_TASK_METHOD, + spanContext, streamingDurationHistogram(), startNanos)); } catch (A2AError error) { + StreamingEventAttributes.recordDuration(streamingDurationHistogram(), + startNanos, A2AMethods.SUBSCRIBE_TO_TASK_METHOD, false); span.setAttribute(ERROR_TYPE, error.getMessage()); span.setStatus(StatusCode.ERROR, error.getMessage()); throw error; @@ -410,6 +484,7 @@ public ListTaskPushNotificationConfigsResult onListTaskPushNotificationConfigs(L spanBuilder.setAttribute(GENAI_TASK_ID, params.id()); } + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); try (Scope scope = span.makeCurrent()) { @@ -442,12 +517,10 @@ public void onDeleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigP if (params.taskId() != null) { spanBuilder.setAttribute(GENAI_TASK_ID, params.taskId()); } - + applyTransportAttributes(spanBuilder, context); Span span = spanBuilder.startSpan(); - try (Scope scope = span.makeCurrent()) { delegate.onDeleteTaskPushNotificationConfig(params, context); - span.setStatus(StatusCode.OK); } catch (A2AError error) { span.setAttribute(ERROR_TYPE, error.getMessage()); @@ -471,4 +544,146 @@ private boolean extractRequest() { private boolean extractResponse() { return Boolean.getBoolean(EXTRACT_RESPONSE_SYS_PROPERTY); } + + private static void applyTransportAttributes(SpanBuilder spanBuilder, ServerCallContext context) { + Map state = context.getState(); + Object transport = state.get(ServerCallContext.TRANSPORT_KEY); + if (transport instanceof TransportProtocol tp) { + spanBuilder.setAttribute(GENAI_TRANSPORT, tp.asString()); + } + String version = context.getRequestedProtocolVersion(); + if (version != null) { + spanBuilder.setAttribute(GENAI_PROTOCOL_VERSION, version); + } + } + + private record StreamingSpanConfig(Tracer tracer, String spanName, SpanContext parentSpanContext, + @Nullable DoubleHistogram streamingDurationHistogram, long startNanos) {} + + private static class OpenTelemetryStreamPublisher implements Flow.Publisher { + + private final Flow.Publisher delegate; + private final StreamingSpanConfig config; + + OpenTelemetryStreamPublisher(Flow.Publisher delegate, StreamingSpanConfig config) { + this.delegate = delegate; + this.config = config; + } + + @Override + public void subscribe(Flow.Subscriber subscriber) { + delegate.subscribe(new OpenTelemetryStreamSubscriber(subscriber, config)); + } + } + + private static class OpenTelemetryStreamSubscriber implements Flow.Subscriber { + + private static final Logger LOGGER = Logger.getLogger(OpenTelemetryStreamSubscriber.class.getName()); + + private final Flow.Subscriber delegate; + private final StreamingSpanConfig config; + private final AtomicBoolean terminated = new AtomicBoolean(false); + + OpenTelemetryStreamSubscriber(Flow.Subscriber delegate, StreamingSpanConfig config) { + this.delegate = delegate; + this.config = config; + } + + @Override + public void onSubscribe(Flow.Subscription subscription) { + delegate.onSubscribe(new Flow.Subscription() { + @Override + public void request(long n) { + subscription.request(n); + } + + @Override + public void cancel() { + if (terminated.compareAndSet(false, true)) { + recordStreamingDuration(true); + Span closingSpan = config.tracer().spanBuilder(config.spanName() + STREAMING_END_SUFFIX) + .setSpanKind(SpanKind.SERVER) + .addLink(config.parentSpanContext()) + .startSpan(); + try (Scope ignored = closingSpan.makeCurrent()) { + closingSpan.setStatus(StatusCode.OK); + } finally { + closingSpan.end(); + } + } + subscription.cancel(); + } + }); + } + + @Override + public void onNext(StreamingEventKind item) { + SpanBuilder builder = config.tracer().spanBuilder(config.spanName() + STREAMING_EVENT_SUFFIX) + .setSpanKind(SpanKind.SERVER) + .addLink(config.parentSpanContext()) + .setAttribute(GENAI_STREAMING_EVENT, item.kind()); + StreamingEventAttributes.apply(builder, item); + Span span = builder.startSpan(); + try (Scope ignored = span.makeCurrent()) { + delegate.onNext(item); + span.setStatus(StatusCode.OK); + } catch (Throwable t) { + span.recordException(t); + span.setStatus(StatusCode.ERROR); + throw t; + } finally { + span.end(); + } + } + + @Override + public void onError(Throwable throwable) { + if (!terminated.compareAndSet(false, true)) { + LOGGER.warning("onError() received after stream termination (publisher protocol violation); ignoring: " + throwable); + return; + } + recordStreamingDuration(false); + Span closingSpan = config.tracer().spanBuilder(config.spanName() + STREAMING_END_SUFFIX) + .setSpanKind(SpanKind.SERVER) + .addLink(config.parentSpanContext()) + .startSpan(); + try (Scope ignored = closingSpan.makeCurrent()) { + closingSpan.setAttribute(ERROR_TYPE, throwable.getClass().getName()); + closingSpan.setStatus(StatusCode.ERROR, throwable.getMessage()); + delegate.onError(throwable); + } catch (Throwable t) { + closingSpan.recordException(t); + throw t; + } finally { + closingSpan.end(); + } + } + + @Override + public void onComplete() { + if (!terminated.compareAndSet(false, true)) { + return; + } + recordStreamingDuration(true); + Span closingSpan = config.tracer().spanBuilder(config.spanName() + STREAMING_END_SUFFIX) + .setSpanKind(SpanKind.SERVER) + .addLink(config.parentSpanContext()) + .startSpan(); + try (Scope ignored = closingSpan.makeCurrent()) { + delegate.onComplete(); + closingSpan.setStatus(StatusCode.OK); + } catch (Throwable t) { + closingSpan.recordException(t); + closingSpan.setStatus(StatusCode.ERROR); + throw t; + } finally { + closingSpan.end(); + } + } + + private void recordStreamingDuration(boolean success) { + StreamingEventAttributes.recordDuration(config.streamingDurationHistogram(), + config.startNanos(), config.spanName(), success); + } + } } diff --git a/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml index e4652bc27..a79cffb3c 100644 --- a/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml +++ b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml @@ -6,4 +6,4 @@ org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryRequestHandlerDecorator - \ No newline at end of file + diff --git a/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecoratorTest.java b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecoratorTest.java index 042cd8595..8aeb5e3bb 100644 --- a/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecoratorTest.java +++ b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryRequestHandlerDecoratorTest.java @@ -5,9 +5,15 @@ import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_REQUEST; import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_END_SUFFIX; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.STREAMING_EVENT_SUFFIX; +import static org.a2aproject.sdk.extras.opentelemetry.OtelSpanMocks.mockClosingSpanBuilder; +import static org.a2aproject.sdk.extras.opentelemetry.OtelSpanMocks.mockEventSpanBuilder; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyDouble; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; @@ -20,14 +26,21 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.DoubleHistogram; +import io.opentelemetry.api.metrics.DoubleHistogramBuilder; +import io.opentelemetry.api.metrics.Meter; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanBuilder; +import io.opentelemetry.api.trace.SpanContext; import io.opentelemetry.api.trace.SpanKind; import io.opentelemetry.api.trace.StatusCode; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.context.Scope; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.Flow; import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.server.ServerCallContext; @@ -53,13 +66,18 @@ import org.a2aproject.sdk.spec.TaskState; import org.a2aproject.sdk.spec.TaskStatus; import org.a2aproject.sdk.spec.TextPart; +import org.a2aproject.sdk.spec.TransportProtocol; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_PROTOCOL_VERSION; +import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TRANSPORT; + @ExtendWith(MockitoExtension.class) class OpenTelemetryRequestHandlerDecoratorTest { @@ -75,12 +93,24 @@ class OpenTelemetryRequestHandlerDecoratorTest { @Mock private Scope scope; + @Mock + private SpanContext spanContext; + @Mock private ServerCallContext context; @Mock private RequestHandler delegate; + @Mock + private Meter meter; + + @Mock + private DoubleHistogramBuilder histogramBuilder; + + @Mock + private DoubleHistogram streamingHistogram; + private TestableOpenTelemetryRequestHandlerDecorator decorator; @BeforeEach @@ -94,14 +124,25 @@ void setUp() { lenient().when(spanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(spanBuilder); lenient().when(spanBuilder.setAttribute(anyString(), anyString())).thenReturn(spanBuilder); lenient().when(spanBuilder.setAttribute(anyString(), anyLong())).thenReturn(spanBuilder); + lenient().when(spanBuilder.addLink(any(SpanContext.class))).thenReturn(spanBuilder); lenient().when(spanBuilder.startSpan()).thenReturn(span); lenient().when(span.makeCurrent()).thenReturn(scope); + lenient().when(span.getSpanContext()).thenReturn(spanContext); lenient().when(span.setAttribute(anyString(), anyString())).thenReturn(span); lenient().when(span.setStatus(any(StatusCode.class))).thenReturn(span); lenient().when(span.setStatus(any(StatusCode.class), anyString())).thenReturn(span); + // Set up meter mock chain + lenient().when(meter.histogramBuilder(anyString())).thenReturn(histogramBuilder); + lenient().when(histogramBuilder.setUnit(anyString())).thenReturn(histogramBuilder); + lenient().when(histogramBuilder.setDescription(anyString())).thenReturn(histogramBuilder); + lenient().when(histogramBuilder.build()).thenReturn(streamingHistogram); + + // Return an empty state map by default so applyTransportProtocol doesn't NPE + lenient().when(context.getState()).thenReturn(new HashMap<>()); + // Create decorator with mocked dependencies - decorator = new TestableOpenTelemetryRequestHandlerDecorator(delegate, tracer); + decorator = new TestableOpenTelemetryRequestHandlerDecorator(delegate, tracer, meter); } /** @@ -111,8 +152,13 @@ static class TestableOpenTelemetryRequestHandlerDecorator extends OpenTelemetryR public TestableOpenTelemetryRequestHandlerDecorator(RequestHandler delegate, Tracer tracer) { super(delegate, tracer); } + + public TestableOpenTelemetryRequestHandlerDecorator(RequestHandler delegate, Tracer tracer, Meter meter) { + super(delegate, tracer, meter); + } } + @Nested class GetTaskTests { @Test @@ -279,7 +325,7 @@ void onMessageSend_withError_setsErrorStatus() throws A2AError { @Nested class MessageSendStreamTests { @Test - void onMessageSendStream_createsSpanWithSpecialMessage() throws A2AError { + void onMessageSendStream_createsSpanAndWrapsPublisher() throws A2AError { Message message = Message.builder() .role(Message.Role.ROLE_USER) .parts(List.of(new TextPart("test message"))) @@ -293,11 +339,12 @@ void onMessageSendStream_createsSpanWithSpecialMessage() throws A2AError { Flow.Publisher actualResult = decorator.onMessageSendStream(params, context); - assertEquals(publisher, actualResult); + assertNotNull(actualResult); verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD); verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString()); verify(span).setAttribute(GENAI_RESPONSE, "Stream publisher created"); verify(span).setStatus(StatusCode.OK); + verify(span).end(); } @Test @@ -318,6 +365,259 @@ void onMessageSendStream_withError_setsErrorStatus() throws A2AError { verify(span).setAttribute(ERROR_TYPE, error.getMessage()); verify(span).setStatus(StatusCode.ERROR, error.getMessage()); verify(span).end(); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void onMessageSendStream_closingSpan_createdOnComplete() throws A2AError { + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test message"))) + .messageId("msg-123") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onMessageSendStream(params, context)).thenReturn(publisher); + + Span closingSpan = mock(Span.class); + SpanBuilder closingSpanBuilder = mockClosingSpanBuilder(tracer, + A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onMessageSendStream(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + otelSubscriber.onComplete(); + + verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX); + verify(closingSpanBuilder).addLink(spanContext); + verify(closingSpan).setStatus(StatusCode.OK); + verify(closingSpan).end(); + verify(testSubscriber).onComplete(); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void onMessageSendStream_closingSpan_recordsEventsAndEndsOnError() throws A2AError { + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test message"))) + .messageId("msg-123") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onMessageSendStream(params, context)).thenReturn(publisher); + StreamingEventKind eventItem = Message.builder() + .role(Message.Role.ROLE_AGENT) + .parts(List.of(new TextPart("event-data"))) + .messageId("event-msg-1") + .build(); + + Span eventSpan = mock(Span.class); + SpanBuilder eventSpanBuilder = mockEventSpanBuilder(tracer, + A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_EVENT_SUFFIX, eventSpan); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onMessageSendStream(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + otelSubscriber.onNext(eventItem); + Throwable error = new RuntimeException("stream failure"); + otelSubscriber.onError(error); + + verify(testSubscriber).onNext(eventItem); + verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_EVENT_SUFFIX); + verify(eventSpanBuilder).setSpanKind(SpanKind.SERVER); + verify(eventSpan).end(); + verify(closingSpan).setAttribute(ERROR_TYPE, RuntimeException.class.getName()); + verify(closingSpan).setStatus(StatusCode.ERROR, "stream failure"); + verify(closingSpan).end(); + verify(testSubscriber).onError(error); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void onMessageSendStream_closingSpan_createdOnSubscriptionCancel() throws A2AError { + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test message"))) + .messageId("msg-123") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onMessageSendStream(params, context)).thenReturn(publisher); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onMessageSendStream(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + // Simulate onSubscribe — the OTel subscriber wraps the subscription before passing it down + Flow.Subscription rawSubscription = mock(Flow.Subscription.class); + otelSubscriber.onSubscribe(rawSubscription); + + // Capture the wrapped subscription that was forwarded to the downstream subscriber + ArgumentCaptor subscriptionCaptor = ArgumentCaptor.forClass(Flow.Subscription.class); + verify(testSubscriber).onSubscribe(subscriptionCaptor.capture()); + + // Cancel via the wrapped subscription — this must produce the terminal span and histogram + subscriptionCaptor.getValue().cancel(); + + verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX); + verify(closingSpan).setStatus(StatusCode.OK); + verify(closingSpan).end(); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + // The underlying subscription must also be cancelled + verify(rawSubscription).cancel(); + } + + @Test + @SuppressWarnings("unchecked") + void onMessageSendStream_closingSpan_cancelledTwiceOnlyEmitsOneSpan() throws A2AError { + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test message"))) + .messageId("msg-123") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onMessageSendStream(params, context)).thenReturn(publisher); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onMessageSendStream(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + Flow.Subscription rawSubscription = mock(Flow.Subscription.class); + otelSubscriber.onSubscribe(rawSubscription); + + ArgumentCaptor subscriptionCaptor = ArgumentCaptor.forClass(Flow.Subscription.class); + verify(testSubscriber).onSubscribe(subscriptionCaptor.capture()); + + subscriptionCaptor.getValue().cancel(); + subscriptionCaptor.getValue().cancel(); + + // Terminal span created exactly once despite two cancels + verify(closingSpan, times(1)).end(); + verify(streamingHistogram, times(1)).record(anyDouble(), any(Attributes.class)); + verify(rawSubscription, times(2)).cancel(); + } + + @Test + @SuppressWarnings("unchecked") + void onMessageSendStream_closingSpan_recordsMultipleEventsOnComplete() throws A2AError { + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test message"))) + .messageId("msg-123") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onMessageSendStream(params, context)).thenReturn(publisher); + StreamingEventKind event1 = Message.builder() + .role(Message.Role.ROLE_AGENT) + .parts(List.of(new TextPart("first event"))) + .messageId("event-1") + .build(); + StreamingEventKind event2 = Message.builder() + .role(Message.Role.ROLE_AGENT) + .parts(List.of(new TextPart("second event"))) + .messageId("event-2") + .build(); + + Span eventSpan = mock(Span.class); + SpanBuilder eventSpanBuilder = mockEventSpanBuilder(tracer, + A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_EVENT_SUFFIX, eventSpan); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onMessageSendStream(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + otelSubscriber.onNext(event1); + otelSubscriber.onNext(event2); + otelSubscriber.onComplete(); + + verify(testSubscriber).onNext(event1); + verify(testSubscriber).onNext(event2); + verify(tracer, times(2)).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_EVENT_SUFFIX); + verify(eventSpanBuilder, times(2)).setSpanKind(SpanKind.SERVER); + verify(eventSpan, times(2)).end(); + verify(closingSpan).setStatus(StatusCode.OK); + verify(closingSpan).end(); + verify(testSubscriber).onComplete(); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void onMessageSendStream_closingSpan_delegateThrowOnComplete_recordsExceptionAndSetsErrorStatus() throws A2AError { + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test message"))) + .messageId("msg-ex") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onMessageSendStream(params, context)).thenReturn(publisher); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onMessageSendStream(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + RuntimeException delegateError = new RuntimeException("delegate onComplete failure"); + doThrow(delegateError).when(testSubscriber).onComplete(); + + assertThrows(RuntimeException.class, otelSubscriber::onComplete); + + verify(closingSpan).recordException(delegateError); + verify(closingSpan).setStatus(StatusCode.ERROR); + verify(closingSpan).end(); } } @@ -392,18 +692,19 @@ void onGetTaskPushNotificationConfig_withError_setsErrorStatus() throws A2AError @Nested class ResubscribeToTaskTests { @Test - void onResubscribeToTask_createsSpanWithSpecialMessage() throws A2AError { + void onResubscribeToTask_createsSpanAndWrapsPublisher() throws A2AError { TaskIdParams params = new TaskIdParams("task-123"); Flow.Publisher publisher = mock(Flow.Publisher.class); when(delegate.onSubscribeToTask(params, context)).thenReturn(publisher); Flow.Publisher actualResult = decorator.onSubscribeToTask(params, context); - assertEquals(publisher, actualResult); + assertNotNull(actualResult); verify(tracer).spanBuilder(A2AMethods.SUBSCRIBE_TO_TASK_METHOD); verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString()); verify(span).setAttribute(GENAI_RESPONSE, "Stream publisher created"); verify(span).setStatus(StatusCode.OK); + verify(span).end(); } @Test @@ -417,6 +718,117 @@ void onResubscribeToTask_withError_setsErrorStatus() throws A2AError { verify(span).setAttribute(ERROR_TYPE, error.getMessage()); verify(span).setStatus(StatusCode.ERROR, error.getMessage()); verify(span).end(); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void onResubscribeToTask_closingSpan_createdOnComplete() throws A2AError { + TaskIdParams params = new TaskIdParams("task-123"); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onSubscribeToTask(params, context)).thenReturn(publisher); + + Span closingSpan = mock(Span.class); + SpanBuilder closingSpanBuilder = mockClosingSpanBuilder(tracer, + A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onSubscribeToTask(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + otelSubscriber.onComplete(); + + verify(tracer).spanBuilder(A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_END_SUFFIX); + verify(closingSpanBuilder).addLink(spanContext); + verify(closingSpan).setStatus(StatusCode.OK); + verify(closingSpan).end(); + verify(testSubscriber).onComplete(); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void onResubscribeToTask_closingSpan_recordsEventsOnCompleteAndError() throws A2AError { + TaskIdParams params = new TaskIdParams("task-123"); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onSubscribeToTask(params, context)).thenReturn(publisher); + StreamingEventKind eventItem = Message.builder() + .role(Message.Role.ROLE_AGENT) + .parts(List.of(new TextPart("streamed-event"))) + .messageId("evt-1") + .build(); + + Span eventSpan = mock(Span.class); + SpanBuilder eventSpanBuilder = mockEventSpanBuilder(tracer, + A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_EVENT_SUFFIX, eventSpan); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onSubscribeToTask(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + otelSubscriber.onNext(eventItem); + Throwable error = new RuntimeException("subscribe failure"); + otelSubscriber.onError(error); + + verify(testSubscriber).onNext(eventItem); + verify(tracer).spanBuilder(A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_EVENT_SUFFIX); + verify(eventSpanBuilder).setSpanKind(SpanKind.SERVER); + verify(eventSpan).end(); + verify(closingSpan).setAttribute(ERROR_TYPE, RuntimeException.class.getName()); + verify(closingSpan).setStatus(StatusCode.ERROR, "subscribe failure"); + verify(closingSpan).end(); + verify(testSubscriber).onError(error); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + } + + @Test + @SuppressWarnings("unchecked") + void onResubscribeToTask_closingSpan_createdOnSubscriptionCancel() throws A2AError { + TaskIdParams params = new TaskIdParams("task-123"); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onSubscribeToTask(params, context)).thenReturn(publisher); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decorator.onSubscribeToTask(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + // Simulate onSubscribe — the OTel subscriber wraps the subscription before passing it down + Flow.Subscription rawSubscription = mock(Flow.Subscription.class); + otelSubscriber.onSubscribe(rawSubscription); + + // Capture the wrapped subscription forwarded to the downstream subscriber + ArgumentCaptor subscriptionCaptor = ArgumentCaptor.forClass(Flow.Subscription.class); + verify(testSubscriber).onSubscribe(subscriptionCaptor.capture()); + + // Cancel via the wrapped subscription — this must produce the terminal span and histogram + subscriptionCaptor.getValue().cancel(); + + verify(tracer).spanBuilder(A2AMethods.SUBSCRIBE_TO_TASK_METHOD + STREAMING_END_SUFFIX); + verify(closingSpan).setStatus(StatusCode.OK); + verify(closingSpan).end(); + verify(streamingHistogram).record(anyDouble(), any(Attributes.class)); + verify(rawSubscription).cancel(); } } @@ -482,6 +894,103 @@ void onDeleteTaskPushNotificationConfig_withError_setsErrorStatus() throws A2AEr } } + @Nested + class NullMeterTests { + @Test + @SuppressWarnings("unchecked") + void onMessageSendStream_withNullMeter_doesNotRecordHistogram() throws A2AError { + TestableOpenTelemetryRequestHandlerDecorator decoratorNoMeter = + new TestableOpenTelemetryRequestHandlerDecorator(delegate, tracer); + + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test message"))) + .messageId("msg-123") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + Flow.Publisher publisher = mock(Flow.Publisher.class); + when(delegate.onMessageSendStream(params, context)).thenReturn(publisher); + + Span closingSpan = mock(Span.class); + mockClosingSpanBuilder(tracer, A2AMethods.SEND_STREAMING_MESSAGE_METHOD + STREAMING_END_SUFFIX, closingSpan); + + Flow.Publisher result = decoratorNoMeter.onMessageSendStream(params, context); + + Flow.Subscriber testSubscriber = mock(Flow.Subscriber.class); + result.subscribe(testSubscriber); + + ArgumentCaptor subscriberCaptor = ArgumentCaptor.forClass(Flow.Subscriber.class); + verify(publisher).subscribe(subscriberCaptor.capture()); + Flow.Subscriber otelSubscriber = (Flow.Subscriber) subscriberCaptor.getValue(); + + otelSubscriber.onComplete(); + + verify(closingSpan).setStatus(StatusCode.OK); + verify(closingSpan).end(); + verify(testSubscriber).onComplete(); + verify(streamingHistogram, never()).record(anyDouble(), any(Attributes.class)); + } + } + + @Nested + class TransportProtocolTests { + @Test + void onMessageSend_setsGenaiProtocolAttribute_whenTransportKeyPresent() throws A2AError { + Map state = new HashMap<>(); + state.put(ServerCallContext.TRANSPORT_KEY, TransportProtocol.JSONRPC); + when(context.getState()).thenReturn(state); + when(context.getRequestedProtocolVersion()).thenReturn(null); + + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test"))) + .messageId("msg-1") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + when(delegate.onMessageSend(params, context)).thenReturn(mock(EventKind.class)); + + decorator.onMessageSend(params, context); + + verify(spanBuilder).setAttribute(GENAI_TRANSPORT, TransportProtocol.JSONRPC.asString()); + verify(spanBuilder, never()).setAttribute(eq(GENAI_PROTOCOL_VERSION), anyString()); + } + + @Test + void onMessageSend_setsGenaiVersionAttribute_whenVersionPresent() throws A2AError { + when(context.getRequestedProtocolVersion()).thenReturn("1.0"); + + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test"))) + .messageId("msg-1") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + when(delegate.onMessageSend(params, context)).thenReturn(mock(EventKind.class)); + + decorator.onMessageSend(params, context); + + verify(spanBuilder).setAttribute(GENAI_PROTOCOL_VERSION, "1.0"); + } + + @Test + void onMessageSend_doesNotSetGenaiProtocolAttribute_whenTransportKeyAbsent() throws A2AError { + // context.getState() returns empty HashMap and getRequestedProtocolVersion() returns null from setUp + + Message message = Message.builder() + .role(Message.Role.ROLE_USER) + .parts(List.of(new TextPart("test"))) + .messageId("msg-1") + .build(); + MessageSendParams params = new MessageSendParams(message, null, null, ""); + when(delegate.onMessageSend(params, context)).thenReturn(mock(EventKind.class)); + + decorator.onMessageSend(params, context); + + verify(spanBuilder, never()).setAttribute(eq(GENAI_TRANSPORT), anyString()); + verify(spanBuilder, never()).setAttribute(eq(GENAI_PROTOCOL_VERSION), anyString()); + } + } + @Nested class SpanLifecycleTests { @Test diff --git a/pom.xml b/pom.xml index 37bad716c..b7524bcae 100644 --- a/pom.xml +++ b/pom.xml @@ -240,6 +240,12 @@ a2a-java-sdk-opentelemetry-common ${project.version} + + org.a2aproject.sdk + a2a-java-sdk-opentelemetry-common + ${project.version} + tests + org.a2aproject.sdk a2a-java-sdk-opentelemetry-client