feat(opentelemetry): add push notification tracing and trace-context propagation by ehsavoie · Pull Request #1229 · a2aproject/a2a-java · GitHub
Skip to content

feat(opentelemetry): add push notification tracing and trace-context propagation #1229

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ehsavoie wants to merge 1 commit into
base: main
Choose a base branch
from
+2,050 −122
Open

feat(opentelemetry): add push notification tracing and trace-context propagation #1229

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions .../test/java/org/a2aproject/sdk/compat03/conversion/AbstractA2ARequestHandlerTest_v0_3.java
View file
Open in desktop
6 changes: 6 additions & 0 deletions docs/content/dev/extra/http-clients.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -266,3 +266,9 @@ public class MyHttpClientProducer {
| 0 | JDK | Always (fallback) |

If the CDI container is active but no `A2AHttpClient` bean is registered, this provider is skipped and the next available provider is used.

### Compatibility with OpenTelemetry

When the [OpenTelemetry server module](opentelemetry.md) is on the classpath alongside a custom `A2AHttpClient` CDI bean, the OpenTelemetry module wraps your bean with trace-context propagation rather than replacing it. Your custom configuration (mTLS, proxy, authentication, timeouts, etc.) is preserved for push notification delivery while `traceparent` and `tracestate` headers are injected automatically.

This works through the `@Instrumented` CDI qualifier (defined in `server-common`): the OpenTelemetry module produces an `@Instrumented A2AHttpClient` that delegates to your unqualified bean. `BasePushNotificationSender` prefers the `@Instrumented` variant when available, so both your custom configuration and the trace-context headers are present on outbound requests. No changes to your producer code are required.
87 changes: 76 additions & 11 deletions docs/content/dev/extra/opentelemetry.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,6 +177,70 @@ The following attributes are automatically added to spans (defined in `A2AObserv
| `gen_ai.agent.a2a.response` | Full response data (only if extraction enabled) |
| `error.type` | Error message (on failures) |

### Push Notification Delivery Tracing

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. The span status reflects the actual delivery outcome: `OK` when all notifications succeed, `ERROR` when one or more deliveries fail (with `error.type` set to `delivery_failed`) or when an exception is thrown.

The module also provides **W3C trace-context propagation** for outbound push notification HTTP requests via the `@Instrumented` CDI qualifier and `OpenTelemetryHttpClientProducer`.

#### How it works

The `OpenTelemetryHttpClientProducer` CDI bean produces an `@Instrumented A2AHttpClient` — a wrapper that injects `traceparent` and `tracestate` headers into every outbound HTTP request. `BasePushNotificationSender` resolves the HTTP client in this order:

1. An `@Instrumented A2AHttpClient` bean (if available — e.g. the one produced by this module)
2. An unqualified `A2AHttpClient` bean (if the application provides one)
3. `A2AHttpClientFactory.create()` as last resort

The `@Instrumented` qualifier (defined in `server-common`) marks HTTP client beans that have been wrapped with observability instrumentation. It ensures the instrumented client does not compete with the application's own `A2AHttpClient` bean in default CDI resolution — the two coexist, and the instrumented variant wraps the application's bean as its delegate.

#### Custom HTTP client compatibility

If your application provides a custom `A2AHttpClient` CDI bean (e.g. for mTLS, proxy, or custom timeouts), the OpenTelemetry wrapper **preserves that configuration** by delegating to your bean rather than replacing it. No additional annotations or configuration are needed — just produce your bean as usual:

```java
@Produces
@ApplicationScoped
public A2AHttpClient myHttpClient() {
// your custom client (mTLS, proxy, auth, etc.)
}
```

The OpenTelemetry module automatically wraps it with trace-context injection.

#### Context-propagating executor

The producer also creates an `@Instrumented Executor` (via `Context.taskWrapping()`) that ensures the OTel trace context survives the async boundary when push notifications are dispatched to multiple recipients in parallel. This allows the receiving webhook to join the distributed trace.

`BasePushNotificationSender` resolves the dispatch executor in this order:

1. An `@Instrumented Executor` bean (if available — e.g. the one produced by this module)
2. A `@Named("a2a.pushNotificationDispatch") Executor` bean (if the application provides one)
3. `ForkJoinPool.commonPool()` as last resort

#### Custom dispatch executor compatibility

If your application provides a custom `@Named("a2a.pushNotificationDispatch") Executor` CDI bean (e.g. for custom thread pool sizing), the OpenTelemetry module **preserves that configuration** by wrapping it with `Context.taskWrapping()` rather than replacing it. Just produce your bean as usual:

```java
@Produces
@ApplicationScoped
@Named("a2a.pushNotificationDispatch")
public Executor myDispatchExecutor() {
// your custom thread pool
}
```

The OpenTelemetry module automatically wraps it with context propagation.

| Attribute | Description |
|-----------|-------------|
| `gen_ai.agent.a2a.operation.name` | `SendPushNotification` |
| `gen_ai.agent.a2a.task_id` | Task identifier |
| `gen_ai.agent.a2a.context_id` | Context identifier (when available) |
| `gen_ai.agent.a2a.push_notification.event_kind` | The type of streaming event delivered (e.g. `TaskStatusUpdateEvent`) |
| `gen_ai.agent.a2a.response` | Full event payload (only if extraction enabled) |
| `error.type` | `delivery_failed` when one or more deliveries fail, or the exception message when an exception is thrown |

### Request/Response Extraction

Enable request and response data extraction in spans using JVM system properties:
Expand All @@ -195,7 +259,7 @@ Enable request and response data extraction in spans using JVM system properties

### Instrumentation

Adds OpenTelemetry spans to A2A client operations:
Adds OpenTelemetry spans and metrics to A2A client operations:

```xml
<dependency>
Expand All @@ -206,11 +270,20 @@ Adds OpenTelemetry spans to A2A client operations:

### Metrics

The client transport records the following histogram:
The client transport wrapper is auto-discovered via ServiceLoader — just adding the dependency enables tracing. To also enable `gen_ai.client.operation.duration` metrics, pass a `Meter` via the transport config:

```java
config.setParameters(Map.of(
OpenTelemetryClientTransportWrapper.OTEL_TRACER_KEY, openTelemetry.getTracer("my-service"),
OpenTelemetryClientTransportWrapper.OTEL_METER_KEY, openTelemetry.getMeter("my-service")
));
```

If no `Meter` is provided, the histogram is silently skipped.

| Instrument | Type | Unit | Description |
|------------|------|------|-------------|
| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of A2A client operations from invocation to completion or error |
| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of each A2A client operation call |

Attributes on `gen_ai.client.operation.duration`:

Expand All @@ -220,14 +293,6 @@ Attributes on `gen_ai.client.operation.duration`:
| `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:
Expand Down
11 changes: 11 additions & 0 deletions ...ent-cdi/src/test/java/org/a2aproject/sdk/client/http/cdi/CdiA2AHttpClientFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,17 @@ public void testCreateDoesNotLogWarningWhenCdiSucceeds() {
assertFalse(records.stream().anyMatch(r -> r.getLevel() == Level.WARNING),
"No warning should be logged when CDI provider succeeds");
}

@Test
public void testCreateExcludingCdiSkipsTheCdiBeanAndFallsBack() {
// Regression guard for the recursive CDI self-delegation bug: a CDI producer that
// produces an A2AHttpClient must call createExcluding("cdi") so the factory does not
// re-enter the container and return the proxy being constructed as its own delegate.
A2AHttpClient client = A2AHttpClientFactory.createExcluding("cdi");
assertNotNull(client);
assertFalse(client instanceof DummyA2AHttpClient,
"createExcluding('cdi') must not return the CDI-registered bean");
}
}

@Nested
Expand Down
4 changes: 4 additions & 0 deletions ...y/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,4 +42,8 @@ public interface A2AObservabilityNames {
String GENAI_STREAMING_DURATION = GENAI_PREFIX + ".streaming.duration";
String GENAI_SYSTEM = "gen_ai.system";
String GENAI_SYSTEM_VALUE = "a2a";

// Push notification
String PUSH_NOTIFICATION_EVENT_KIND = GENAI_PREFIX + ".push_notification.event_kind";
String PUSH_NOTIFICATION_DELIVERY_FAILED = "delivery_failed";
}
87 changes: 85 additions & 2 deletions ...gration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
package org.a2aproject.sdk.extras.opentelemetry.it;



import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE;
import static java.net.HttpURLConnection.HTTP_OK;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
Expand All @@ -12,6 +11,7 @@
import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil;
import org.a2aproject.sdk.spec.Task;
import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent;
import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.trace.Span;
Expand All @@ -32,7 +32,10 @@
import jakarta.enterprise.inject.Produces;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;

/**
* Test routes for OpenTelemetry integration testing.
Expand All @@ -45,6 +48,11 @@ public class A2ATestRoutes {
private static final String TEXT_PLAIN = "text/plain";
private static final Gson gson = new GsonBuilder().create();

/**
* Captured webhook requests, each entry is a map of header name to value.
*/
private final List<Map<String, String>> webhookRequests = new CopyOnWriteArrayList<>();

@Inject
TestUtilsBean testUtilsBean;
@Inject
Expand All @@ -53,6 +61,8 @@ public class A2ATestRoutes {
InMemoryMetricExporter inMemoryMetricExporter;
@Inject
OpenTelemetry openTelemetry;
@Inject
CustomDispatchExecutorProducer customDispatchExecutorProducer;

@Inject
Tracer tracer;
Expand Down Expand Up @@ -85,6 +95,37 @@ void setupRoutes(@Observes Router router) {
.produces(TEXT_PLAIN)
.handler(ctx -> getChildQueueCount(ctx.pathParam("taskId"), ctx));

router.post("/test/pushconfig/:taskId")
.consumes(APPLICATION_JSON)
.handler(BodyHandler.create())
.blockingHandler(ctx -> savePushNotificationConfig(ctx.pathParam("taskId"), ctx.body().asString(), ctx));

router.delete("/test/pushconfig/:taskId/:configId")
.blockingHandler(ctx -> deletePushNotificationConfig(
ctx.pathParam("taskId"), ctx.pathParam("configId"), ctx));

router.post("/test/webhook")
.handler(BodyHandler.create())
.handler(ctx -> captureWebhook(ctx));

router.get("/test/webhook/requests")
.produces(APPLICATION_JSON)
.handler(ctx -> getWebhookRequests(ctx));

router.delete("/test/webhook/requests")
.handler(ctx -> clearWebhookRequests(ctx));

router.get("/test/dispatch-executor/count")
.produces(TEXT_PLAIN)
.handler(ctx -> ctx.response().setStatusCode(HTTP_OK)
.end(String.valueOf(customDispatchExecutorProducer.getExecutionCount())));

router.delete("/test/dispatch-executor/count")
.handler(ctx -> {
customDispatchExecutorProducer.resetExecutionCount();
ctx.response().setStatusCode(HTTP_OK).end();
});

router.get("/hello")
.produces(TEXT_PLAIN)
.handler(ctx -> hello(ctx));
Expand Down Expand Up @@ -197,6 +238,48 @@ public void getChildQueueCount(String taskId, RoutingContext rc) {
.end(String.valueOf(count));
}

public void savePushNotificationConfig(String taskId, String body, RoutingContext rc) {
try {
TaskPushNotificationConfig config = JsonUtil.fromJson(body, TaskPushNotificationConfig.class);
testUtilsBean.savePushNotificationConfig(taskId, config);
rc.response()
.setStatusCode(200)
.end();
} catch (Throwable t) {
errorResponse(t, rc);
}
}

public void deletePushNotificationConfig(String taskId, String configId, RoutingContext rc) {
try {
testUtilsBean.deletePushNotificationConfig(taskId, configId);
rc.response()
.setStatusCode(200)
.end();
} catch (Throwable t) {
errorResponse(t, rc);
}
}

private void captureWebhook(RoutingContext rc) {
Map<String, String> headers = new HashMap<>();
rc.request().headers().forEach(entry -> headers.put(entry.getKey(), entry.getValue()));
webhookRequests.add(headers);
rc.response().setStatusCode(HTTP_OK).end();
}

private void getWebhookRequests(RoutingContext rc) {
rc.response()
.setStatusCode(HTTP_OK)
.putHeader(CONTENT_TYPE, APPLICATION_JSON)
.end(gson.toJson(webhookRequests));
}

private void clearWebhookRequests(RoutingContext rc) {
webhookRequests.clear();
rc.response().setStatusCode(HTTP_OK).end();
}

public void hello(RoutingContext rc) {
Span span = tracer.spanBuilder("hello").startSpan();
try (Scope scope = span.makeCurrent()) {
Expand Down
70 changes: 70 additions & 0 deletions .../main/java/org/a2aproject/sdk/extras/opentelemetry/it/CustomDispatchExecutorProducer.java
View file
Open in desktop
Loading
Loading