Skip to content

Commit c56957f

Browse files
committed
feat(opentelemetry): add streaming spans, metrics, and push tracing
- Two-span model for server streaming: method span + closing span with events and link - Client streaming events recorded as span events on the parent span - gen_ai.client.operation.duration and gen_ai.agent.a2a.streaming.duration histograms - OpenTelemetryPushNotificationSenderDecorator with CLIENT spans for push delivery Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>
1 parent a5e76a0 commit c56957f

21 files changed

Lines changed: 1483 additions & 91 deletions

File tree

‎docs/content/dev/extra/opentelemetry.md‎

Lines changed: 104 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ Adds distributed tracing and context propagation to A2A servers and clients usin
1414
- **Context Propagation**: OpenTelemetry trace context propagation across async operations
1515
- **Request/Response Logging**: Optional extraction of request and response data into spans
1616
- **Error Tracking**: Automatic error status and error type attributes on failures
17+
- **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
1718

1819
## Modules
1920

@@ -65,6 +66,74 @@ Async agent execution (maintains trace context)
6566
Response (with trace headers)
6667
```
6768

69+
### Streaming Span Lifecycle
70+
71+
Streaming methods (`message/stream` and `tasks/resubscribe`) follow a two-span model to correctly represent the full duration of a streaming response.
72+
73+
**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).
74+
75+
**Span 2 — end span** (`<method>-end`) is created when the stream terminates — either on completion or error. It carries a [link](https://opentelemetry.io/docs/concepts/signals/traces/#span-links) to Span 1 for correlation, and records every A2A event that was published as a [span event](https://opentelemetry.io/docs/concepts/signals/traces/#span-events).
76+
77+
<pre class="mermaid">
78+
sequenceDiagram
79+
participant C as Client
80+
participant D as OTel Decorator
81+
participant H as RequestHandler
82+
participant W as WrappedPublisher
83+
84+
C->>D: onMessageSendStream(params)
85+
D->>D: create Span₁ (SERVER)
86+
D->>H: onMessageSendStream(params)
87+
H-->>D: publisher
88+
D->>D: capture Span₁.context
89+
D->>D: Span₁.setStatus(OK) · Span₁.end()
90+
D-->>C: WrappedPublisher
91+
92+
note over D: Span₁ closed immediately
93+
94+
loop for each streaming event
95+
W-->>C: onNext(event)
96+
W->>W: collect PendingEvent(name, attributes)
97+
end
98+
99+
alt stream completes normally
100+
W->>W: create Span₂ (SERVER, link→Span₁)
101+
W->>W: addEvent per PendingEvent
102+
W->>W: Span₂.setStatus(OK) · Span₂.end()
103+
W-->>C: onComplete()
104+
else stream errors
105+
W->>W: create Span₂ (SERVER, link→Span₁)
106+
W->>W: addEvent per PendingEvent
107+
W->>W: Span₂.setStatus(ERROR) · Span₂.end()
108+
W-->>C: onError(throwable)
109+
end
110+
</pre>
111+
112+
Each span event on Span₂ carries:
113+
114+
| Attribute | Value |
115+
|-----------|-------|
116+
| `gen_ai.agent.a2a.streaming-event` | `toString()` of the published `StreamingEventKind` |
117+
| `gen_ai.agent.a2a.status.code` | `OK` |
118+
119+
### Metrics
120+
121+
The server decorator records the following histogram instruments:
122+
123+
| Instrument | Type | Unit | Description |
124+
|------------|------|------|-------------|
125+
| `gen_ai.agent.a2a.streaming.duration` | Histogram | `s` | Duration from stream initiation to last event, for `message/stream` and `tasks/resubscribe` |
126+
127+
Attributes on `gen_ai.agent.a2a.streaming.duration`:
128+
129+
| Attribute | Value |
130+
|-----------|-------|
131+
| `gen_ai.agent.a2a.operation.name` | A2A method name (`message/stream` or `tasks/resubscribe`) |
132+
| `gen_ai.system` | `a2a` |
133+
| `error.type` | `error` (only on stream failure) |
134+
135+
Meter injection uses CDI `Instance<Meter>`, so the histogram is silently skipped if no `Meter` bean is available in the container.
136+
68137
### Context-Aware Async Executor
69138

70139
> **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.
@@ -106,6 +175,19 @@ The following attributes are automatically added to spans (defined in `A2AObserv
106175
| `gen_ai.agent.a2a.response` | Full response data (only if extraction enabled) |
107176
| `error.type` | Error message (on failures) |
108177

178+
### Push Notification Delivery Tracing
179+
180+
When the server delivers a push notification to a client webhook, `OpenTelemetryPushNotificationSenderDecorator` creates a `CLIENT` span named `SendPushNotification`. This span is independent of the original request span but carries the `gen_ai.agent.a2a.task_id` attribute, allowing correlation by task ID in your observability tool.
181+
182+
| Attribute | Description |
183+
|-----------|-------------|
184+
| `gen_ai.agent.a2a.operation.name` | `SendPushNotification` |
185+
| `gen_ai.agent.a2a.task_id` | Task identifier |
186+
| `gen_ai.agent.a2a.context_id` | Context identifier (when available) |
187+
| `gen_ai.agent.a2a.push_notification.event_kind` | The type of streaming event delivered (e.g. `TaskStatusUpdateEvent`) |
188+
| `gen_ai.agent.a2a.response` | Full event payload (only if extraction enabled) |
189+
| `error.type` | Error message (on delivery failure) |
190+
109191
### Request/Response Extraction
110192

111193
Enable request and response data extraction in spans using JVM system properties:
@@ -124,7 +206,7 @@ Enable request and response data extraction in spans using JVM system properties
124206

