This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 4e8f7dcc4 [integration][observability] Add OpenTelemetry GenAI
exporter for Agent Traces (Phase 1) (#1160)
4e8f7dcc4 is described below
commit 4e8f7dcc4a2bd153f33f0317a212ead368c0b263
Author: Edson <[email protected]>
AuthorDate: Mon Oct 26 05:18:06 2026 -0400
[integration][observability] Add OpenTelemetry GenAI exporter for Agent
Traces (Phase 1) (#1160)
Generated-by: Claude Code 2.1.259 (Claude Fable 5, Claude Opus 5.5)
---
docs/content/docs/operations/monitoring.md | 57 ++
integrations/observability/otel/pom.xml | 79 +++
.../observability/otel/AgentTraceSpanData.java | 163 +++++
.../observability/otel/AgentTraceSpans.java | 548 ++++++++++++++
.../observability/otel/ConverterDiagnostic.java | 89 +++
.../integrations/observability/otel/OTelIds.java | 83 +++
.../observability/otel/TraceLogOTelExporter.java | 447 ++++++++++++
.../observability/otel/AgentTraceSpansTest.java | 783 +++++++++++++++++++++
.../otel/TraceLogOTelExporterTest.java | 413 +++++++++++
integrations/{ => observability}/pom.xml | 23 +-
integrations/pom.xml | 2 +
11 files changed, 2670 insertions(+), 17 deletions(-)
diff --git a/docs/content/docs/operations/monitoring.md
b/docs/content/docs/operations/monitoring.md
index 75f20ba7d..812c49e38 100644
--- a/docs/content/docs/operations/monitoring.md
+++ b/docs/content/docs/operations/monitoring.md
@@ -507,3 +507,60 @@ The JSON result exposes the same graph:
| `nodes[id].upstreamEdges` | Distinct recorded upstream Event and Action
references. Consult `warnings` for missing or invalid relationships. |
| `nodes[id].actions` | Actions linked from this Event; each has a `name` and
a `children` list of output Event IDs. |
| `warnings` | Problems encountered while reading records or reconstructing
relationships, such as malformed records, conflicting Event identities, missing
upstream Events, edges without an Action name, and cycles. Cycle edges are
removed before rendering. |
+
+### Exporting to OpenTelemetry
+
+The `flink-agents-integrations-observability-otel` module converts Trace Log
records into [OpenTelemetry](https://opentelemetry.io) traces that follow the
[GenAI semantic
conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/), so agent
runs can be viewed in any OTLP-compatible backend, such as Jaeger or Grafana
Tempo.
+
+The exporter runs **out of band**: it reads Trace Log files and sends spans
over OTLP, so it has no effect on the running job and also works on logs from
finished runs. It needs execution records, which the default `EVENT_ONLY`
target does not write: use `scope: ALL` as in [Recording
Targets](#recording-targets), or add a target per type, such as `scope:
{action: "*"}` and `scope: {llm: "*"}`, for `action`, `llm`, `tool`, `parser`,
and `subagent`.
+
+#### Mapping
+
+| Trace Log | OpenTelemetry |
+|---|---|
+| Input run (`inputRunId`) | Trace, with a synthesized `invoke_agent` root
span |
+| Execution (`executionId` / `parentExecutionId`) | Span / parent span. A
child whose parent has no records, for example when targets keep Tools but not
Actions, is attached to the root span and keeps the parent id in
`flink_agents.parent_execution_id` |
+| `llm` execution | `chat {model}` span, kind CLIENT:
`gen_ai.operation.name=chat`, `gen_ai.request.model` from
`entityMetadata.model` (span name `chat` when absent), and
`gen_ai.usage.input_tokens` / `gen_ai.usage.output_tokens` from the success
record's `entityMetadata.usage`. `gen_ai.provider.name` is not set: the record
does not carry the provider |
+| `tool` execution | `execute_tool {tool}` span, kind INTERNAL:
`gen_ai.operation.name=execute_tool`, `gen_ai.tool.name`, `gen_ai.tool.call.id`
from `entityMetadata.toolCallId`, `gen_ai.tool.type` (`function` for function
tools, `extension` for remote-function and MCP tools; the raw value is always
in `flink_agents.tool.type`), and `gen_ai.agent.name` |
+| `subagent` execution | `invoke_agent {name}` span, kind INTERNAL:
`gen_ai.operation.name=invoke_agent`, `gen_ai.agent.name` from the sub-agent
name, and `gen_ai.tool.call.id` from `entityMetadata.toolCallId` |
+| `action` execution | `action {name}` span, kind INTERNAL |
+| `parser` execution | `parse {name}` span, kind INTERNAL,
`gen_ai.operation.name=parse` (a custom low-cardinality value) |
+| `businessKey` | `flink_agents.business_key`. It is the keyed-stream key,
which is a conversation only in chat-style pipelines, so it is reported as
`gen_ai.conversation.id` only when enabled
(`--business-key-as-conversation-id`, or
`setBusinessKeyAsConversationId(true)`) |
+| Failed execution | Span status `ERROR` with `attributes.errorMessage`;
`error.type` from `attributes.errorType`, else `problemCategory` |
+
+Event records are read but do not become spans. Trace and span ids are derived
deterministically from `inputRunId` and `executionId`, so exporting the same
log twice is idempotent for deduplicating backends; delivery itself is
at-least-once. The framework ids are always attached as `flink_agents.*`
attributes for correlation with the raw Trace Log. The GenAI semantic
conventions are still in development, so the exported attribute set is pinned
per Flink Agents release.
+
+**Input.** File arguments are read as streams of JSON records with
`schemaVersion: 1`, whatever their names; the `level` field and the SLF4J
`jobId`, `taskName`, and `subtaskId` fields are ignored. SLF4J output must come
from an appender that writes plain JSON, such as the automatic
`{log.file}.trace-log.log`. A directory argument is expanded to the
`traces-*.log` and `*.trace-log.log` files it contains, as in the Trace Tree
reader. Reading every subtask's file is the caller's responsibi [...]
+
+**Incomplete executions.** An execution with a start record but no terminal
record is still exported, as a zero-duration span with status `UNSET` and the
`flink_agents.execution.incomplete` attribute, and the summary reports an
`INCOMPLETE_EXECUTION` diagnostic (`MISSING_START` for the reverse case). A
crash, a dropped best-effort write, and recovery cannot be told apart
afterwards. A `reused` execution is a single record by design and becomes a
zero-duration span with `flink_agents.exec [...]
+
+**Delivery.** Spans are sent in OTLP requests of at most 512 spans
(`--batch-size`, or `setMaxExportBatchSize`), since each request must fit the
receiver's size limit (4 MiB by default for gRPC), and each request may take up
to 30 seconds. If a request fails or times out, the export stops with an
`ExportFailedException` reporting how many spans the earlier requests
delivered; running it again is safe because the span ids are deterministic.
+
+#### Usage
+
+Command line:
+
+```bash
+java -cp <otel module jar, its runtime dependencies, and slf4j-api> \
+ org.apache.flink.agents.integrations.observability.otel.TraceLogOTelExporter
\
+ --endpoint http://localhost:4317 --protocol grpc --service-name my-agent-job
\
+ [--batch-size 512] [--business-key-as-conversation-id] \
+ /tmp/flink-agent-logs
+```
+
+The module declares `slf4j-api` as `provided`, since Flink supplies it inside
a job, so it is not on Maven's runtime classpath; without it the command fails
with `NoClassDefFoundError: org/slf4j/LoggerFactory`. Add `slf4j-api`, and an
SLF4J binding if you want the exporter's log output.
+
+Programmatic:
+
+```java
+try (TraceLogOTelExporter exporter =
+ TraceLogOTelExporter.builder()
+ .setEndpoint("http://localhost:4317";)
+ .setProtocol("grpc")
+ .setServiceName("my-agent-job")
+ .build()) {
+ exporter.exportFiles(List.of(Path.of("/tmp/flink-agent-logs")));
+}
+```
+
+The exporter currently produces spans only; capturing prompt and response
content is follow-up work.
diff --git a/integrations/observability/otel/pom.xml
b/integrations/observability/otel/pom.xml
new file mode 100644
index 000000000..325fd99e5
--- /dev/null
+++ b/integrations/observability/otel/pom.xml
@@ -0,0 +1,79 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-agents-integrations-observability</artifactId>
+ <version>1.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>flink-agents-integrations-observability-otel</artifactId>
+ <name>Flink Agents : Integrations: Observability: OpenTelemetry</name>
+ <packaging>jar</packaging>
+
+ <dependencyManagement>
+ <dependencies>
+ <!-- Aligns every OpenTelemetry artifact, including transitive
ones, on one version. -->
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-bom</artifactId>
+ <version>${opentelemetry.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-agents-api</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk-trace</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-exporter-otlp</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>${slf4j.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git
a/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpanData.java
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpanData.java
new file mode 100644
index 000000000..749eea635
--- /dev/null
+++
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpanData.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.integrations.observability.otel;
+
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.data.EventData;
+import io.opentelemetry.sdk.trace.data.LinkData;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.data.StatusData;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Immutable {@link SpanData} assembled from Trace Log records.
+ *
+ * <p>Spans are built directly as {@link SpanData} (rather than through a
{@code Tracer}) because
+ * the exporter must control the trace and span ids: they are derived
deterministically from the
+ * framework's {@code inputRunId} / {@code executionId} so that re-exports are
idempotent, and the
+ * SDK tracer does not allow supplying explicit ids.
+ */
+final class AgentTraceSpanData implements SpanData {
+
+ private final String name;
+ private final SpanKind kind;
+ private final SpanContext spanContext;
+ private final SpanContext parentSpanContext;
+ private final StatusData status;
+ private final long startEpochNanos;
+ private final long endEpochNanos;
+ private final Attributes attributes;
+ private final Resource resource;
+ private final InstrumentationScopeInfo scope;
+
+ AgentTraceSpanData(
+ String name,
+ SpanKind kind,
+ SpanContext spanContext,
+ SpanContext parentSpanContext,
+ StatusData status,
+ long startEpochNanos,
+ long endEpochNanos,
+ Attributes attributes,
+ Resource resource,
+ InstrumentationScopeInfo scope) {
+ this.name = name;
+ this.kind = kind;
+ this.spanContext = spanContext;
+ this.parentSpanContext = parentSpanContext;
+ this.status = status;
+ this.startEpochNanos = startEpochNanos;
+ this.endEpochNanos = endEpochNanos;
+ this.attributes = attributes;
+ this.resource = resource;
+ this.scope = scope;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public SpanKind getKind() {
+ return kind;
+ }
+
+ @Override
+ public SpanContext getSpanContext() {
+ return spanContext;
+ }
+
+ @Override
+ public SpanContext getParentSpanContext() {
+ return parentSpanContext;
+ }
+
+ @Override
+ public StatusData getStatus() {
+ return status;
+ }
+
+ @Override
+ public long getStartEpochNanos() {
+ return startEpochNanos;
+ }
+
+ @Override
+ public Attributes getAttributes() {
+ return attributes;
+ }
+
+ @Override
+ public List<EventData> getEvents() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List<LinkData> getLinks() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public long getEndEpochNanos() {
+ return endEpochNanos;
+ }
+
+ @Override
+ public boolean hasEnded() {
+ return true;
+ }
+
+ @Override
+ public int getTotalRecordedEvents() {
+ return 0;
+ }
+
+ @Override
+ public int getTotalRecordedLinks() {
+ return 0;
+ }
+
+ @Override
+ public int getTotalAttributeCount() {
+ return attributes.size();
+ }
+
+ @Override
+ public InstrumentationScopeInfo getInstrumentationScopeInfo() {
+ return scope;
+ }
+
+ @Override
+ @SuppressWarnings("deprecation") // SpanData still requires the deprecated
accessor.
+ public io.opentelemetry.sdk.common.InstrumentationLibraryInfo
getInstrumentationLibraryInfo() {
+ return io.opentelemetry.sdk.common.InstrumentationLibraryInfo.create(
+ scope.getName(), scope.getVersion());
+ }
+
+ @Override
+ public Resource getResource() {
+ return resource;
+ }
+}
diff --git
a/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java
new file mode 100644
index 000000000..0edd8bede
--- /dev/null
+++
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java
@@ -0,0 +1,548 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.integrations.observability.otel;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.TraceFlags;
+import io.opentelemetry.api.trace.TraceState;
+import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.data.StatusData;
+import org.apache.flink.agents.api.trace.ExecutionReporter;
+import org.apache.flink.agents.api.trace.LLMExecutionMetadataKeys;
+import org.apache.flink.agents.api.trace.ToolExecutionMetadataKeys;
+import org.apache.flink.agents.api.trace.TraceContext;
+import org.apache.flink.agents.api.trace.TraceRecord;
+
+import java.time.Instant;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Assembles OpenTelemetry spans from Trace Log records, following the mapping
agreed in the Agent
+ * Trace design discussions:
+ *
+ * <ul>
+ * <li>one input run = one Trace, with a synthesized {@code invoke_agent}
root span;
+ * <li>each execution ({@code action} / {@code llm} / {@code parser} /
{@code tool}) = one span,
+ * parented via {@code parentExecutionId}, or the run root when there is
none or the parent
+ * has no records;
+ * <li>Event records are not spans; they are skipped;
+ * <li>span and trace ids are derived deterministically from the framework
ids, so re-exports are
+ * idempotent (see {@link OTelIds}).
+ * </ul>
+ *
+ * <p>Attributes follow the OpenTelemetry GenAI semantic conventions
(development stability; the
+ * targeted convention set is documented per attribute below). The
framework-native ids are always
+ * attached under {@code flink_agents.*} so backends can correlate spans with
the raw Trace Log
+ * regardless of semantic-convention evolution.
+ */
+public final class AgentTraceSpans {
+
+ // OpenTelemetry GenAI semantic convention attributes (development
stability). Keys are pinned
+ // as literals on purpose: the gen_ai conventions are still evolving, and
pinning makes the
+ // exported schema explicit and stable per flink-agents release.
+ static final AttributeKey<String> GEN_AI_OPERATION_NAME =
+ AttributeKey.stringKey("gen_ai.operation.name");
+ static final AttributeKey<String> GEN_AI_AGENT_NAME =
+ AttributeKey.stringKey("gen_ai.agent.name");
+ static final AttributeKey<String> GEN_AI_TOOL_NAME =
AttributeKey.stringKey("gen_ai.tool.name");
+ static final AttributeKey<String> GEN_AI_TOOL_CALL_ID =
+ AttributeKey.stringKey("gen_ai.tool.call.id");
+ static final AttributeKey<String> GEN_AI_TOOL_TYPE =
AttributeKey.stringKey("gen_ai.tool.type");
+ static final AttributeKey<String> GEN_AI_REQUEST_MODEL =
+ AttributeKey.stringKey("gen_ai.request.model");
+ static final AttributeKey<String> GEN_AI_CONVERSATION_ID =
+ AttributeKey.stringKey("gen_ai.conversation.id");
+ static final AttributeKey<Long> GEN_AI_USAGE_INPUT_TOKENS =
+ AttributeKey.longKey("gen_ai.usage.input_tokens");
+ static final AttributeKey<Long> GEN_AI_USAGE_OUTPUT_TOKENS =
+ AttributeKey.longKey("gen_ai.usage.output_tokens");
+ static final AttributeKey<String> ERROR_TYPE =
AttributeKey.stringKey("error.type");
+
+ // Framework-native correlation attributes.
+ static final AttributeKey<String> FA_INPUT_RUN_ID =
+ AttributeKey.stringKey("flink_agents.input_run_id");
+ static final AttributeKey<String> FA_EXECUTION_ID =
+ AttributeKey.stringKey("flink_agents.execution_id");
+ static final AttributeKey<String> FA_PARENT_EXECUTION_ID =
+ AttributeKey.stringKey("flink_agents.parent_execution_id");
+ static final AttributeKey<String> FA_ENTITY_TYPE =
+ AttributeKey.stringKey("flink_agents.entity_type");
+ static final AttributeKey<String> FA_ENTITY_NAME =
+ AttributeKey.stringKey("flink_agents.entity_name");
+ static final AttributeKey<String> FA_EXECUTION_STATUS =
+ AttributeKey.stringKey("flink_agents.execution.status");
+ static final AttributeKey<Boolean> FA_EXECUTION_INCOMPLETE =
+ AttributeKey.booleanKey("flink_agents.execution.incomplete");
+ static final AttributeKey<String> FA_TOOL_TYPE =
+ AttributeKey.stringKey("flink_agents.tool.type");
+ static final AttributeKey<String> FA_BUSINESS_KEY =
+ AttributeKey.stringKey("flink_agents.business_key");
+
+ static final String INSTRUMENTATION_SCOPE_NAME =
"org.apache.flink.agents.otel";
+
+ private final Resource resource;
+ private final InstrumentationScopeInfo scope;
+ private final boolean businessKeyAsConversationId;
+
+ public AgentTraceSpans(String serviceName) {
+ this(serviceName, false);
+ }
+
+ /**
+ * Creates an assembler.
+ *
+ * @param serviceName the {@code service.name} resource attribute
+ * @param businessKeyAsConversationId whether to also report {@code
businessKey} as {@code
+ * gen_ai.conversation.id}. The business key is the keyed-stream key,
which is a
+ * conversation only in chat-style pipelines, so this is off by
default; the key is always
+ * attached as {@code flink_agents.business_key}.
+ */
+ public AgentTraceSpans(String serviceName, boolean
businessKeyAsConversationId) {
+ this.businessKeyAsConversationId = businessKeyAsConversationId;
+ this.resource =
+ Resource.getDefault().toBuilder()
+ .put(AttributeKey.stringKey("service.name"),
serviceName)
+ .build();
+ this.scope =
InstrumentationScopeInfo.create(INSTRUMENTATION_SCOPE_NAME);
+ }
+
+ /** Assembles spans from Trace Log records; ordering of the input records
does not matter. */
+ public List<SpanData> assemble(List<TraceRecord> records) {
+ return assemble(records, new ArrayList<>());
+ }
+
+ /**
+ * Assembles spans and appends machine-readable {@link
ConverterDiagnostic}s (incomplete
+ * executions, terminal records without a start, parents without records,
records with an
+ * unparseable timestamp) to the given collector.
+ */
+ public List<SpanData> assemble(
+ List<TraceRecord> records, List<ConverterDiagnostic> diagnostics) {
+ // executionId -> collected lifecycle records; LinkedHashMap keeps
output ordering stable.
+ Map<String, ExecutionSpanBuilder> executions = new LinkedHashMap<>();
+ Map<String, RunAccumulator> runs = new LinkedHashMap<>();
+
+ for (TraceRecord record : records) {
+ TraceContext context = record.getContext();
+ // Event records and execution records without a run or a status
are not spans.
+ if (TraceContext.EVENT_ENTITY_TYPE.equals(context.getEntityType())
+ || context.getExecutionId() == null
+ || context.getInputRunId() == null
+ || !isKnownStatus(record.getStatus())) {
+ continue;
+ }
+ if (!isValidTimestamp(record.getTimestamp())) {
+ // Skip just this record, as for an undecodable one, rather
than failing the run.
+ diagnostics.add(
+ new ConverterDiagnostic(
+ ConverterDiagnostic.MALFORMED_RECORD,
+ context.getExecutionId(),
+ "Invalid timestamp '"
+ + record.getTimestamp()
+ + "'; the record was skipped.",
+ null));
+ continue;
+ }
+ executions
+ .computeIfAbsent(context.getExecutionId(), id -> new
ExecutionSpanBuilder())
+ .accept(record);
+ runs.computeIfAbsent(context.getInputRunId(), id -> new
RunAccumulator())
+ .accept(record);
+ }
+
+ List<SpanData> spans = new ArrayList<>(executions.size() +
runs.size());
+ for (Map.Entry<String, RunAccumulator> run : runs.entrySet()) {
+ spans.add(buildRunRootSpan(run.getKey(), run.getValue()));
+ }
+ for (ExecutionSpanBuilder execution : executions.values()) {
+ spans.add(buildExecutionSpan(execution, executions.keySet(),
diagnostics));
+ }
+ return spans;
+ }
+
+ private SpanData buildRunRootSpan(String inputRunId, RunAccumulator run) {
+ AttributesBuilder attributes = Attributes.builder();
+ attributes.put(GEN_AI_OPERATION_NAME, "invoke_agent");
+ attributes.put(FA_INPUT_RUN_ID, inputRunId);
+ if (run.agentName != null) {
+ attributes.put(GEN_AI_AGENT_NAME, run.agentName);
+ }
+ putBusinessKey(attributes, run.businessKey);
+ String name = run.agentName != null ? "invoke_agent " + run.agentName
: "invoke_agent";
+ return new AgentTraceSpanData(
+ name,
+ SpanKind.INTERNAL,
+ spanContext(inputRunId, OTelIds.runRootSpanId(inputRunId)),
+ SpanContext.getInvalid(),
+ StatusData.unset(),
+ run.minEpochNanos,
+ run.maxEpochNanos,
+ attributes.build(),
+ resource,
+ scope);
+ }
+
+ private SpanData buildExecutionSpan(
+ ExecutionSpanBuilder execution,
+ Set<String> executionIds,
+ List<ConverterDiagnostic> diagnostics) {
+ TraceContext any = execution.anyRecord().getContext();
+ String inputRunId = any.getInputRunId();
+ String entityType = any.getEntityType();
+ String entityName = any.getEntityName();
+
+ AttributesBuilder attributes = Attributes.builder();
+ attributes.put(FA_INPUT_RUN_ID, inputRunId);
+ attributes.put(FA_EXECUTION_ID, any.getExecutionId());
+ attributes.put(FA_ENTITY_TYPE, entityType);
+ attributes.put(FA_ENTITY_NAME, entityName);
+ putBusinessKey(attributes, any.getBusinessKey());
+
+ String name;
+ SpanKind kind;
+ Map<String, Object> metadata = any.getEntityMetadata();
+ if (ExecutionReporter.EntityTypes.LLM.equals(entityType)) {
+ // The entity name is the chat model resource; the requested model
id travels in
+ // entityMetadata. gen_ai.provider.name stays unset: the record
does not carry it.
+ String model = stringValue(metadata,
LLMExecutionMetadataKeys.MODEL);
+ name = model != null ? "chat " + model : "chat";
+ kind = SpanKind.CLIENT;
+ attributes.put(GEN_AI_OPERATION_NAME, "chat");
+ if (model != null) {
+ attributes.put(GEN_AI_REQUEST_MODEL, model);
+ }
+ } else if (ExecutionReporter.EntityTypes.TOOL.equals(entityType)) {
+ name = "execute_tool " + entityName;
+ // INTERNAL, as the GenAI conventions specify for execute_tool:
the span measures
+ // the framework running the tool, whatever transport the tool
itself uses.
+ kind = SpanKind.INTERNAL;
+ attributes.put(GEN_AI_OPERATION_NAME, "execute_tool");
+ attributes.put(GEN_AI_TOOL_NAME, entityName);
+ String agentName = execution.agentName();
+ if (agentName != null) {
+ attributes.put(GEN_AI_AGENT_NAME, agentName);
+ }
+ // The call id the model's tool-call request carries.
+ String callId = stringValue(metadata,
ToolExecutionMetadataKeys.TOOL_CALL_ID);
+ if (callId != null) {
+ attributes.put(GEN_AI_TOOL_CALL_ID, callId);
+ }
+ String toolType = stringValue(metadata,
ToolExecutionMetadataKeys.TOOL_TYPE);
+ if (toolType != null) {
+ attributes.put(FA_TOOL_TYPE, toolType);
+ String conventionType = conventionToolType(toolType);
+ if (conventionType != null) {
+ attributes.put(GEN_AI_TOOL_TYPE, conventionType);
+ }
+ }
+ } else if (ExecutionReporter.EntityTypes.SUBAGENT.equals(entityType)) {
+ // A sub-agent delegation the model requested as a tool call.
Records don't
+ // distinguish in-process agents from remote async jobs, so this
keeps the
+ // fallback's INTERNAL kind until they carry locality.
+ name = "invoke_agent " + entityName;
+ kind = SpanKind.INTERNAL;
+ attributes.put(GEN_AI_OPERATION_NAME, "invoke_agent");
+ attributes.put(GEN_AI_AGENT_NAME, entityName);
+ String callId = stringValue(metadata,
ToolExecutionMetadataKeys.TOOL_CALL_ID);
+ if (callId != null) {
+ attributes.put(GEN_AI_TOOL_CALL_ID, callId);
+ }
+ } else if (ExecutionReporter.EntityTypes.ACTION.equals(entityType)) {
+ name = "action " + entityName;
+ kind = SpanKind.INTERNAL;
+ } else if (ExecutionReporter.EntityTypes.PARSER.equals(entityType)) {
+ name = "parse " + entityName;
+ kind = SpanKind.INTERNAL;
+ // A low-cardinality custom value: the GenAI conventions permit
custom operation
+ // names when no well-known value applies (parser has none).
+ attributes.put(GEN_AI_OPERATION_NAME, "parse");
+ } else {
+ name = entityType + " " + entityName;
+ kind = SpanKind.INTERNAL;
+ }
+
+ StatusData status = StatusData.unset();
+ TraceRecord terminal = execution.terminal;
+ if (terminal != null) {
+ if (TraceRecord.Statuses.FAILED.equals(terminal.getStatus())) {
+ String errorMessage = attributeString(terminal,
"errorMessage");
+ status =
+ StatusData.create(
+ StatusCode.ERROR, errorMessage != null ?
errorMessage : "");
+ String errorType = attributeString(terminal, "errorType");
+ if (errorType == null) {
+ errorType = terminal.getProblemCategory();
+ }
+ if (errorType != null) {
+ attributes.put(ERROR_TYPE, errorType);
+ }
+ }
+ attributes.put(FA_EXECUTION_STATUS, terminal.getStatus());
+ // Usage travels on the success record's metadata, which can
differ from the start
+ // record's metadata for the same execution.
+ putUsageIfPresent(
+ attributes,
+
terminal.getContext().getEntityMetadata().get(LLMExecutionMetadataKeys.USAGE));
+ }
+ boolean reused =
+ terminal != null &&
TraceRecord.Statuses.REUSED.equals(terminal.getStatus());
+ if (terminal == null) {
+ // A start with no terminal: crash, a best-effort write that
dropped the terminal
+ // record, or recovery discarding the transient pairing —
indistinguishable here, so
+ // the span keeps status UNSET and carries an explicit marker
instead of ERROR.
+ attributes.put(FA_EXECUTION_INCOMPLETE, true);
+ diagnostics.add(
+ new ConverterDiagnostic(
+ ConverterDiagnostic.INCOMPLETE_EXECUTION,
+ any.getExecutionId(),
+ "Execution has a start record but no terminal
record; exported as a"
+ + " zero-duration span with status UNSET.",
+ null));
+ } else if (execution.start() == null && !reused) {
+ // A terminal with no start (reused executions are single-record
by design).
+ attributes.put(FA_EXECUTION_INCOMPLETE, true);
+ diagnostics.add(
+ new ConverterDiagnostic(
+ ConverterDiagnostic.MISSING_START,
+ any.getExecutionId(),
+ "Execution has a terminal record but no start
record; exported as a"
+ + " zero-duration span at the terminal
timestamp.",
+ null));
+ }
+
+ TraceRecord start = execution.start();
+ long startNanos =
+ start != null
+ ? epochNanos(start.getTimestamp())
+ : epochNanos(terminal.getTimestamp());
+ long endNanos = terminal != null ? epochNanos(terminal.getTimestamp())
: startNanos;
+
+ String parentId = any.getParentExecutionId();
+ SpanContext parent;
+ if (parentId != null && executionIds.contains(parentId)) {
+ parent = spanContext(inputRunId, OTelIds.spanId(parentId));
+ } else {
+ parent = spanContext(inputRunId,
OTelIds.runRootSpanId(inputRunId));
+ if (parentId != null) {
+ // The parent has no records, e.g. Trace Log targets that keep
tool calls but not
+ // Actions; attach to the run root rather than to a span that
never arrives.
+ attributes.put(FA_PARENT_EXECUTION_ID, parentId);
+ diagnostics.add(
+ new ConverterDiagnostic(
+ ConverterDiagnostic.MISSING_PARENT,
+ any.getExecutionId(),
+ "Parent execution "
+ + parentId
+ + " has no records; the span was
attached to the run"
+ + " root.",
+ null));
+ }
+ }
+
+ return new AgentTraceSpanData(
+ name,
+ kind,
+ spanContext(inputRunId, OTelIds.spanId(any.getExecutionId())),
+ parent,
+ status,
+ startNanos,
+ Math.max(endNanos, startNanos),
+ attributes.build(),
+ resource,
+ scope);
+ }
+
+ /**
+ * Maps the framework tool type onto the GenAI well-known {@code
gen_ai.tool.type} values: a
+ * function the agent runs itself is {@code function}; remote functions
and MCP tools call out
+ * to external systems, which is {@code extension}. Model built-in tools
have no well-known
+ * counterpart and keep only the raw {@code flink_agents.tool.type}.
+ */
+ private static String conventionToolType(String toolType) {
+ switch (toolType) {
+ case "function":
+ return "function";
+ case "remote_function":
+ case "mcp":
+ return "extension";
+ default:
+ return null;
+ }
+ }
+
+ private void putBusinessKey(AttributesBuilder attributes, String
businessKey) {
+ if (businessKey == null) {
+ return;
+ }
+ attributes.put(FA_BUSINESS_KEY, businessKey);
+ if (businessKeyAsConversationId) {
+ attributes.put(GEN_AI_CONVERSATION_ID, businessKey);
+ }
+ }
+
+ private static String stringValue(Map<String, Object> map, String key) {
+ Object value = map.get(key);
+ return value == null ? null : String.valueOf(value);
+ }
+
+ /**
+ * A string attribute of the record. At {@code STANDARD} level the Trace
Log replaces a long
+ * string with {@code {"truncatedString": "<prefix>...", "omittedChars":
N}}; the prefix is
+ * kept.
+ */
+ private static String attributeString(TraceRecord record, String name) {
+ Object value = record.getAttributes().get(name);
+ if (value instanceof Map) {
+ value = ((Map<?, ?>) value).get("truncatedString");
+ }
+ return value instanceof String ? (String) value : null;
+ }
+
+ private static void putUsageIfPresent(AttributesBuilder attributes, Object
usage) {
+ Map<?, ?> tokens = usage instanceof Map ? (Map<?, ?>) usage :
Collections.emptyMap();
+ Object prompt = tokens.get("promptTokens");
+ if (prompt instanceof Number) {
+ attributes.put(GEN_AI_USAGE_INPUT_TOKENS, ((Number)
prompt).longValue());
+ }
+ Object completion = tokens.get("completionTokens");
+ if (completion instanceof Number) {
+ attributes.put(GEN_AI_USAGE_OUTPUT_TOKENS, ((Number)
completion).longValue());
+ }
+ }
+
+ private static boolean isKnownStatus(String status) {
+ return TraceRecord.Statuses.CREATED.equals(status)
+ || TraceRecord.Statuses.STARTED.equals(status)
+ || TraceRecord.Statuses.SUCCESS.equals(status)
+ || TraceRecord.Statuses.FAILED.equals(status)
+ || TraceRecord.Statuses.REUSED.equals(status);
+ }
+
+ private static SpanContext spanContext(String inputRunId, String
spanIdHex) {
+ return SpanContext.create(
+ OTelIds.traceId(inputRunId),
+ spanIdHex,
+ TraceFlags.getSampled(),
+ TraceState.getDefault());
+ }
+
+ private static boolean isValidTimestamp(String isoTimestamp) {
+ try {
+ Instant.parse(isoTimestamp);
+ return true;
+ } catch (DateTimeParseException e) {
+ return false;
+ }
+ }
+
+ private static long epochNanos(String isoTimestamp) {
+ Instant instant = Instant.parse(isoTimestamp);
+ return TimeUnit.SECONDS.toNanos(instant.getEpochSecond()) +
instant.getNano();
+ }
+
+ /**
+ * Created/started/terminal records for one execution id. When the Trace
Log repeats a record
+ * (outputs are best-effort and may duplicate), the earliest one of each
kind is kept, so the
+ * result does not depend on record order.
+ */
+ private static final class ExecutionSpanBuilder {
+ private TraceRecord created;
+ private TraceRecord started;
+ private TraceRecord terminal;
+
+ void accept(TraceRecord record) {
+ String status = record.getStatus();
+ if (TraceRecord.Statuses.CREATED.equals(status)) {
+ created = earlier(created, record);
+ } else if (TraceRecord.Statuses.STARTED.equals(status)) {
+ started = earlier(started, record);
+ } else {
+ // success / failed / reused all terminate the execution. A
reused execution has
+ // no started record: it becomes a point-in-time span with
status "reused".
+ terminal = earlier(terminal, record);
+ }
+ }
+
+ private static TraceRecord earlier(TraceRecord current, TraceRecord
candidate) {
+ return current == null
+ || epochNanos(candidate.getTimestamp())
+ < epochNanos(current.getTimestamp())
+ ? candidate
+ : current;
+ }
+
+ /**
+ * Where the span starts: the started record, else the created record.
A created record
+ * without a started one is a Tool that was queued but never reported
starting, including
+ * one whose preparation failed; the created record is never a
terminal.
+ */
+ TraceRecord start() {
+ return started != null ? started : created;
+ }
+
+ TraceRecord anyRecord() {
+ TraceRecord start = start();
+ return start != null ? start : terminal;
+ }
+
+ /** The agent name from whichever of this execution's records carries
one. */
+ String agentName() {
+ for (TraceRecord record : new TraceRecord[] {started, created,
terminal}) {
+ if (record != null && record.getContext().getAgentName() !=
null) {
+ return record.getContext().getAgentName();
+ }
+ }
+ return null;
+ }
+ }
+
+ /** Aggregated run-level facts for the synthesized root span. */
+ private static final class RunAccumulator {
+ private long minEpochNanos = Long.MAX_VALUE;
+ private long maxEpochNanos = Long.MIN_VALUE;
+ private String agentName;
+ private String businessKey;
+
+ void accept(TraceRecord record) {
+ long nanos = epochNanos(record.getTimestamp());
+ minEpochNanos = Math.min(minEpochNanos, nanos);
+ maxEpochNanos = Math.max(maxEpochNanos, nanos);
+ TraceContext context = record.getContext();
+ if (agentName == null && context.getAgentName() != null) {
+ agentName = context.getAgentName();
+ }
+ if (businessKey == null && context.getBusinessKey() != null) {
+ businessKey = context.getBusinessKey();
+ }
+ }
+ }
+}
diff --git
a/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/ConverterDiagnostic.java
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/ConverterDiagnostic.java
new file mode 100644
index 000000000..50e6d6b66
--- /dev/null
+++
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/ConverterDiagnostic.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.integrations.observability.otel;
+
+import java.util.Objects;
+
+/**
+ * One machine-readable converter diagnostic, mirroring the warning model of
{@code trace_tree.py}
+ * ({@code code} + id + {@code message} + optional file location) so the two
Trace Log tools share
+ * one diagnostic vocabulary.
+ *
+ * <p>Codes emitted by this converter:
+ *
+ * <ul>
+ * <li>{@link #MALFORMED_RECORD}: an input record could not be parsed, or
its timestamp could not;
+ * it was skipped.
+ * <li>{@link #INCOMPLETE_EXECUTION}: a start record has no terminal record.
The span is still
+ * exported, closed at the observed start timestamp (zero duration) with
status UNSET and the
+ * {@code flink_agents.execution.incomplete} attribute — a missing
terminal cannot be
+ * distinguished between a crash, a best-effort Trace Log write that
dropped the terminal
+ * record, and recovery discarding the transient start/terminal pairing.
+ * <li>{@link #MISSING_START}: a terminal record has no start record (same
export policy).
+ * <li>{@link #MISSING_PARENT}: the parent execution has no records, for
example when the Trace
+ * Log targets keep calls but not their Actions. The span is attached to
the run root and
+ * keeps the parent id in {@code flink_agents.parent_execution_id}.
+ * </ul>
+ */
+public final class ConverterDiagnostic {
+
+ public static final String MALFORMED_RECORD = "MALFORMED_RECORD";
+ public static final String INCOMPLETE_EXECUTION = "INCOMPLETE_EXECUTION";
+ public static final String MISSING_START = "MISSING_START";
+ public static final String MISSING_PARENT = "MISSING_PARENT";
+
+ private final String code;
+ private final String executionId;
+ private final String message;
+ private final String filePath;
+
+ ConverterDiagnostic(String code, String executionId, String message,
String filePath) {
+ this.code = Objects.requireNonNull(code);
+ this.executionId = executionId;
+ this.message = Objects.requireNonNull(message);
+ this.filePath = filePath;
+ }
+
+ public String getCode() {
+ return code;
+ }
+
+ public String getExecutionId() {
+ return executionId;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public String getFilePath() {
+ return filePath;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder(code);
+ if (executionId != null) {
+ sb.append(" [executionId=").append(executionId).append(']');
+ }
+ if (filePath != null) {
+ sb.append(" [file=").append(filePath).append(']');
+ }
+ return sb.append(": ").append(message).toString();
+ }
+}
diff --git
a/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/OTelIds.java
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/OTelIds.java
new file mode 100644
index 000000000..1e6358817
--- /dev/null
+++
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/OTelIds.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.integrations.observability.otel;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
+/**
+ * Deterministic derivation of fixed-width OpenTelemetry ids from the opaque
framework ids.
+ *
+ * <p>Per the design in the Agent Trace discussions, the framework keeps
{@code inputRunId} and
+ * {@code executionId} opaque; the exporter derives valid 128-bit trace ids
and 64-bit span ids by
+ * hashing them. Determinism makes re-exports idempotent: exporting the same
Trace Log twice
+ * produces byte-identical ids, so an OTLP backend deduplicates instead of
double-counting.
+ */
+final class OTelIds {
+
+ /** Domain separator so a run-root span id can never collide with an
execution span id. */
+ private static final String RUN_SPAN_DOMAIN = "/input-run-root";
+
+ private OTelIds() {}
+
+ /** Derives the 32-hex-char OTel trace id for an input run. */
+ static String traceId(String inputRunId) {
+ return hexDigest(inputRunId, 16);
+ }
+
+ /** Derives the 16-hex-char OTel span id for an execution. */
+ static String spanId(String executionId) {
+ return hexDigest(executionId, 8);
+ }
+
+ /** Derives the 16-hex-char OTel span id for the synthesized run-root
span. */
+ static String runRootSpanId(String inputRunId) {
+ return hexDigest(inputRunId + RUN_SPAN_DOMAIN, 8);
+ }
+
+ private static String hexDigest(String value, int bytes) {
+ byte[] digest = sha256(value.getBytes(StandardCharsets.UTF_8));
+ StringBuilder hex = new StringBuilder(bytes * 2);
+ boolean allZero = true;
+ for (int i = 0; i < bytes; i++) {
+ if (digest[i] != 0) {
+ allZero = false;
+ }
+ }
+ // An all-zero id is invalid in OTel. Astronomically unlikely from
SHA-256, but cheap to
+ // guard: flip the last byte so the id stays deterministic and valid.
+ if (allZero) {
+ digest[bytes - 1] = 1;
+ }
+ for (int i = 0; i < bytes; i++) {
+ hex.append(Character.forDigit((digest[i] >> 4) & 0xF, 16));
+ hex.append(Character.forDigit(digest[i] & 0xF, 16));
+ }
+ return hex.toString();
+ }
+
+ private static byte[] sha256(byte[] input) {
+ try {
+ return MessageDigest.getInstance("SHA-256").digest(input);
+ } catch (NoSuchAlgorithmException e) {
+ // SHA-256 is mandatory on every JVM.
+ throw new IllegalStateException("SHA-256 unavailable", e);
+ }
+ }
+}
diff --git
a/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/TraceLogOTelExporter.java
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/TraceLogOTelExporter.java
new file mode 100644
index 000000000..a4406d319
--- /dev/null
+++
b/integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/TraceLogOTelExporter.java
@@ -0,0 +1,447 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.integrations.observability.otel;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.MappingIterator;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
+import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+import org.apache.flink.agents.api.trace.TraceRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Exports Trace Logs as OpenTelemetry GenAI traces.
+ *
+ * <p>This is the out-of-band exporter agreed in the Agent Trace design
discussions: it reads the
+ * Trace Log written by the File or SLF4J trace logger, assembles spans per
{@link AgentTraceSpans},
+ * and pushes them to any OTLP endpoint. The Trace Log must keep execution
records ({@code
+ * trace-log.targets} other than the default {@code EVENT_ONLY}). Running out
of band means zero
+ * impact on the Flink job, and it works retroactively on logs from already
finished runs. Because
+ * span/trace ids are derived deterministically from the framework ids,
exporting the same log twice
+ * is idempotent on the backend.
+ *
+ * <p>Input files are streams of JSON objects — the JSONL written by the File
trace logger, and also
+ * a pretty-printed variant (the parser consumes concatenated JSON objects
regardless of line
+ * breaks).
+ *
+ * <p>Example usage:
+ *
+ * <pre>{@code
+ * TraceLogOTelExporter exporter =
+ * TraceLogOTelExporter.builder()
+ * .setEndpoint("http://localhost:4317";)
+ * .setProtocol("grpc")
+ * .setServiceName("my-agent-job")
+ * .build();
+ *
exporter.exportFiles(List.of(Path.of("/tmp/flink-agents/traces-<jobId>-<task>-0.log")));
+ * exporter.shutdown();
+ * }</pre>
+ */
+public class TraceLogOTelExporter implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TraceLogOTelExporter.class);
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ /** Per-file cap on malformed-record diagnostics, guarding against a stuck
parser. */
+ private static final int MAX_MALFORMED_PER_FILE = 1000;
+
+ /** Default spans per OTLP request, matching the SDK {@code
BatchSpanProcessor} default. */
+ static final int DEFAULT_MAX_EXPORT_BATCH_SIZE = 512;
+
+ /** How long each OTLP request may take before the export is reported as
failed. */
+ private static final long BATCH_EXPORT_TIMEOUT_SECONDS = 30;
+
+ private final AgentTraceSpans assembler;
+ private final SpanExporter spanExporter;
+ private final int maxExportBatchSize;
+
+ TraceLogOTelExporter(
+ AgentTraceSpans assembler, SpanExporter spanExporter, int
maxExportBatchSize) {
+ this.assembler = assembler;
+ this.spanExporter = spanExporter;
+ this.maxExportBatchSize = maxExportBatchSize;
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * Reads the given Trace Log files and exports the assembled spans.
+ *
+ * <p>A directory argument is expanded to the {@code traces-*.log} and
{@code *.trace-log.log}
+ * files it contains (the File trace logger naming contract, matching
{@code trace_tree.py}); a
+ * file argument is consumed as-is regardless of its name, so records
collected from any Trace
+ * Log sink work once they are on disk. Completeness of a multi-subtask
file set is the caller's
+ * responsibility under the batch contract; the summary reports what was
read.
+ */
+ public ExportSummary exportFiles(List<Path> files) throws IOException {
+ List<TraceRecord> records = new ArrayList<>();
+ List<ConverterDiagnostic> diagnostics = new ArrayList<>();
+ for (Path file : expandDirectories(files)) {
+ try (InputStream in = Files.newInputStream(file)) {
+ MappingIterator<TraceRecord> it;
+ try {
+ it = MAPPER.readerFor(TraceRecord.class).readValues(in);
+ } catch (JsonProcessingException e) {
+ // The stream is broken from its first token: skip this
file, keep the others.
+ diagnostics.add(malformed(e, file));
+ continue;
+ }
+ readRecords(it, file, records, diagnostics);
+ }
+ }
+ return export(records, diagnostics);
+ }
+
+ private static void readRecords(
+ MappingIterator<TraceRecord> it,
+ Path file,
+ List<TraceRecord> records,
+ List<ConverterDiagnostic> diagnostics)
+ throws IOException {
+ try (it) {
+ int malformedInFile = 0;
+ boolean fileAborted = false;
+ while (!fileAborted) {
+ try {
+ if (!it.hasNext()) {
+ break;
+ }
+ records.add(it.next());
+ } catch (RuntimeException e) {
+ // A malformed record must not abort the whole export.
Value-level errors
+ // allow the iterator to continue; a structurally broken
stream (or a
+ // parser that stops making progress) aborts just this
file, with the
+ // diagnostic naming it either way.
+ diagnostics.add(malformed(e, file));
+ malformedInFile++;
+ fileAborted = malformedInFile >= MAX_MALFORMED_PER_FILE ||
!canContinue(it);
+ }
+ }
+ }
+ }
+
+ private static ConverterDiagnostic malformed(Exception e, Path file) {
+ return new ConverterDiagnostic(
+ ConverterDiagnostic.MALFORMED_RECORD,
+ null,
+ "Could not decode a Trace Log record: " + e.getMessage(),
+ file.toString());
+ }
+
+ private static boolean canContinue(MappingIterator<TraceRecord> it) {
+ try {
+ // Probe the iterator: a value-level bind error leaves it usable,
a structurally
+ // broken stream makes hasNext() itself throw.
+ it.hasNext();
+ return true;
+ } catch (RuntimeException e) {
+ return false;
+ }
+ }
+
+ private static List<Path> expandDirectories(List<Path> paths) throws
IOException {
+ List<Path> expanded = new ArrayList<>();
+ for (Path path : paths) {
+ if (Files.isDirectory(path)) {
+ TreeSet<Path> discovered = new TreeSet<>();
+ for (String glob : new String[] {"traces-*.log",
"*.trace-log.log"}) {
+ try (DirectoryStream<Path> stream =
Files.newDirectoryStream(path, glob)) {
+ stream.forEach(discovered::add);
+ }
+ }
+ expanded.addAll(discovered);
+ } else {
+ expanded.add(path);
+ }
+ }
+ return expanded;
+ }
+
+ /** Assembles and exports spans from already-parsed records. */
+ public ExportSummary exportRecords(List<TraceRecord> records) {
+ return export(records, new ArrayList<>());
+ }
+
+ private ExportSummary export(List<TraceRecord> records,
List<ConverterDiagnostic> diagnostics) {
+ List<SpanData> spans = assembler.assemble(records, diagnostics);
+ for (ConverterDiagnostic diagnostic : diagnostics) {
+ LOG.warn("{}", diagnostic);
+ }
+ // Each OTLP exporter marshals a whole collection into one request,
and receivers cap the
+ // request size (4 MiB by default for gRPC), so the spans go out in
bounded batches.
+ int exported = 0;
+ while (exported < spans.size()) {
+ List<SpanData> batch =
+ spans.subList(exported, Math.min(exported +
maxExportBatchSize, spans.size()));
+ CompletableResultCode result = spanExporter.export(batch);
+ result.join(BATCH_EXPORT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ if (!result.isSuccess()) {
+ throw new ExportFailedException(exported, spans.size());
+ }
+ exported += batch.size();
+ }
+ return new ExportSummary(records.size(), exported, diagnostics);
+ }
+
+ @Override
+ public void close() {
+ shutdown();
+ }
+
+ /** Flushes and shuts down the underlying OTLP exporter. */
+ public void shutdown() {
+ spanExporter.flush().join(10, TimeUnit.SECONDS);
+ spanExporter.shutdown().join(10, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Thrown when an OTLP request fails or times out. Earlier batches have
already been delivered;
+ * re-running the export is safe because span and trace ids are
deterministic.
+ */
+ public static final class ExportFailedException extends
IllegalStateException {
+ private static final long serialVersionUID = 1L;
+
+ private final int spansExported;
+ private final int spansTotal;
+
+ ExportFailedException(int spansExported, int spansTotal) {
+ super(
+ "OTLP export did not complete successfully: exported "
+ + spansExported
+ + " of "
+ + spansTotal
+ + " spans before a batch failed.");
+ this.spansExported = spansExported;
+ this.spansTotal = spansTotal;
+ }
+
+ /** Spans delivered by the batches that completed before the failure.
*/
+ public int getSpansExported() {
+ return spansExported;
+ }
+
+ public int getSpansTotal() {
+ return spansTotal;
+ }
+ }
+
+ /** Counters and diagnostics describing one export invocation. */
+ public static final class ExportSummary {
+ private final int recordsRead;
+ private final int spansExported;
+ private final List<ConverterDiagnostic> diagnostics;
+
+ ExportSummary(int recordsRead, int spansExported,
List<ConverterDiagnostic> diagnostics) {
+ this.recordsRead = recordsRead;
+ this.spansExported = spansExported;
+ this.diagnostics = Collections.unmodifiableList(new
ArrayList<>(diagnostics));
+ }
+
+ public int getRecordsRead() {
+ return recordsRead;
+ }
+
+ public int getSpansExported() {
+ return spansExported;
+ }
+
+ /** Machine-readable diagnostics, per the {@code trace_tree.py}
warning model. */
+ public List<ConverterDiagnostic> getDiagnostics() {
+ return diagnostics;
+ }
+
+ @Override
+ public String toString() {
+ return "read "
+ + recordsRead
+ + " records, exported "
+ + spansExported
+ + " spans, "
+ + diagnostics.size()
+ + " diagnostics";
+ }
+ }
+
+ /** Builder for {@link TraceLogOTelExporter}. */
+ public static final class Builder {
+ private String endpoint = "http://localhost:4317";;
+ private String protocol = "grpc";
+ private String serviceName = "flink-agents";
+ private int maxExportBatchSize = DEFAULT_MAX_EXPORT_BATCH_SIZE;
+ private boolean businessKeyAsConversationId;
+ private SpanExporter spanExporter;
+
+ /** OTLP endpoint, e.g. {@code http://localhost:4317} (grpc) or {@code
.../v1/traces}. */
+ public Builder setEndpoint(String endpoint) {
+ this.endpoint = endpoint;
+ return this;
+ }
+
+ /** OTLP transport protocol: {@code grpc} (default) or {@code
http/protobuf}. */
+ public Builder setProtocol(String protocol) {
+ this.protocol = protocol;
+ return this;
+ }
+
+ /** Value of the {@code service.name} resource attribute; defaults to
"flink-agents". */
+ public Builder setServiceName(String serviceName) {
+ this.serviceName = serviceName;
+ return this;
+ }
+
+ /** Maximum spans per OTLP request; defaults to 512. */
+ public Builder setMaxExportBatchSize(int maxExportBatchSize) {
+ if (maxExportBatchSize <= 0) {
+ throw new IllegalArgumentException(
+ "The export batch size must be positive, got " +
maxExportBatchSize + ".");
+ }
+ this.maxExportBatchSize = maxExportBatchSize;
+ return this;
+ }
+
+ /**
+ * Whether to also report {@code businessKey} as {@code
gen_ai.conversation.id}, for
+ * pipelines keyed by conversation. Off by default; the key is always
exported as {@code
+ * flink_agents.business_key}.
+ */
+ public Builder setBusinessKeyAsConversationId(boolean
businessKeyAsConversationId) {
+ this.businessKeyAsConversationId = businessKeyAsConversationId;
+ return this;
+ }
+
+ /** Overrides the underlying exporter; intended for tests. */
+ public Builder setSpanExporter(SpanExporter spanExporter) {
+ this.spanExporter = spanExporter;
+ return this;
+ }
+
+ public TraceLogOTelExporter build() {
+ SpanExporter exporter = spanExporter;
+ if (exporter == null) {
+ if ("grpc".equalsIgnoreCase(protocol)) {
+ exporter =
OtlpGrpcSpanExporter.builder().setEndpoint(endpoint).build();
+ } else if ("http/protobuf".equalsIgnoreCase(protocol)
+ || "http".equalsIgnoreCase(protocol)) {
+ exporter =
OtlpHttpSpanExporter.builder().setEndpoint(endpoint).build();
+ } else {
+ throw new IllegalArgumentException(
+ "Unsupported OTLP protocol: "
+ + protocol
+ + " (expected 'grpc' or
'http/protobuf').");
+ }
+ }
+ return new TraceLogOTelExporter(
+ new AgentTraceSpans(serviceName,
businessKeyAsConversationId),
+ exporter,
+ maxExportBatchSize);
+ }
+ }
+
+ /**
+ * Command-line entry point.
+ *
+ * <p>Usage: {@code TraceLogOTelExporter [--endpoint URL] [--protocol
grpc|http/protobuf]
+ * [--service-name NAME] [--batch-size N]
[--business-key-as-conversation-id]
+ * traceLogFileOrDir...}
+ */
+ public static void main(String[] args) throws IOException {
+ Builder builder = builder();
+ List<Path> files = new ArrayList<>();
+ for (int i = 0; i < args.length; i++) {
+ String arg = args[i];
+ boolean isFlag =
+ arg.equals("--endpoint")
+ || arg.equals("--protocol")
+ || arg.equals("--service-name")
+ || arg.equals("--batch-size");
+ if (isFlag && i + 1 >= args.length) {
+ exitWithUsage("Missing value for " + arg + ".");
+ }
+ switch (arg) {
+ case "--endpoint":
+ builder.setEndpoint(args[++i]);
+ break;
+ case "--protocol":
+ builder.setProtocol(args[++i]);
+ break;
+ case "--service-name":
+ builder.setServiceName(args[++i]);
+ break;
+ case "--batch-size":
+ builder.setMaxExportBatchSize(parseBatchSize(args[++i]));
+ break;
+ case "--business-key-as-conversation-id":
+ builder.setBusinessKeyAsConversationId(true);
+ break;
+ default:
+ files.add(Path.of(arg));
+ }
+ }
+ if (files.isEmpty()) {
+ exitWithUsage("No Trace Log file or directory given.");
+ }
+ try (TraceLogOTelExporter exporter = builder.build()) {
+ ExportSummary summary = exporter.exportFiles(files);
+ System.out.println(summary);
+ for (ConverterDiagnostic diagnostic : summary.getDiagnostics()) {
+ System.err.println(diagnostic);
+ }
+ }
+ }
+
+ private static int parseBatchSize(String value) {
+ try {
+ int batchSize = Integer.parseInt(value);
+ if (batchSize > 0) {
+ return batchSize;
+ }
+ } catch (NumberFormatException e) {
+ // Reported below.
+ }
+ exitWithUsage("--batch-size needs a positive integer, got '" + value +
"'.");
+ return DEFAULT_MAX_EXPORT_BATCH_SIZE;
+ }
+
+ private static void exitWithUsage(String problem) {
+ System.err.println(problem);
+ System.err.println(
+ "Usage: TraceLogOTelExporter [--endpoint URL] [--protocol
grpc|http/protobuf]"
+ + " [--service-name NAME] [--batch-size N]"
+ + " [--business-key-as-conversation-id]
traceLogFileOrDir...");
+ System.exit(2);
+ }
+}
diff --git
a/integrations/observability/otel/src/test/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpansTest.java
b/integrations/observability/otel/src/test/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpansTest.java
new file mode 100644
index 000000000..e3991b4ee
--- /dev/null
+++
b/integrations/observability/otel/src/test/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpansTest.java
@@ -0,0 +1,783 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.integrations.observability.otel;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.flink.agents.api.trace.TraceRecord;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link AgentTraceSpans}: span topology, attributes, and id
determinism. */
+class AgentTraceSpansTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final String RUN = "7f1b5d20-86c6-4a5d-b65a-9c41757f2e11";
+ private static final String ACTION = "act-1111";
+ private static final String LLM = "llm-2222";
+ private static final String TOOL = "tool-3333";
+
+ // The metadata ToolCallAction records: the provider's call id and the
tool type.
+ private static final String TOOL_METADATA =
+
"\"entityMetadata\":{\"toolCallId\":\"call_abc\",\"toolType\":\"function\"},";
+
+ private static TraceRecord record(String json) {
+ try {
+ return MAPPER.readValue(json, TraceRecord.class);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static List<TraceRecord> sampleRun() {
+ return List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\",\"inputRunId\":\""
+ + RUN
+ +
"\",\"businessKey\":\"order-1001\",\"agentName\":\"ReActAgent\","
+ + "\"executionId\":\""
+ + ACTION
+ + "\",\"entityType\":\"action\","
+ + "\"entityName\":\"review\","
+ + "\"status\":\"started\"}"),
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:01Z\",\"inputRunId\":\""
+ + RUN
+ +
"\",\"businessKey\":\"order-1001\",\"executionId\":\""
+ + LLM
+ + "\",\"parentExecutionId\":\""
+ + ACTION
+ + "\","
+ +
"\"entityType\":\"llm\",\"entityName\":\"chatModel\","
+ +
"\"entityMetadata\":{\"model\":\"qwen-max\"},"
+ + "\"status\":\"started\"}"),
+ record(
+
"{\"schemaVersion\":1,\"timestamp\":\"2026-01-15T10:30:03Z\",\"inputRunId\":\""
+ + RUN
+ + "\",\"executionId\":\""
+ + LLM
+ + "\",\"parentExecutionId\":\""
+ + ACTION
+ +
"\",\"entityType\":\"llm\",\"entityName\":\"chatModel\","
+ + "\"entityMetadata\":{\"model\":\"qwen-max\","
+ +
"\"usage\":{\"promptTokens\":120,\"completionTokens\":45}},"
+ + "\"status\":\"success\",\"attributes\":{}}"),
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:04Z\",\"inputRunId\":\""
+ + RUN
+ + "\",\"executionId\":\""
+ + TOOL
+ + "\",\"parentExecutionId\":\""
+ + ACTION
+ +
"\",\"entityType\":\"tool\",\"entityName\":\"get_weather\","
+ + TOOL_METADATA
+ + "\"status\":\"started\"}"),
+ record(
+
"{\"schemaVersion\":1,\"timestamp\":\"2026-01-15T10:30:05Z\",\"inputRunId\":\""
+ + RUN
+ + "\",\"executionId\":\""
+ + TOOL
+ + "\",\"parentExecutionId\":\""
+ + ACTION
+ +
"\",\"entityType\":\"tool\",\"entityName\":\"get_weather\","
+ + TOOL_METADATA
+ + "\"status\":\"failed\","
+ + "\"problemCategory\":\"tool_call_failed\","
+ +
"\"attributes\":{\"errorType\":\"java.io.IOException\","
+ + "\"errorMessage\":\"connection refused\"}}"),
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:06Z\",\"inputRunId\":\""
+ + RUN
+ + "\",\"executionId\":\""
+ + ACTION
+ + "\",\"entityType\":\"action\","
+ + "\"entityName\":\"review\","
+ + "\"status\":\"success\"}"));
+ }
+
+ private static SpanData spanNamed(List<SpanData> spans, String name) {
+ return spans.stream()
+ .filter(s -> s.getName().equals(name))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new AssertionError(
+ "no span named '"
+ + name
+ + "' in "
+ + spans.stream()
+ .map(SpanData::getName)
+
.collect(Collectors.toList())));
+ }
+
+ @Test
+ @DisplayName("One input run becomes one trace with a synthesized
invoke_agent root span")
+ void testRunBecomesSingleTraceWithRootSpan() {
+ List<SpanData> spans = new
AgentTraceSpans("test-service").assemble(sampleRun());
+
+ assertThat(spans).hasSize(4);
+ assertThat(spans.stream().map(s ->
s.getSpanContext().getTraceId()).distinct()).hasSize(1);
+
+ SpanData root = spanNamed(spans, "invoke_agent ReActAgent");
+ assertThat(root.getParentSpanContext().isValid()).isFalse();
+
assertThat(root.getAttributes().get(AgentTraceSpans.GEN_AI_OPERATION_NAME))
+ .isEqualTo("invoke_agent");
+ assertThat(root.getAttributes().get(AgentTraceSpans.GEN_AI_AGENT_NAME))
+ .isEqualTo("ReActAgent");
+ assertThat(root.getAttributes().get(AgentTraceSpans.FA_BUSINESS_KEY))
+ .isEqualTo("order-1001");
+ // Root covers the whole run.
+ assertThat(root.getStartEpochNanos())
+ .isEqualTo(
+
spans.stream().mapToLong(SpanData::getStartEpochNanos).min().orElseThrow());
+ assertThat(root.getEndEpochNanos())
+ .isEqualTo(
+
spans.stream().mapToLong(SpanData::getEndEpochNanos).max().orElseThrow());
+ }
+
+ @Test
+ @DisplayName("Execution hierarchy maps to span parenting via
parentExecutionId")
+ void testParenting() {
+ List<SpanData> spans = new
AgentTraceSpans("test-service").assemble(sampleRun());
+
+ SpanData root = spanNamed(spans, "invoke_agent ReActAgent");
+ SpanData action = spanNamed(spans, "action review");
+ SpanData llm = spanNamed(spans, "chat qwen-max");
+ SpanData tool = spanNamed(spans, "execute_tool get_weather");
+
+ assertThat(action.getParentSpanContext().getSpanId())
+ .isEqualTo(root.getSpanContext().getSpanId());
+ assertThat(llm.getParentSpanContext().getSpanId())
+ .isEqualTo(action.getSpanContext().getSpanId());
+ assertThat(tool.getParentSpanContext().getSpanId())
+ .isEqualTo(action.getSpanContext().getSpanId());
+ }
+
+ @Test
+ @DisplayName("GenAI attributes: chat span carries model + usage, tool
failure carries error")
+ void testGenAiAttributes() {
+ List<SpanData> spans = new
AgentTraceSpans("test-service").assemble(sampleRun());
+
+ SpanData llm = spanNamed(spans, "chat qwen-max");
+ assertThat(llm.getKind()).isEqualTo(SpanKind.CLIENT);
+
assertThat(llm.getAttributes().get(AgentTraceSpans.GEN_AI_OPERATION_NAME))
+ .isEqualTo("chat");
+
assertThat(llm.getAttributes().get(AgentTraceSpans.GEN_AI_REQUEST_MODEL))
+ .isEqualTo("qwen-max");
+
assertThat(llm.getAttributes().get(AgentTraceSpans.GEN_AI_USAGE_INPUT_TOKENS))
+ .isEqualTo(120L);
+
assertThat(llm.getAttributes().get(AgentTraceSpans.GEN_AI_USAGE_OUTPUT_TOKENS))
+ .isEqualTo(45L);
+
+ SpanData tool = spanNamed(spans, "execute_tool get_weather");
+ // INTERNAL: the record cannot distinguish an in-process function tool
from a remote
+ // MCP tool, so the converter does not claim remoteness.
+ assertThat(tool.getKind()).isEqualTo(SpanKind.INTERNAL);
+
assertThat(tool.getStatus().getStatusCode()).isEqualTo(StatusCode.ERROR);
+ assertThat(tool.getStatus().getDescription()).isEqualTo("connection
refused");
+ assertThat(tool.getAttributes().get(AgentTraceSpans.ERROR_TYPE))
+ .isEqualTo("java.io.IOException");
+ assertThat(tool.getAttributes().get(AgentTraceSpans.GEN_AI_TOOL_NAME))
+ .isEqualTo("get_weather");
+ // The provider-issued id wins over the framework-assigned one.
+
assertThat(tool.getAttributes().get(AgentTraceSpans.GEN_AI_TOOL_CALL_ID))
+ .isEqualTo("call_abc");
+ assertThat(tool.getAttributes().get(AgentTraceSpans.GEN_AI_TOOL_TYPE))
+ .isEqualTo("function");
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_TOOL_TYPE)).isEqualTo("function");
+ }
+
+ @Test
+ @DisplayName("Chat span without a recorded model is named 'chat' and sets
no request model")
+ void testChatWithoutModel() {
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:01Z\","
+ +
"\"inputRunId\":\"r\",\"executionId\":\"l\","
+ +
"\"entityType\":\"llm\",\"entityName\":\"chatModel\","
+ +
"\"status\":\"success\"}")));
+
+ SpanData llm = spanNamed(spans, "chat");
+
assertThat(llm.getAttributes().get(AgentTraceSpans.GEN_AI_REQUEST_MODEL)).isNull();
+ }
+
+ @Test
+ @DisplayName("Tool type maps onto gen_ai.tool.type well-known values, raw
value always kept")
+ void testToolTypeMapping() {
+
assertThat(toolSpan("{\"toolCallId\":\"c\",\"toolType\":\"mcp\"}").getAttributes())
+ .satisfies(
+ a -> {
+ assertThat(a.get(AgentTraceSpans.GEN_AI_TOOL_TYPE))
+ .isEqualTo("extension");
+
assertThat(a.get(AgentTraceSpans.FA_TOOL_TYPE)).isEqualTo("mcp");
+ // Without a provider id, the framework call id is
used.
+
assertThat(a.get(AgentTraceSpans.GEN_AI_TOOL_CALL_ID)).isEqualTo("c");
+ });
+
assertThat(toolSpan("{\"toolType\":\"remote_function\"}").getAttributes())
+ .satisfies(
+ a ->
+
assertThat(a.get(AgentTraceSpans.GEN_AI_TOOL_TYPE))
+ .isEqualTo("extension"));
+
assertThat(toolSpan("{\"toolType\":\"model_built_in\"}").getAttributes())
+ .satisfies(
+ a -> {
+
assertThat(a.get(AgentTraceSpans.GEN_AI_TOOL_TYPE)).isNull();
+ assertThat(a.get(AgentTraceSpans.FA_TOOL_TYPE))
+ .isEqualTo("model_built_in");
+ });
+ }
+
+ @Test
+ @DisplayName("error.type falls back to problemCategory when no error type
is recorded")
+ void testErrorTypeFallsBackToProblemCategory() {
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:05Z\","
+ +
"\"inputRunId\":\"r\",\"executionId\":\"t\","
+ +
"\"entityType\":\"tool\",\"entityName\":\"lookup\","
+ +
"\"status\":\"failed\","
+ +
"\"problemCategory\":\"tool_call_failed\"}")));
+
+ SpanData tool = spanNamed(spans, "execute_tool lookup");
+
assertThat(tool.getStatus().getStatusCode()).isEqualTo(StatusCode.ERROR);
+ assertThat(tool.getAttributes().get(AgentTraceSpans.ERROR_TYPE))
+ .isEqualTo("tool_call_failed");
+ }
+
+ private static SpanData toolSpan(String entityMetadataJson) {
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:05Z\","
+ +
"\"inputRunId\":\"r\",\"executionId\":\"t\","
+ +
"\"entityType\":\"tool\",\"entityName\":\"lookup\","
+ + "\"entityMetadata\":"
+ + entityMetadataJson
+ + ","
+ +
"\"status\":\"success\"}")));
+ return spanNamed(spans, "execute_tool lookup");
+ }
+
+ @Test
+ @DisplayName("Ids are derived deterministically: re-assembly yields
identical trace/span ids")
+ void testDeterministicIds() {
+ AgentTraceSpans assembler = new AgentTraceSpans("test-service");
+ List<SpanData> first = assembler.assemble(sampleRun());
+ List<SpanData> second = assembler.assemble(sampleRun());
+
+ assertThat(
+ first.stream()
+ .map(s -> s.getSpanContext().getSpanId())
+ .collect(Collectors.toList()))
+ .isEqualTo(
+ second.stream()
+ .map(s -> s.getSpanContext().getSpanId())
+ .collect(Collectors.toList()));
+ assertThat(first.get(0).getSpanContext().getTraceId())
+ .isEqualTo(second.get(0).getSpanContext().getTraceId());
+ }
+
+ @Test
+ @DisplayName("Record order does not affect the assembled spans")
+ void testRecordOrderDoesNotMatter() {
+ AgentTraceSpans assembler = new AgentTraceSpans("test-service");
+ List<TraceRecord> reversed = new ArrayList<>(sampleRun());
+ Collections.reverse(reversed);
+
+ assertThat(spanShapes(assembler.assemble(reversed)))
+
.containsExactlyInAnyOrderElementsOf(spanShapes(assembler.assemble(sampleRun())));
+ }
+
+ private static List<String> spanShapes(List<SpanData> spans) {
+ return spans.stream()
+ .map(
+ s ->
+ s.getName()
+ + "|"
+ + s.getSpanContext().getSpanId()
+ + "|"
+ + s.getParentSpanContext().getSpanId()
+ + "|"
+ + s.getStartEpochNanos()
+ + "|"
+ + s.getEndEpochNanos()
+ + "|"
+ + s.getStatus().getStatusCode()
+ + "|"
+ + s.getAttributes())
+ .collect(Collectors.toList());
+ }
+
+ @Test
+ @DisplayName("Every execution span carries the framework correlation
attributes")
+ void testCorrelationAttributes() {
+ SpanData tool =
+ spanNamed(
+ new
AgentTraceSpans("test-service").assemble(sampleRun()),
+ "execute_tool get_weather");
+
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_INPUT_RUN_ID)).isEqualTo(RUN);
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_EXECUTION_ID)).isEqualTo(TOOL);
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_ENTITY_TYPE)).isEqualTo("tool");
+ assertThat(tool.getAttributes().get(AgentTraceSpans.FA_ENTITY_NAME))
+ .isEqualTo("get_weather");
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_EXECUTION_STATUS))
+ .isEqualTo("failed");
+ }
+
+ @Test
+ @DisplayName("businessKey is always flink_agents.business_key;
conversation id is opt-in")
+ void testBusinessKeyAsConversationIdIsOptIn() {
+ SpanData byDefault =
+ spanNamed(
+ new
AgentTraceSpans("test-service").assemble(sampleRun()),
+ "invoke_agent ReActAgent");
+
assertThat(byDefault.getAttributes().get(AgentTraceSpans.FA_BUSINESS_KEY))
+ .isEqualTo("order-1001");
+
assertThat(byDefault.getAttributes().get(AgentTraceSpans.GEN_AI_CONVERSATION_ID)).isNull();
+
+ List<SpanData> optedIn = new AgentTraceSpans("test-service",
true).assemble(sampleRun());
+ for (String name : List.of("invoke_agent ReActAgent", "chat
qwen-max")) {
+ SpanData span = spanNamed(optedIn, name);
+
assertThat(span.getAttributes().get(AgentTraceSpans.FA_BUSINESS_KEY))
+ .isEqualTo("order-1001");
+
assertThat(span.getAttributes().get(AgentTraceSpans.GEN_AI_CONVERSATION_ID))
+ .isEqualTo("order-1001");
+ }
+ }
+
+ @Test
+ @DisplayName("Tool spans carry gen_ai.agent.name when their records have
one")
+ void testToolSpanCarriesAgentName() {
+ SpanData named =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ toolRecord(
+ "2026-01-15T10:30:04Z",
+ "started",
+
",\"agentName\":\"ReActAgent\""),
+ toolRecord("2026-01-15T10:30:05Z",
"success")))
+ .get(1);
+
assertThat(named.getAttributes().get(AgentTraceSpans.GEN_AI_AGENT_NAME))
+ .isEqualTo("ReActAgent");
+
+ SpanData unnamed =
+ new AgentTraceSpans("test-service")
+ .assemble(List.of(toolRecord("2026-01-15T10:30:04Z",
"started")))
+ .get(1);
+
assertThat(unnamed.getAttributes().get(AgentTraceSpans.GEN_AI_AGENT_NAME)).isNull();
+ }
+
+ @Test
+ @DisplayName("A record with an invalid timestamp is skipped with a
diagnostic")
+ void testInvalidTimestampIsSkippedWithDiagnostic() {
+ List<ConverterDiagnostic> diagnostics = new ArrayList<>();
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ toolRecord("yesterday", "started"),
+ toolRecord("2026-01-15T10:30:05Z",
"success")),
+ diagnostics);
+
+ // The valid terminal record still becomes a span; the invalid start
is reported.
+ assertThat(spans).hasSize(2);
+ assertThat(diagnostics)
+ .anySatisfy(
+ d -> {
+
assertThat(d.getCode()).isEqualTo(ConverterDiagnostic.MALFORMED_RECORD);
+ assertThat(d.getMessage()).contains("yesterday");
+ });
+ }
+
+ private static TraceRecord toolRecord(String timestamp, String status) {
+ return toolRecord(timestamp, status, "");
+ }
+
+ /**
+ * {@code extraJson} adds fields after the status; empty attributes are
added if it has none.
+ */
+ private static TraceRecord toolRecord(String timestamp, String status,
String extraJson) {
+ return record(
+ "{\"schemaVersion\":1,\"timestamp\":\""
+ + timestamp
+ + "\",\"inputRunId\":\"r\",\"executionId\":\"t\","
+ + "\"entityType\":\"tool\",\"entityName\":\"lookup\","
+ + "\"status\":\""
+ + status
+ + "\""
+ + extraJson
+ + (extraJson.contains("\"attributes\"") ? "" :
",\"attributes\":{}")
+ + "}");
+ }
+
+ @Test
+ @DisplayName("created -> started -> failed spans started..failed in any
record order")
+ void testCreatedStartedTerminal() {
+ List<TraceRecord> records =
+ new ArrayList<>(
+ List.of(
+ toolRecord("2026-01-15T10:30:00Z", "created"),
+ toolRecord("2026-01-15T10:30:01Z", "started"),
+ toolRecord(
+ "2026-01-15T10:30:03Z",
+ "failed",
+
",\"problemCategory\":\"tool_call_failed\"")));
+ for (int pass = 0; pass < 2; pass++) {
+ List<ConverterDiagnostic> diagnostics = new ArrayList<>();
+ SpanData tool =
+ spanNamed(
+ new
AgentTraceSpans("test-service").assemble(records, diagnostics),
+ "execute_tool lookup");
+
+
assertThat(tool.getStatus().getStatusCode()).isEqualTo(StatusCode.ERROR);
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_EXECUTION_STATUS))
+ .isEqualTo("failed");
+ assertThat(tool.getEndEpochNanos() - tool.getStartEpochNanos())
+ .isEqualTo(2_000_000_000L);
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_EXECUTION_INCOMPLETE)).isNull();
+ assertThat(diagnostics).isEmpty();
+ Collections.reverse(records);
+ }
+ }
+
+ @Test
+ @DisplayName("A created-only execution is incomplete, not a terminal
without a start")
+ void testCreatedOnly() {
+ List<ConverterDiagnostic> diagnostics = new ArrayList<>();
+ SpanData tool =
+ spanNamed(
+ new AgentTraceSpans("test-service")
+ .assemble(
+
List.of(toolRecord("2026-01-15T10:30:00Z", "created")),
+ diagnostics),
+ "execute_tool lookup");
+
+
assertThat(tool.getStatus().getStatusCode()).isEqualTo(StatusCode.UNSET);
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_EXECUTION_INCOMPLETE)).isTrue();
+
assertThat(tool.getStartEpochNanos()).isEqualTo(tool.getEndEpochNanos());
+ assertThat(diagnostics)
+ .extracting(ConverterDiagnostic::getCode)
+ .containsExactly(ConverterDiagnostic.INCOMPLETE_EXECUTION);
+ }
+
+ @Test
+ @DisplayName("created -> failed without a start (preparation failure)
spans created..failed")
+ void testCreatedThenFailedWithoutStart() {
+ List<ConverterDiagnostic> diagnostics = new ArrayList<>();
+ SpanData tool =
+ spanNamed(
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+
toolRecord("2026-01-15T10:30:00Z", "created"),
+
toolRecord("2026-01-15T10:30:00.500Z", "failed")),
+ diagnostics),
+ "execute_tool lookup");
+
+
assertThat(tool.getStatus().getStatusCode()).isEqualTo(StatusCode.ERROR);
+ assertThat(tool.getEndEpochNanos() -
tool.getStartEpochNanos()).isEqualTo(500_000_000L);
+ assertThat(diagnostics).isEmpty();
+ }
+
+ @Test
+ @DisplayName("A STANDARD-level truncated error message still becomes the
status description")
+ void testTruncatedErrorMessage() {
+ SpanData tool =
+ spanNamed(
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ toolRecord(
+ "2026-01-15T10:30:00Z",
+ "failed",
+
",\"attributes\":{\"errorType\":\"x.Boom\","
+ +
"\"errorMessage\":{\"truncatedString\":"
+ + "\"long
message...\",\"omittedChars\":42}}"))),
+ "execute_tool lookup");
+
+ assertThat(tool.getStatus().getDescription()).isEqualTo("long
message...");
+
assertThat(tool.getAttributes().get(AgentTraceSpans.ERROR_TYPE)).isEqualTo("x.Boom");
+ }
+
+ @Test
+ @DisplayName("A reused execution becomes a point-in-time span with status
attribute 'reused'")
+ void testReusedExecution() {
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\","
+ + "\"inputRunId\":\""
+ + RUN
+ + "\","
+ + "\"executionId\":\""
+ + ACTION
+ + "\","
+ +
"\"entityType\":\"action\",\"entityName\":\"review\","
+ +
"\"status\":\"reused\"}")));
+
+ SpanData reused = spanNamed(spans, "action review");
+
assertThat(reused.getStartEpochNanos()).isEqualTo(reused.getEndEpochNanos());
+
assertThat(reused.getAttributes().get(AgentTraceSpans.FA_EXECUTION_STATUS))
+ .isEqualTo("reused");
+ // A reused execution is single-record by design: complete, no
diagnostic.
+
assertThat(reused.getAttributes().get(AgentTraceSpans.FA_EXECUTION_INCOMPLETE)).isNull();
+ }
+
+ @Test
+ @DisplayName("A started execution without a terminal record is exported as
incomplete")
+ void testIncompleteExecution() {
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\","
+ + "\"inputRunId\":\""
+ + RUN
+ + "\","
+ + "\"executionId\":\""
+ + LLM
+ + "\","
+ +
"\"entityType\":\"llm\",\"entityName\":\"chatModel\","
+ +
"\"entityMetadata\":{\"model\":\"qwen-max\"},"
+ +
"\"status\":\"started\"}")));
+
+ SpanData span = spanNamed(spans, "chat qwen-max");
+
assertThat(span.getAttributes().get(AgentTraceSpans.FA_EXECUTION_INCOMPLETE)).isTrue();
+
assertThat(span.getStartEpochNanos()).isEqualTo(span.getEndEpochNanos());
+ // UNSET, not ERROR: a missing terminal is indistinguishable between a
crash, a dropped
+ // best-effort write, and recovery discarding the pairing.
+
assertThat(span.getStatus().getStatusCode()).isEqualTo(StatusCode.UNSET);
+ }
+
+ @Test
+ @DisplayName("Incomplete and start-less executions produce
machine-readable diagnostics")
+ void testDiagnosticsForIncompleteAndMissingStart() {
+ List<ConverterDiagnostic> diagnostics = new ArrayList<>();
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\","
+ + "\"inputRunId\":\""
+ + RUN
+ + "\","
+ + "\"executionId\":\""
+ + LLM
+ + "\","
+ +
"\"entityType\":\"llm\",\"entityName\":\"chatModel\","
+ +
"\"entityMetadata\":{\"model\":\"qwen-max\"},"
+ +
"\"status\":\"started\"}"),
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:02Z\","
+ + "\"inputRunId\":\""
+ + RUN
+ + "\","
+ + "\"executionId\":\""
+ + TOOL
+ + "\","
+ +
"\"entityType\":\"tool\",\"entityName\":\"get_weather\","
+ +
"\"status\":\"success\"}")),
+ diagnostics);
+
+ assertThat(spans).hasSize(3); // run root + two executions
+ assertThat(diagnostics).hasSize(2);
+ assertThat(diagnostics)
+ .anySatisfy(
+ d -> {
+ assertThat(d.getCode())
+
.isEqualTo(ConverterDiagnostic.INCOMPLETE_EXECUTION);
+ assertThat(d.getExecutionId()).isEqualTo(LLM);
+ });
+ assertThat(diagnostics)
+ .anySatisfy(
+ d -> {
+
assertThat(d.getCode()).isEqualTo(ConverterDiagnostic.MISSING_START);
+ assertThat(d.getExecutionId()).isEqualTo(TOOL);
+ });
+ SpanData tool = spanNamed(spans, "execute_tool get_weather");
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_EXECUTION_INCOMPLETE)).isTrue();
+ }
+
+ @Test
+ @DisplayName("Parser executions map to 'parse {name}', custom operation
name, INTERNAL kind")
+ void testParserMapping() {
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\","
+ + "\"inputRunId\":\""
+ + RUN
+ + "\","
+ +
"\"executionId\":\"p-1\","
+ +
"\"entityType\":\"parser\","
+ +
"\"entityName\":\"json_output\","
+ +
"\"status\":\"started\"}"),
+ record(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:01Z\","
+ + "\"inputRunId\":\""
+ + RUN
+ + "\","
+ +
"\"executionId\":\"p-1\","
+ +
"\"entityType\":\"parser\","
+ +
"\"entityName\":\"json_output\","
+ +
"\"status\":\"success\"}")));
+
+ SpanData parser = spanNamed(spans, "parse json_output");
+ assertThat(parser.getKind()).isEqualTo(SpanKind.INTERNAL);
+
assertThat(parser.getAttributes().get(AgentTraceSpans.GEN_AI_OPERATION_NAME))
+ .isEqualTo("parse");
+ }
+
+ @Test
+ @DisplayName("Event records and execution records without a run are
ignored")
+ void testNonLifecycleRecordsIgnored() {
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+ "{\"schemaVersion\":1,"
+ +
"\"timestamp\":\"2026-01-15T10:30:00Z\","
+ +
"\"inputRunId\":\"r\",\"entityType\":\"event\","
+ +
"\"entityName\":\"_chat_request_event\","
+ +
"\"entityMetadata\":{\"eventId\":\"e-1\"},"
+ +
"\"attributes\":{\"model\":\"gpt-4\"}}"),
+ record(
+ "{\"schemaVersion\":1,"
+ +
"\"timestamp\":\"2026-01-15T10:30:00Z\","
+ +
"\"executionId\":\"x\",\"entityType\":\"tool\","
+ +
"\"entityName\":\"lookup\","
+ +
"\"status\":\"started\",\"attributes\":{}}")));
+
+ assertThat(spans).isEmpty();
+ }
+
+ @Test
+ @DisplayName(
+ "A child whose parent has no records is attached to the run root,
with a diagnostic")
+ void testMissingParentAttachesToRunRoot() {
+ List<ConverterDiagnostic> diagnostics = new ArrayList<>();
+ List<SpanData> spans =
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ toolRecord(
+ "2026-01-15T10:30:04Z",
+ "started",
+
",\"parentExecutionId\":\"act-missing\""),
+ toolRecord(
+ "2026-01-15T10:30:05Z",
+ "success",
+
",\"parentExecutionId\":\"act-missing\"")),
+ diagnostics);
+
+ SpanData root = spanNamed(spans, "invoke_agent");
+ SpanData tool = spanNamed(spans, "execute_tool lookup");
+ assertThat(tool.getParentSpanId()).isEqualTo(root.getSpanId());
+
assertThat(tool.getAttributes().get(AgentTraceSpans.FA_PARENT_EXECUTION_ID))
+ .isEqualTo("act-missing");
+ assertThat(diagnostics)
+ .singleElement()
+ .satisfies(
+ d -> {
+
assertThat(d.getCode()).isEqualTo(ConverterDiagnostic.MISSING_PARENT);
+ assertThat(d.getExecutionId()).isEqualTo("t");
+ });
+ }
+
+ @Test
+ @DisplayName("Repeated records keep the earliest of each kind, whatever
their order")
+ void testDuplicateRecordsKeepEarliest() {
+ List<TraceRecord> records =
+ new ArrayList<>(
+ List.of(
+ toolRecord("2026-01-15T10:30:01Z", "started"),
+ toolRecord("2026-01-15T10:30:02Z", "started"),
+ toolRecord("2026-01-15T10:30:04Z", "success"),
+ toolRecord("2026-01-15T10:30:06Z",
"success")));
+ for (int pass = 0; pass < 2; pass++) {
+ SpanData tool =
+ spanNamed(
+ new
AgentTraceSpans("test-service").assemble(records),
+ "execute_tool lookup");
+
+ assertThat(tool.getEndEpochNanos() - tool.getStartEpochNanos())
+ .isEqualTo(3_000_000_000L);
+ Collections.reverse(records);
+ }
+ }
+
+ @Test
+ @DisplayName("A sub-agent delegation maps to invoke_agent {name} with the
tool call id")
+ void testSubagentMapping() {
+ String delegation =
+
"\"inputRunId\":\"r\",\"executionId\":\"s\",\"entityType\":\"subagent\","
+ + "\"entityName\":\"researcher\","
+ +
"\"entityMetadata\":{\"toolCallId\":\"call_1\"},\"attributes\":{},";
+ SpanData span =
+ spanNamed(
+ new AgentTraceSpans("test-service")
+ .assemble(
+ List.of(
+ record(
+ "{\"schemaVersion\":1,"
+ + delegation
+ +
"\"timestamp\":\"2026-01-15T10:30:00Z\","
+ +
"\"status\":\"started\"}"),
+ record(
+ "{\"schemaVersion\":1,"
+ + delegation
+ +
"\"timestamp\":\"2026-01-15T10:30:02Z\","
+ +
"\"status\":\"success\"}"))),
+ "invoke_agent researcher");
+
+ assertThat(span.getKind()).isEqualTo(SpanKind.INTERNAL);
+
assertThat(span.getAttributes().get(AgentTraceSpans.GEN_AI_OPERATION_NAME))
+ .isEqualTo("invoke_agent");
+ assertThat(span.getAttributes().get(AgentTraceSpans.GEN_AI_AGENT_NAME))
+ .isEqualTo("researcher");
+
assertThat(span.getAttributes().get(AgentTraceSpans.GEN_AI_TOOL_CALL_ID))
+ .isEqualTo("call_1");
+ }
+}
diff --git
a/integrations/observability/otel/src/test/java/org/apache/flink/agents/integrations/observability/otel/TraceLogOTelExporterTest.java
b/integrations/observability/otel/src/test/java/org/apache/flink/agents/integrations/observability/otel/TraceLogOTelExporterTest.java
new file mode 100644
index 000000000..78347cf3d
--- /dev/null
+++
b/integrations/observability/otel/src/test/java/org/apache/flink/agents/integrations/observability/otel/TraceLogOTelExporterTest.java
@@ -0,0 +1,413 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.agents.integrations.observability.otel;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SpanExporter;
+import org.apache.flink.agents.api.trace.TraceContext;
+import org.apache.flink.agents.api.trace.TraceRecord;
+import org.apache.flink.agents.api.trace.TraceRecords;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** End-to-end test: Trace Log JSONL file in, spans out through an in-memory
exporter. */
+class TraceLogOTelExporterTest {
+
+ /** Collects exported spans in memory; no network involved. */
+ private static final class CollectingSpanExporter implements SpanExporter {
+ private final List<SpanData> spans = new ArrayList<>();
+ private final List<Integer> batchSizes = new ArrayList<>();
+ private int failingCall = -1;
+ private boolean shutDown;
+
+ @Override
+ public CompletableResultCode export(Collection<SpanData> collection) {
+ if (batchSizes.size() == failingCall) {
+ batchSizes.add(collection.size());
+ return CompletableResultCode.ofFailure();
+ }
+ batchSizes.add(collection.size());
+ spans.addAll(collection);
+ return CompletableResultCode.ofSuccess();
+ }
+
+ @Override
+ public CompletableResultCode flush() {
+ return CompletableResultCode.ofSuccess();
+ }
+
+ @Override
+ public CompletableResultCode shutdown() {
+ shutDown = true;
+ return CompletableResultCode.ofSuccess();
+ }
+ }
+
+ @Test
+ @DisplayName("Exports a JSONL Trace Log file, reading Event records
without making spans")
+ void testExportJsonlFile(@TempDir Path tempDir) throws Exception {
+ String run = "7f1b5d20-86c6-4a5d-b65a-9c41757f2e11";
+ Path log = tempDir.resolve("traces-job-task-0.log");
+ Files.write(
+ log,
+ List.of(
+ // An Event record: read but produces no span.
+
"{\"schemaVersion\":1,\"timestamp\":\"2026-01-15T10:29:59Z\","
+ + "\"inputRunId\":\""
+ + run
+ +
"\",\"entityType\":\"event\",\"entityName\":\"_input_event\","
+ +
"\"entityMetadata\":{\"eventId\":\"e-1\"},\"attributes\":{}}",
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\",\"inputRunId\":\""
+ + run
+ + "\","
+ +
"\"agentName\":\"ReActAgent\",\"executionId\":\"act-1\","
+ +
"\"entityType\":\"action\",\"entityName\":\"review\","
+ + "\"status\":\"started\"}",
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:02Z\",\"inputRunId\":\""
+ + run
+ + "\","
+ +
"\"executionId\":\"act-1\",\"entityType\":\"action\","
+ + "\"entityName\":\"review\","
+ + "\"status\":\"success\"}"));
+
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ TraceLogOTelExporter exporter =
+ TraceLogOTelExporter.builder()
+ .setServiceName("test-job")
+ .setSpanExporter(collector)
+ .build();
+
+ TraceLogOTelExporter.ExportSummary summary =
exporter.exportFiles(List.of(log));
+
+ assertThat(summary.getRecordsRead()).isEqualTo(3);
+ // Root run span + the action span.
+ assertThat(summary.getSpansExported()).isEqualTo(2);
+ assertThat(collector.spans).hasSize(2);
+ assertThat(collector.spans)
+ .allSatisfy(
+ span ->
+ assertThat(
+ span.getResource()
+ .getAttributes()
+ .get(
+
io.opentelemetry.api.common
+
.AttributeKey.stringKey(
+
"service.name")))
+ .isEqualTo("test-job"));
+
+ exporter.shutdown();
+ assertThat(collector.shutDown).isTrue();
+ }
+
+ @Test
+ @DisplayName("A malformed record yields a MALFORMED_RECORD diagnostic
naming the file")
+ void testMalformedRecordDiagnostic(@TempDir Path tempDir) throws Exception
{
+ Path log = tempDir.resolve("traces-job-task-0.log");
+ Files.write(
+ log,
+ List.of(
+ "{\"timestamp\": not-valid",
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\",\"inputRunId\":\"run-1\","
+ +
"\"executionId\":\"act-1\",\"entityType\":\"action\","
+ + "\"entityName\":\"review\","
+ + "\"status\":\"started\"}"));
+
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ TraceLogOTelExporter exporter =
+ TraceLogOTelExporter.builder()
+ .setServiceName("test-job")
+ .setSpanExporter(collector)
+ .build();
+
+ TraceLogOTelExporter.ExportSummary summary =
exporter.exportFiles(List.of(log));
+
+ assertThat(summary.getDiagnostics())
+ .anySatisfy(
+ d -> {
+
assertThat(d.getCode()).isEqualTo(ConverterDiagnostic.MALFORMED_RECORD);
+
assertThat(d.getFilePath()).isEqualTo(log.toString());
+ });
+ exporter.shutdown();
+ }
+
+ @Test
+ @DisplayName("A file that is not JSON from its first byte does not stop
the other files")
+ void testUnparseableFileDoesNotStopOtherFiles(@TempDir Path tempDir)
throws Exception {
+ Path broken = tempDir.resolve("traces-job-task-0.log");
+ Files.write(broken, List.of("not-valid"));
+ Path valid = tempDir.resolve("traces-job-task-1.log");
+ String action =
+
"{\"schemaVersion\":1,\"attributes\":{},\"inputRunId\":\"run-1\","
+ +
"\"executionId\":\"act-1\",\"entityType\":\"action\","
+ + "\"entityName\":\"review\",";
+ Files.write(
+ valid,
+ List.of(
+ action +
"\"timestamp\":\"2026-01-15T10:30:00Z\",\"status\":\"started\"}",
+ action +
"\"timestamp\":\"2026-01-15T10:30:01Z\",\"status\":\"success\"}"));
+
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ TraceLogOTelExporter exporter =
+
TraceLogOTelExporter.builder().setSpanExporter(collector).build();
+ TraceLogOTelExporter.ExportSummary summary =
exporter.exportFiles(List.of(broken, valid));
+
+ assertThat(summary.getRecordsRead()).isEqualTo(2);
+ assertThat(collector.spans).hasSize(2);
+ assertThat(summary.getDiagnostics())
+ .singleElement()
+ .satisfies(
+ d -> {
+
assertThat(d.getCode()).isEqualTo(ConverterDiagnostic.MALFORMED_RECORD);
+
assertThat(d.getFilePath()).isEqualTo(broken.toString());
+ });
+ exporter.shutdown();
+ }
+
+ @Test
+ @DisplayName("A directory argument is expanded to its traces-*.log and
*.trace-log.log files")
+ void testDirectoryDiscovery(@TempDir Path tempDir) throws Exception {
+ Files.write(
+ tempDir.resolve("traces-job-task-0.log"),
+ List.of(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\",\"inputRunId\":\"run-1\","
+ +
"\"executionId\":\"act-1\",\"entityType\":\"action\","
+ + "\"entityName\":\"review\","
+ + "\"status\":\"started\"}"));
+ Files.write(
+ tempDir.resolve("job-task-1.trace-log.log"),
+ List.of(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:01Z\",\"inputRunId\":\"run-2\","
+ +
"\"executionId\":\"act-2\",\"entityType\":\"action\","
+ + "\"entityName\":\"review\","
+ + "\"status\":\"started\"}"));
+ // Not matching the File trace logger naming contract: ignored during
discovery.
+ Files.write(tempDir.resolve("other.txt"), List.of("not a log"));
+
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ TraceLogOTelExporter exporter =
+ TraceLogOTelExporter.builder()
+ .setServiceName("test-job")
+ .setSpanExporter(collector)
+ .build();
+
+ TraceLogOTelExporter.ExportSummary summary =
exporter.exportFiles(List.of(tempDir));
+
+ assertThat(summary.getRecordsRead()).isEqualTo(2);
+ // Two runs, each with a root span and one (incomplete) action span.
+ assertThat(summary.getSpansExported()).isEqualTo(4);
+ exporter.shutdown();
+ }
+
+ /** Two runs, each with a finished action: four spans. */
+ private static Path twoRunLog(Path tempDir) throws Exception {
+ Path log = tempDir.resolve("traces-job-task-0.log");
+ List<String> lines = new ArrayList<>();
+ for (String run : List.of("run-1", "run-2")) {
+ for (String event : List.of("started", "finished")) {
+ lines.add(
+
"{\"schemaVersion\":1,\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:0"
+ + (event.equals("started") ? "0" : "1")
+ + "Z\",\"inputRunId\":\""
+ + run
+ + "\",\"executionId\":\"act-"
+ + run
+ +
"\",\"entityType\":\"action\",\"entityName\":\"review\","
+ + "\"status\":\""
+ + (event.equals("started") ? "started" :
"success")
+ + "\"}");
+ }
+ }
+ Files.write(log, lines);
+ return log;
+ }
+
+ @Test
+ @DisplayName("Spans are exported in batches of at most the configured
size")
+ void testExportsInBoundedBatches(@TempDir Path tempDir) throws Exception {
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ TraceLogOTelExporter exporter =
+ TraceLogOTelExporter.builder()
+ .setSpanExporter(collector)
+ .setMaxExportBatchSize(3)
+ .build();
+
+ TraceLogOTelExporter.ExportSummary summary =
+ exporter.exportFiles(List.of(twoRunLog(tempDir)));
+
+ assertThat(summary.getSpansExported()).isEqualTo(4);
+ assertThat(collector.batchSizes).containsExactly(3, 1);
+ exporter.shutdown();
+ }
+
+ @Test
+ @DisplayName("A failed batch reports how many spans the earlier batches
delivered")
+ void testFailedBatchReportsPartialProgress(@TempDir Path tempDir) throws
Exception {
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ collector.failingCall = 1;
+ TraceLogOTelExporter exporter =
+ TraceLogOTelExporter.builder()
+ .setSpanExporter(collector)
+ .setMaxExportBatchSize(3)
+ .build();
+
+ assertThatThrownBy(() ->
exporter.exportFiles(List.of(twoRunLog(tempDir))))
+ .isInstanceOfSatisfying(
+ TraceLogOTelExporter.ExportFailedException.class,
+ e -> {
+ assertThat(e.getSpansExported()).isEqualTo(3);
+ assertThat(e.getSpansTotal()).isEqualTo(4);
+ });
+ assertThat(collector.spans).hasSize(3);
+ exporter.shutdown();
+ }
+
+ @Test
+ @DisplayName("Rejects a non-positive export batch size")
+ void testRejectsNonPositiveBatchSize() {
+ assertThatThrownBy(() ->
TraceLogOTelExporter.builder().setMaxExportBatchSize(0))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ @DisplayName("Rejects unsupported OTLP protocols at build time")
+ void testUnsupportedProtocol() {
+ try {
+
TraceLogOTelExporter.builder().setProtocol("carrier-pigeon").build();
+ throw new AssertionError("expected IllegalArgumentException");
+ } catch (IllegalArgumentException expected) {
+ assertThat(expected).hasMessageContaining("carrier-pigeon");
+ }
+ }
+
+ @Test
+ @DisplayName("Records written by the Trace Log serializer export as one
parented trace")
+ void testReadsSerializedTraceRecords(@TempDir Path tempDir) throws
Exception {
+ TraceContext action =
+ TraceContext.forAction(
+ TraceContext.forInputRun("order-1001", "ReActAgent"),
"review", "input");
+ TraceContext llm =
+ action.childExecution("llm-1", "llm", "chatModel",
Map.of("model", "qwen-max"));
+ TraceContext llmDone =
+ action.childExecution(
+ "llm-1",
+ "llm",
+ "chatModel",
+ Map.of(
+ "model",
+ "qwen-max",
+ "usage",
+ Map.of("promptTokens", 120,
"completionTokens", 45)));
+ ObjectMapper mapper = new ObjectMapper();
+ List<String> lines = new ArrayList<>();
+ for (TraceRecord record :
+ List.of(
+ TraceRecords.started(action),
+ TraceRecords.started(llm),
+ TraceRecords.succeeded(llmDone),
+ TraceRecords.succeeded(action))) {
+ lines.add(mapper.writeValueAsString(record));
+ }
+ Path log = tempDir.resolve("traces-job-task-0.log");
+ Files.write(log, lines);
+
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ TraceLogOTelExporter exporter =
+
TraceLogOTelExporter.builder().setSpanExporter(collector).build();
+ TraceLogOTelExporter.ExportSummary summary =
exporter.exportFiles(List.of(log));
+
+ assertThat(summary.getDiagnostics()).isEmpty();
+ assertThat(collector.spans).hasSize(3);
+ SpanData actionSpan = spanNamed(collector.spans, "action review");
+ SpanData chatSpan = spanNamed(collector.spans, "chat qwen-max");
+
assertThat(chatSpan.getParentSpanId()).isEqualTo(actionSpan.getSpanId());
+
assertThat(chatSpan.getAttributes().get(AgentTraceSpans.GEN_AI_USAGE_INPUT_TOKENS))
+ .isEqualTo(120L);
+
assertThat(chatSpan.getAttributes().get(AgentTraceSpans.GEN_AI_USAGE_OUTPUT_TOKENS))
+ .isEqualTo(45L);
+ exporter.shutdown();
+ }
+
+ @Test
+ @DisplayName("A record without schemaVersion is malformed; reading
continues with the next")
+ void testMissingSchemaVersionIsMalformed(@TempDir Path tempDir) throws
Exception {
+ Path log = tempDir.resolve("traces-job-task-0.log");
+ String started =
+ "\"attributes\":{},\"timestamp\":\"2026-01-15T10:30:00Z\","
+ + "\"inputRunId\":\"run-1\",\"executionId\":\"act-1\","
+ +
"\"entityType\":\"action\",\"entityName\":\"review\","
+ + "\"status\":\"started\"}";
+ Files.write(log, List.of("{" + started, "{\"schemaVersion\":1," +
started));
+
+ TraceLogOTelExporter exporter =
+ TraceLogOTelExporter.builder()
+ .setSpanExporter(new CollectingSpanExporter())
+ .build();
+ TraceLogOTelExporter.ExportSummary summary =
exporter.exportFiles(List.of(log));
+
+ assertThat(summary.getRecordsRead()).isEqualTo(1);
+ assertThat(summary.getDiagnostics())
+ .anySatisfy(
+ d ->
+ assertThat(d.getCode())
+
.isEqualTo(ConverterDiagnostic.MALFORMED_RECORD));
+ exporter.shutdown();
+ }
+
+ @Test
+ @DisplayName("Fields the File and SLF4J trace loggers add around a record
are ignored")
+ void testLoggerFieldsAreIgnored(@TempDir Path tempDir) throws Exception {
+ Path log = tempDir.resolve("traces-job-task-0.log");
+ Files.write(
+ log,
+ List.of(
+
"{\"schemaVersion\":1,\"level\":\"STANDARD\",\"jobId\":\"j\","
+ +
"\"taskName\":\"t\",\"subtaskId\":0,\"attributes\":{},"
+ +
"\"timestamp\":\"2026-01-15T10:30:00Z\",\"inputRunId\":\"run-1\","
+ +
"\"executionId\":\"act-1\",\"entityType\":\"action\","
+ +
"\"entityName\":\"review\",\"status\":\"started\"}"));
+
+ CollectingSpanExporter collector = new CollectingSpanExporter();
+ TraceLogOTelExporter exporter =
+
TraceLogOTelExporter.builder().setSpanExporter(collector).build();
+ TraceLogOTelExporter.ExportSummary summary =
exporter.exportFiles(List.of(log));
+
+ assertThat(summary.getRecordsRead()).isEqualTo(1);
+ assertThat(collector.spans).hasSize(2);
+ exporter.shutdown();
+ }
+
+ private static SpanData spanNamed(List<SpanData> spans, String name) {
+ return spans.stream()
+ .filter(s -> s.getName().equals(name))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("no span named '" + name
+ "'"));
+ }
+}
diff --git a/integrations/pom.xml b/integrations/observability/pom.xml
similarity index 64%
copy from integrations/pom.xml
copy to integrations/observability/pom.xml
index e691f2508..c80fc134e 100644
--- a/integrations/pom.xml
+++ b/integrations/observability/pom.xml
@@ -7,7 +7,9 @@ regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
+
http://www.apache.org/licenses/LICENSE-2.0
+
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -22,29 +24,16 @@ under the License.
<parent>
<groupId>org.apache.flink</groupId>
- <artifactId>flink-agents</artifactId>
+ <artifactId>flink-agents-integrations</artifactId>
<version>1.0-SNAPSHOT</version>
</parent>
- <artifactId>flink-agents-integrations</artifactId>
- <name>Flink Agents : Integrations:</name>
+ <artifactId>flink-agents-integrations-observability</artifactId>
+ <name>Flink Agents : Integrations: Observability:</name>
<packaging>pom</packaging>
- <properties>
- <ollama4j.version>1.1.5</ollama4j.version>
- <elasticsearch.version>8.19.0</elasticsearch.version>
- <milvus.version>2.6.18</milvus.version>
- <openai.version>4.8.0</openai.version>
- <anthropic.version>2.12.0</anthropic.version>
- <aws.sdk.version>2.41.22</aws.sdk.version>
- <google.genai.version>1.56.0</google.genai.version>
- </properties>
-
<modules>
- <module>chat-models</module>
- <module>embedding-models</module>
- <module>vector-stores</module>
- <module>mcp</module>
+ <module>otel</module>
</modules>
</project>
diff --git a/integrations/pom.xml b/integrations/pom.xml
index e691f2508..aa87c4e22 100644
--- a/integrations/pom.xml
+++ b/integrations/pom.xml
@@ -38,6 +38,7 @@ under the License.
<anthropic.version>2.12.0</anthropic.version>
<aws.sdk.version>2.41.22</aws.sdk.version>
<google.genai.version>1.56.0</google.genai.version>
+ <opentelemetry.version>1.51.0</opentelemetry.version>
</properties>
<modules>
@@ -45,6 +46,7 @@ under the License.
<module>embedding-models</module>
<module>vector-stores</module>
<module>mcp</module>
+ <module>observability</module>
</modules>
</project>
