Skip to content

Commit 0126356

Browse files
ehsavoieehsavoie
committed
feat(opentelemetry): trace push notification delivery
Adds OpenTelemetryPushNotificationSenderDecorator, a CDI decorator that creates a CLIENT span (SendPushNotification) for each push notification delivery, with task_id, context_id, and event_kind attributes. - Unit tests: 6 cases in OpenTelemetryPushNotificationSenderDecoratorTest - Integration test: end-to-end delivery span verified via InMemorySpanExporter - Docs: push notification delivery tracing section in opentelemetry.md Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>
1 parent fecb409 commit 0126356

8 files changed

Lines changed: 440 additions & 1 deletion

File tree

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -175,6 +175,19 @@ The following attributes are automatically added to spans (defined in `A2AObserv
175175
| `gen_ai.agent.a2a.response` | Full response data (only if extraction enabled) |
176176
| `error.type` | Error message (on failures) |
177177

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+
178191
### Request/Response Extraction
179192

180193
Enable request and response data extraction in spans using JVM system properties:

‎extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,10 @@ public interface A2AObservabilityNames {
2121
String GENAI_ROLE = GENAI_PREFIX + ".role";
2222
String GENAI_TASK_ID = GENAI_PREFIX + ".task_id";
2323

24+
// Push notification
25+
String PUSH_NOTIFICATION_EVENT_KIND = GENAI_PREFIX + ".push_notification.event_kind";
26+
String PUSH_NOTIFICATION_URL = GENAI_PREFIX + ".push_notification.url";
27+
2428
// Metrics
2529
String GENAI_CLIENT_OPERATION_DURATION = "gen_ai.client.operation.duration";
2630
String GENAI_STREAMING_DURATION = GENAI_PREFIX + ".streaming.duration";

‎extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import jakarta.inject.Inject;
3434
import jakarta.inject.Singleton;
3535
import java.util.List;
36+
import java.util.concurrent.atomic.AtomicInteger;
3637
import java.util.stream.Collectors;
3738

3839
/**
@@ -46,6 +47,8 @@ public class A2ATestRoutes {
4647
private static final String TEXT_PLAIN = "text/plain";
4748
private static final Gson gson = new GsonBuilder().create();
4849

50+
private final AtomicInteger webhookCallCount = new AtomicInteger(0);
51+
4952
@Inject
5053
TestUtilsBean testUtilsBean;
5154
@Inject
@@ -105,6 +108,14 @@ void setupRoutes(@Observes Router router) {
105108
router.get("/reset-metrics")
106109
.produces(TEXT_PLAIN)
107110
.handler(ctx -> resetMetrics(ctx));
111+
112+
router.post("/test/webhook")
113+
.handler(BodyHandler.create())
114+
.handler(ctx -> receiveWebhook(ctx));
115+
116+
router.get("/test/webhook/count")
117+
.produces(TEXT_PLAIN)
118+
.handler(ctx -> getWebhookCount(ctx));
108119
}
109120

110121
public void saveTask(String body, RoutingContext rc) {
@@ -309,8 +320,21 @@ public void resetMetrics(RoutingContext rc) {
309320
rc.response().setStatusCode(200).end();
310321
}
311322

323+
public void receiveWebhook(RoutingContext rc) {
324+
webhookCallCount.incrementAndGet();
325+
rc.response().setStatusCode(200).end();
326+
}
327+
328+
public void getWebhookCount(RoutingContext rc) {
329+
rc.response()
330+
.setStatusCode(200)
331+
.putHeader(CONTENT_TYPE, TEXT_PLAIN)
332+
.end(String.valueOf(webhookCallCount.get()));
333+
}
334+
312335
public void reset(RoutingContext rc) {
313336
inMemorySpanExporter.reset();
337+
webhookCallCount.set(0);
314338
rc.response().setStatusCode(200).end();
315339
}
316340

‎extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestAgentCardProducer.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ public AgentCard agentCard() {
3131
.documentationUrl("http://example.com/test")
3232
.capabilities(AgentCapabilities.builder()
3333
.streaming(true)
34-
.pushNotifications(false)
34+
.pushNotifications(true)
3535
.build())
3636
.defaultInputModes(Collections.singletonList("text"))
3737
.defaultOutputModes(Collections.singletonList("text"))

‎extras/opentelemetry/integration-tests/src/test/java/org/a2aproject/sdk/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -320,6 +320,59 @@ void testSendMessageStreamRecordsStreamingDuration() throws Exception {
320320
"Histogram should have at least one data point");
321321
}
322322

323+
@Test
324+
void testPushNotificationDeliveryCreatesSpan() throws Exception {
325+
String taskId = "push-test-task-1";
326+
String contextId = "push-test-ctx-1";
327+
328+
Task task = Task.builder()
329+
.id(taskId)
330+
.contextId(contextId)
331+
.status(new TaskStatus(TaskState.TASK_STATE_WORKING))
332+
.history(Collections.emptyList())
333+
.artifacts(Collections.emptyList())
334+
.build();
335+
saveTaskInTaskStore(task);
336+
ensureQueueForTask(taskId);
337+
reset();
338+
339+
try {
340+
TaskPushNotificationConfig config = TaskPushNotificationConfig.builder()
341+
.id("push-test-config-1")
342+
.taskId(taskId)
343+
.url("http://localhost:" + serverPort + "/test/webhook")
344+
.build();
345+
TaskPushNotificationConfig saved = client.createTaskPushNotificationConfiguration(config, null);
346+
assertNotNull(saved);
347+
348+
reset();
349+
350+
TaskStatusUpdateEvent event = new TaskStatusUpdateEvent(
351+
taskId, new TaskStatus(TaskState.TASK_STATE_COMPLETED), contextId, null);
352+
enqueueTaskStatusUpdateEventViaHttp(taskId, event);
353+
354+
await().atMost(15, SECONDS).until(() -> {
355+
List<Map<String, Object>> spans = getSpans();
356+
return spans.stream().anyMatch(s -> "SendPushNotification".equals(s.get("name")));
357+
});
358+
359+
List<Map<String, Object>> spans = getSpans();
360+
Map<String, Object> pushSpan = spans.stream()
361+
.filter(s -> "SendPushNotification".equals(s.get("name")))
362+
.findFirst()
363+
.orElseThrow(() -> new AssertionError("No SendPushNotification span found"));
364+
365+
assertEquals(Boolean.TRUE, pushSpan.get("ended"), "Push notification span should be ended");
366+
assertEquals("CLIENT", pushSpan.get("kind"), "Push notification span should be CLIENT");
367+
assertEquals(taskId, pushSpan.get("attr_gen_ai.agent.a2a.task_id"),
368+
"Task ID should be set on push notification span");
369+
assertNotNull(pushSpan.get("attr_gen_ai.agent.a2a.push_notification.event_kind"),
370+
"Event kind should be set on push notification span");
371+
} finally {
372+
deleteTaskInTaskStore(taskId);
373+
}
374+
}
375+
323376
protected void saveTaskInTaskStore(Task task) throws Exception {
324377
HttpClient httpClient = HttpClient.newBuilder()
325378
.version(HttpClient.Version.HTTP_2)
@@ -382,4 +435,19 @@ protected void ensureQueueForTask(String taskId) throws Exception {
382435
throw new RuntimeException(String.format("Ensuring queue failed! Status: %d, Body: %s", response.statusCode(), response.body()));
383436
}
384437
}
438+
439+
protected void enqueueTaskStatusUpdateEventViaHttp(String taskId, TaskStatusUpdateEvent event) throws Exception {
440+
HttpClient httpClient = HttpClient.newBuilder()
441+
.version(HttpClient.Version.HTTP_2)
442+
.build();
443+
HttpRequest request = HttpRequest.newBuilder()
444+
.uri(URI.create("http://localhost:" + serverPort + "/test/queue/enqueueTaskStatusUpdateEvent/" + taskId))
445+
.POST(HttpRequest.BodyPublishers.ofString(JsonUtil.toJson(event)))
446+
.header("Content-Type", "application/json")
447+
.build();
448+
HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
449+
if (response.statusCode() != 200) {
450+
throw new RuntimeException(String.format("Enqueue event failed! Status: %d, Body: %s", response.statusCode(), response.body()));
451+
}
452+
}
385453
}

‎extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecorator.java‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
package org.a2aproject.sdk.extras.opentelemetry.server;
2+
3+
import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE;
4+
import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY;
5+
import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CONTEXT_ID;
6+
import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME;
7+
import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE;
8+
import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID;
9+
import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.PUSH_NOTIFICATION_EVENT_KIND;
10+
11+
import io.opentelemetry.api.trace.Span;
12+
import io.opentelemetry.api.trace.SpanKind;
13+
import io.opentelemetry.api.trace.StatusCode;
14+
import io.opentelemetry.api.trace.Tracer;
15+
import io.opentelemetry.context.Scope;
16+
import jakarta.annotation.Priority;
17+
import jakarta.decorator.Decorator;
18+
import jakarta.decorator.Delegate;
19+
import jakarta.enterprise.inject.Any;
20+
import jakarta.inject.Inject;
21+
import org.a2aproject.sdk.server.tasks.PushNotificationSender;
22+
import org.a2aproject.sdk.spec.Message;
23+
import org.a2aproject.sdk.spec.StreamingEventKind;
24+
import org.a2aproject.sdk.spec.Task;
25+
import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent;
26+
import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
27+
import org.jspecify.annotations.Nullable;
28+
29+
/**
30+
* OpenTelemetry CDI Decorator for {@link PushNotificationSender}.
31+
* <p>
32+
* Creates a {@code CLIENT} span named {@code SendPushNotification} for every push
33+
* notification delivery attempt, capturing the task ID, context ID, and event kind
34+
* as span attributes.
35+
* <p>
36+
* To enable this decorator, add it to your beans.xml:
37+
* <pre>{@code
38+
* <decorators>
39+
* <class>org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryPushNotificationSenderDecorator</class>
40+
* </decorators>
41+
* }</pre>
42+
*/
43+
@Decorator
44+
@Priority(100)
45+
public abstract class OpenTelemetryPushNotificationSenderDecorator implements PushNotificationSender {
46+
47+
static final String SEND_PUSH_NOTIFICATION = "SendPushNotification";
48+
49+
@Inject
50+
@Delegate
51+
@Any
52+
private PushNotificationSender delegate;
53+
54+
@Inject
55+
private Tracer tracer;
56+
57+
/**
58+
* Default constructor for CDI.
59+
*/
60+
public OpenTelemetryPushNotificationSenderDecorator() {
61+
}
62+
63+
/**
64+
* Constructor for unit testing.
65+
*
66+
* @param delegate the delegate push notification sender
67+
* @param tracer the tracer to use
68+
*/
69+
OpenTelemetryPushNotificationSenderDecorator(PushNotificationSender delegate, Tracer tracer) {
70+
this.delegate = delegate;
71+
this.tracer = tracer;
72+
}
73+
74+
@Override
75+
public void sendNotification(StreamingEventKind event, @Nullable Task taskSnapshot) {
76+
var spanBuilder = tracer.spanBuilder(SEND_PUSH_NOTIFICATION)
77+
.setSpanKind(SpanKind.CLIENT)
78+
.setAttribute(GENAI_OPERATION_NAME, SEND_PUSH_NOTIFICATION)
79+
.setAttribute(PUSH_NOTIFICATION_EVENT_KIND, event.getClass().getSimpleName());
80+
81+
String taskId = extractTaskId(event, taskSnapshot);
82+
if (taskId != null) {
83+
spanBuilder.setAttribute(GENAI_TASK_ID, taskId);
84+
}
85+
String contextId = extractContextId(event, taskSnapshot);
86+
if (contextId != null) {
87+
spanBuilder.setAttribute(GENAI_CONTEXT_ID, contextId);
88+
}
89+
if (extractResponse()) {
90+
spanBuilder.setAttribute(GENAI_RESPONSE, event.toString());
91+
}
92+
93+
Span span = spanBuilder.startSpan();
94+
try (Scope scope = span.makeCurrent()) {
95+
delegate.sendNotification(event, taskSnapshot);
96+
span.setStatus(StatusCode.OK);
97+
} catch (Exception ex) {
98+
span.setAttribute(ERROR_TYPE, ex.getMessage() != null ? ex.getMessage() : ex.getClass().getName());
99+
span.setStatus(StatusCode.ERROR, ex.getMessage());
100+
throw ex;
101+
} finally {
102+
span.end();
103+
}
104+
}
105+
106+
@Nullable
107+
private static String extractTaskId(StreamingEventKind event, @Nullable Task taskSnapshot) {
108+
if (taskSnapshot != null) {
109+
return taskSnapshot.id();
110+
}
111+
if (event instanceof Task task) {
112+
return task.id();
113+
}
114+
if (event instanceof TaskStatusUpdateEvent statusUpdate) {
115+
return statusUpdate.taskId();
116+
}
117+
if (event instanceof TaskArtifactUpdateEvent artifactUpdate) {
118+
return artifactUpdate.taskId();
119+
}
120+
return null;
121+
}
122+
123+
@Nullable
124+
private static String extractContextId(StreamingEventKind event, @Nullable Task taskSnapshot) {
125+
if (taskSnapshot != null) {
126+
return taskSnapshot.contextId();
127+
}
128+
if (event instanceof Task task) {
129+
return task.contextId();
130+
}
131+
if (event instanceof TaskStatusUpdateEvent statusUpdate) {
132+
return statusUpdate.contextId();
133+
}
134+
if (event instanceof Message message) {
135+
return message.contextId();
136+
}
137+
return null;
138+
}
139+
140+
private boolean extractResponse() {
141+
return Boolean.getBoolean(EXTRACT_RESPONSE_SYS_PROPERTY);
142+
}
143+
}

‎extras/opentelemetry/server/src/main/resources/META-INF/beans.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,5 +5,6 @@
55
bean-discovery-mode="annotated">
66
<decorators>
77
<class>org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryRequestHandlerDecorator</class>
8+
<class>org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryPushNotificationSenderDecorator</class>
89
</decorators>
910
</beans>

0 commit comments

Comments
 (0)