125207
### Instrumentation
126208

127-
Adds OpenTelemetry spans to A2A client operations:
209+
Adds OpenTelemetry spans and metrics to A2A client operations:
128210

129211
```xml
130212
<dependency>
@@ -133,6 +215,27 @@ Adds OpenTelemetry spans to A2A client operations:
133215
</dependency>
134216
```
135217

218+
To also enable `gen_ai.client.operation.duration` metrics, pass a `Meter` via the transport config:
219+
220+
```java
221+
config.setParameters(Map.of(
222+
OpenTelemetryClientTransportWrapper.OTEL_TRACER_KEY, openTelemetry.getTracer("my-service"),
223+
OpenTelemetryClientTransportWrapper.OTEL_METER_KEY, openTelemetry.getMeter("my-service")
224+
));
225+
```
226+
227+
| Instrument | Type | Unit | Description |
228+
|------------|------|------|-------------|
229+
| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of each A2A client operation call |
230+
231+
Attributes on `gen_ai.client.operation.duration`:
232+
233+
| Attribute | Value |
234+
|-----------|-------|
235+
| `gen_ai.agent.a2a.operation.name` | A2A method name (e.g. `message/send`) |
236+
| `gen_ai.system` | `a2a` |
237+
| `error.type` | `error` (only on failure) |
238+
136239
### Context Propagation
137240

138241
For trace context propagation across async client boundaries:

‎examples/helloworld/client/README.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -161,11 +161,11 @@ The Java client (`HelloWorldClient.java`) performs the following actions:
161161

162162
## Enable OpenTelemetry (Optional)
163163

164-
The client includes support for distributed tracing with OpenTelemetry. To enable it:
164+
The client includes support for distributed tracing and metrics with OpenTelemetry. To enable it:
165165

166166
### Prerequisites
167167

168-
**IMPORTANT**: The client expects an OpenTelemetry collector to be ready and accepting traces. You have two options:
168+
**IMPORTANT**: The client expects an OpenTelemetry collector to be ready and accepting traces and metrics. You have two options:
169169

170170
#### Option 1: Use the Java Server Example (Recommended)
171171

‎examples/helloworld/client/src/main/java/org/a2aproject/sdk/examples/helloworld/client/HelloWorldClient.java‎

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package org.a2aproject.sdk.examples.helloworld.client;
22

3+
import static org.a2aproject.sdk.extras.opentelemetry.client.OpenTelemetryClientTransportWrapper.OTEL_METER_KEY;
34
import static org.a2aproject.sdk.extras.opentelemetry.client.OpenTelemetryClientTransportWrapper.OTEL_TRACER_KEY;
45
import static org.a2aproject.sdk.extras.opentelemetry.client.propagation.OpenTelemetryClientPropagatorTransportWrapper.OTEL_OPEN_TELEMETRY_KEY;
56

@@ -34,8 +35,11 @@
3435
import io.grpc.ManagedChannelBuilder;
3536
import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
3637
import io.opentelemetry.context.propagation.ContextPropagators;
38+
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
3739
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
3840
import io.opentelemetry.sdk.OpenTelemetrySdk;
41+
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
42+
import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
3943
import io.opentelemetry.sdk.resources.Resource;
4044
import io.opentelemetry.sdk.trace.SdkTracerProvider;
4145
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
@@ -136,20 +140,32 @@ public static void main(String[] args) {
136140
}
137141

138142
static OpenTelemetrySdk initOpenTelemetry() {
143+
Resource resource = Resource.getDefault().toBuilder()
144+
.put("service.version", "1.0")
145+
.put("service.name", "helloworld-client")
146+
.build();
147+
139148
SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
140149
.addSpanProcessor(BatchSpanProcessor.builder(
141150
OtlpGrpcSpanExporter.builder()
142151
.setEndpoint("http://localhost:5317")
143152
.build()
144153
).build())
145-
.setResource(Resource.getDefault().toBuilder()
146-
.put("service.version", "1.0")
147-
.put("service.name", "helloworld-client")
148-
.build())
154+
.setResource(resource)
155+
.build();
156+
157+
SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder()
158+
.registerMetricReader(PeriodicMetricReader.builder(
159+
OtlpGrpcMetricExporter.builder()
160+
.setEndpoint("http://localhost:5317")
161+
.build()
162+
).build())
163+
.setResource(resource)
149164
.build();
150165

151166
return OpenTelemetrySdk.builder()
152167
.setTracerProvider(sdkTracerProvider)
168+
.setMeterProvider(sdkMeterProvider)
153169
.setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
154170
.build();
155171
}
@@ -186,6 +202,7 @@ private static void updateTransportConfig(ClientTransportConfig transportConfig,
186202
if (openTelemetrySdk != null) {
187203
Map<String, Object> parameters = new HashMap<>(transportConfig.getParameters());
188204
parameters.put(OTEL_TRACER_KEY, openTelemetrySdk.getTracer("helloworld-client"));
205+
parameters.put(OTEL_METER_KEY, openTelemetrySdk.getMeter("helloworld-client"));
189206
parameters.put(OTEL_OPEN_TELEMETRY_KEY, openTelemetrySdk);
190207
transportConfig.setParameters(parameters);
191208
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ quarkus.agentcard.protocol=JSONRPC
55

66
# OpenTelemetry configuration
77
quarkus.otel.sdk.disabled=false
8+
quarkus.otel.metrics.enabled=true
89
quarkus.observability.lgtm.grafana-port=3001
910
quarkus.observability.lgtm.otel-grpc-port=5317
1011
quarkus.observability.lgtm.otel-http-port=5318

0 commit comments

Comments
 (0)