feat(opentelemetry): add push notification tracing with CLIENT spans … · a2aproject/a2a-java@ada3581 · GitHub
Skip to content

Commit ada3581

Browse files
ehsavoieehsavoie
authored
feat(opentelemetry): add push notification tracing with CLIENT spans (#1228)
Add OpenTelemetryPushNotificationSenderDecorator that wraps PushNotificationSender to create a CLIENT span for every push delivery. The span carries task_id, context_id, push_notification.event_kind, and optional response payload attributes, with error.type on delivery failure. Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>
1 parent 7a1dfcc commit ada3581

15 files changed

Lines changed: 1062 additions & 10 deletions

File tree

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

Lines changed: 37 additions & 1 deletion

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,4 +20,7 @@ public interface A2AObservabilityNames {
2020
String GENAI_RESPONSE = GENAI_PREFIX + ".response"; // gen_ai.output.messages ?
2121
String GENAI_ROLE = GENAI_PREFIX + ".role";
2222
String GENAI_TASK_ID = GENAI_PREFIX + ".task_id";
23+
24+
// Push notification
25+
String PUSH_NOTIFICATION_EVENT_KIND = GENAI_PREFIX + ".push_notification.event_kind";
2326
}

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

Lines changed: 71 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,5 @@
11
package org.a2aproject.sdk.extras.opentelemetry.it;
22

3-
4-
53
import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE;
64

75
import com.google.gson.Gson;
@@ -12,6 +10,7 @@
1210
import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil;
1311
import org.a2aproject.sdk.spec.Task;
1412
import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent;
13+
import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
1514
import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
1615
import io.opentelemetry.api.trace.Span;
1716
import io.opentelemetry.api.trace.Tracer;
@@ -27,6 +26,9 @@
2726
import jakarta.inject.Inject;
2827
import jakarta.inject.Singleton;
2928
import java.util.List;
29+
import java.util.Map;
30+
import java.util.concurrent.ConcurrentHashMap;
31+
import java.util.concurrent.CopyOnWriteArrayList;
3032
import java.util.stream.Collectors;
3133

3234
/**
@@ -40,6 +42,11 @@ public class A2ATestRoutes {
4042
private static final String TEXT_PLAIN = "text/plain";
4143
private static final Gson gson = new GsonBuilder().create();
4244

45+
/**
46+
* Captured webhook requests, each entry is a map of header name to value.
47+
*/
48+
private final List<Map<String, String>> webhookRequests = new CopyOnWriteArrayList<>();
49+
4350
@Inject
4451
TestUtilsBean testUtilsBean;
4552
@Inject
@@ -76,6 +83,26 @@ void setupRoutes(@Observes Router router) {
7683
.produces(TEXT_PLAIN)
7784
.handler(ctx -> getChildQueueCount(ctx.pathParam("taskId"), ctx));
7885

86+
router.post("/test/pushconfig/:taskId")
87+
.consumes(APPLICATION_JSON)
88+
.handler(BodyHandler.create())
89+
.blockingHandler(ctx -> savePushNotificationConfig(ctx.pathParam("taskId"), ctx.body().asString(), ctx));
90+
91+
router.delete("/test/pushconfig/:taskId/:configId")
92+
.blockingHandler(ctx -> deletePushNotificationConfig(
93+
ctx.pathParam("taskId"), ctx.pathParam("configId"), ctx));
94+
95+
router.post("/test/webhook")
96+
.handler(BodyHandler.create())
97+
.handler(ctx -> captureWebhook(ctx));
98+
99+
router.get("/test/webhook/requests")
100+
.produces(APPLICATION_JSON)
101+
.handler(ctx -> getWebhookRequests(ctx));
102+
103+
router.delete("/test/webhook/requests")
104+
.handler(ctx -> clearWebhookRequests(ctx));
105+
79106
router.get("/hello")
80107
.produces(TEXT_PLAIN)
81108
.handler(ctx -> hello(ctx));
@@ -180,6 +207,48 @@ public void getChildQueueCount(String taskId, RoutingContext rc) {
180207
.end(String.valueOf(count));
181208
}
182209

210+
public void savePushNotificationConfig(String taskId, String body, RoutingContext rc) {
211+
try {
212+
TaskPushNotificationConfig config = JsonUtil.fromJson(body, TaskPushNotificationConfig.class);
213+
testUtilsBean.savePushNotificationConfig(taskId, config);
214+
rc.response()
215+
.setStatusCode(200)
216+
.end();
217+
} catch (Throwable t) {
218+
errorResponse(t, rc);
219+
}
220+
}
221+
222+
public void deletePushNotificationConfig(String taskId, String configId, RoutingContext rc) {
223+
try {
224+
testUtilsBean.deletePushNotificationConfig(taskId, configId);
225+
rc.response()
226+
.setStatusCode(200)
227+
.end();
228+
} catch (Throwable t) {
229+
errorResponse(t, rc);
230+
}
231+
}
232+
233+
private void captureWebhook(RoutingContext rc) {
234+
Map<String, String> headers = new ConcurrentHashMap<>();
235+
rc.request().headers().forEach(entry -> headers.put(entry.getKey(), entry.getValue()));
236+
webhookRequests.add(headers);
237+
rc.response().setStatusCode(200).end();
238+
}
239+
240+
private void getWebhookRequests(RoutingContext rc) {
241+
rc.response()
242+
.setStatusCode(200)
243+
.putHeader(CONTENT_TYPE, APPLICATION_JSON)
244+
.end(gson.toJson(webhookRequests));
245+
}
246+
247+
private void clearWebhookRequests(RoutingContext rc) {
248+
webhookRequests.clear();
249+
rc.response().setStatusCode(200).end();
250+
}
251+
183252
public void hello(RoutingContext rc) {
184253
Span span = tracer.spanBuilder("hello").startSpan();
185254
try (Scope scope = span.makeCurrent()) {

‎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/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestUtilsBean.java‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,17 @@
11
package org.a2aproject.sdk.extras.opentelemetry.it;
22

33
import org.a2aproject.sdk.server.events.QueueManager;
4+
import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore;
45
import org.a2aproject.sdk.server.tasks.TaskStore;
56
import org.a2aproject.sdk.spec.Event;
67
import org.a2aproject.sdk.spec.Task;
8+
import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
79
import jakarta.enterprise.context.ApplicationScoped;
810
import jakarta.inject.Inject;
911

1012
/**
1113
* Test utilities for OpenTelemetry integration tests.
12-
* Allows direct manipulation of tasks and queues for testing.
14+
* Allows direct manipulation of tasks, queues, and push notification configs for testing.
1315
*/
1416
@ApplicationScoped
1517
public class TestUtilsBean {
@@ -20,6 +22,9 @@ public class TestUtilsBean {
2022
@Inject
2123
QueueManager queueManager;
2224

25+
@Inject
26+
PushNotificationConfigStore pushNotificationConfigStore;
27+
2328
public void saveTask(Task task) {
2429
taskStore.save(task, false);
2530
}
@@ -43,4 +48,14 @@ public void enqueueEvent(String taskId, Event event) {
4348
public int getChildQueueCount(String taskId) {
4449
return queueManager.getActiveChildQueueCount(taskId);
4550
}
51+
52+
public void savePushNotificationConfig(String taskId, TaskPushNotificationConfig config) {
53+
pushNotificationConfigStore.setInfo(
54+
TaskPushNotificationConfig.builder(config).taskId(taskId).build(),
55+
null);
56+
}
57+
58+
public void deletePushNotificationConfig(String taskId, String configId) {
59+
pushNotificationConfigStore.deleteInfo(taskId, configId);
60+
}
4661
}

‎extras/opentelemetry/integration-tests/src/main/resources/application.properties‎

Lines changed: 4 additions & 0 deletions

0 commit comments

Comments
 (0)