Zhuoxi2000 opened a new pull request, #1160: URL: https://github.com/apache/flink-agents/pull/1160
<html><head></head><body><p>Linked issue: #970</p><h3>Purpose of change</h3><p>An Agent Trace Event Log (<code inline="">event-log.trace.enabled: true</code>) from a Java or Python agent can now be exported to any OTLP backend as OpenTelemetry GenAI traces:</p><pre><code class="language-text">EventLogOTelExporter [--endpoint URL] [--protocol grpc|http/protobuf] [--service-name NAME] <file-or-dir>... </code></pre><p>This is a new optional module, <code inline="">flink-agents-integrations-observability-otel</code>, and is not bundled into <code inline="">dist</code>. The goal is to make the execution hierarchy already recorded in Agent Trace visible in standard tracing tools without adding runtime overhead.</p><h4>Runtime flow</h4><ol><li><p><code inline="">exportFiles</code> expands directories to their <code inline="">events-*.log</code> files and streams JSON records into <code inline="">TraceRecord</code> (unknown fields are ignored).</p></li><li><p><code inline="">AgentTraceSpans.assemble</code> keeps lifecycle records with <code inline="">executionId</code>, <code inline="">inputRunId</code>, and <code inline="">timestamp</code>, grouped by execution and input run.</p></li><li><p>Each input run becomes a trace with a synthesized <code inline="">invoke_agent</code> root. Each execution becomes a span under its <code inline="">parentExecutionId</code>, or under the root if no parent is present, based on <code inline="">entityType</code>.</p></li><li><p>All spans are passed to the OTLP <code inline="">SpanExporter</code> in one call. Diagnostics are logged and returned in <code inline="">ExportSummary</code>.</p></li></ol><h4>Key decisions</h4><ul><li><p><strong>Batch conversion from the Event Log</strong>, rather than an in-job exporter: no runtime overhead, and the same input format works for both Java and Python. Continuous consumption is out of scope.</p></li><li><p><strong>Deterministic IDs</strong> (SHA-256 of <code inline="">inputRunId</code> / <code inline="">executionId</code>): re-exporting the same records produces the same IDs; delivery remains at-least-once.</p></li><li><p><strong>Incomplete executions use status UNSET</strong>, not ERROR: a missing terminal record may come from a crash, dropped write, or recovery, and the log alone cannot distinguish them.</p></li><li><p><code inline="">gen_ai.request.model</code> comes from <code inline= "">entityMetadata.model</code> (<code inline="">entityName</code> is the ChatModel resource). <code inline="">gen_ai.provider.name</code> is left unset because the record does not contain it.</p></li><li><p><code inline="">execute_tool</code> spans remain INTERNAL regardless of the tool transport.</p></li></ul><h3>Behavioral Semantics</h3><h4>Interaction decisions</h4><p><code inline="">started</code> is used as the start record when present, otherwise <code inline="">created</code>; <code inline="">created</code> is never treated as a terminal record.</p> Records for one execution | Span | Status | Diagnostic -- | -- | -- | -- start + finished | start → terminal | UNSET | none start + failed | start → terminal | ERROR + error.type | none start only (incl. created only) | zero-length, incomplete=true | UNSET | INCOMPLETE_EXECUTION terminal only, not reused | zero-length, incomplete=true | from terminal | MISSING_START reused only | zero-length, status=reused | UNSET | none <p>Span mapping is covered for each entity type and lifecycle case; input handling uses real files.</p><p>Not verified:</p><ul><li><p>A real OTLP endpoint: tests use an in-memory <code inline="">SpanExporter</code>, so transport, the 30 s timeout, and export failures are not covered.</p></li><li><p>The 1,000-malformed cap, broken-stream abort, and CLI exit codes (manual only).</p></li><li><p>A Python-produced log (same format, no fixture).</p></li><li><p>Very large logs: one invocation currently holds all records in memory.</p></li></ul><ul><li><p>Test fixtures match the shape written by <code inline="">EventLogRecordJsonSerializer</code>: <code inline="">entityMetadata</code> is an object, <code inline="">status</code> / <code inline="">problemCategory</code> are top-level fields, and <code inline="">errorType</code> / <code inline="">errorMessage</code> are inside <code inline="">eventAttributes</code>. A failed-tool record and an LLM record produced by the real serializer (<cod e inline="">EventLogRecord</code> + <code inline="">ExecutionLifecycleEvents</code>) assemble into the same spans as the fixtures.</p></li><li><p><code inline="">OTelIds</code>: trace id = first 16 bytes of SHA-256(<code inline="">inputRunId</code>), span id = first 8 bytes of SHA-256(<code inline="">executionId</code>); the run root hashes <code inline="">inputRunId</code> with a domain suffix so it cannot collide with an execution span id; an all-zero id is adjusted to remain valid.</p></li><li><p>Attribute keys are pinned as literals rather than taken from the incubating semconv artifact, so the exported schema changes only with a Flink Agents release.</p></li><li><p>OpenTelemetry artifacts are aligned through <code inline="">opentelemetry-bom</code> (all resolve to 1.51.0).</p></li><li><p>Verified locally: module tests 22/22, <code inline="">spotless:check</code>, and <code inline="">verify</code> for the module and its upstream modules.</p></li></ul><h3>API</h3><p>No existing A PI, record format, or runtime behavior changes. <code inline="">dist</code> is unchanged, so nothing is added to a job's classpath.</p><p>New public entry points: <code inline="">EventLogOTelExporter</code> (builder and <code inline="">main</code>), <code inline="">AgentTraceSpans</code>, and <code inline="">TraceRecord</code>.</p><p>Existing trace-enabled Event Logs are consumed as-is. The GenAI conventions are still at development stability, so the exported attribute set is pinned per release.</p><h3>Documentation</h3><ul class="contains-task-list"><li class="task-list-item"><p><input type="checkbox" disabled=""> <code inline="">doc-needed</code></p></li><li class="task-list-item"><p><input type="checkbox" disabled=""> <code inline="">doc-not-needed</code></p></li><li class="task-list-item"><p><input type="checkbox" checked="" disabled=""> <code inline="">doc-included</code></p></li></ul><h3>Was this patch authored or co-authored using generative AI tooling?</h3><ul class="contain s-task-list"><li class="task-list-item"><p><input type="checkbox" checked="" disabled=""> Yes</p></li><li class="task-list-item"><p><input type="checkbox" disabled=""> No</p></li></ul><p>If yes, include a <code inline="">Generated-by: <tool name and version> (<model name and version>)</code> line, for example <code inline="">Generated-by: Claude Code 2.1.226 (Claude Opus 4.6)</code>, in the commit message so it reaches Git history. Repeat the same line here for reviewer visibility. See the <a href="https://www.apache.org/legal/generative-tooling.html";>ASF generative tooling guidance</a>.</p><p>Generated-by: Claude Code 2.1.259 (Claude Fable 5, Claude Opus 5.5)</p></body></html><!-- * Thank you very much for contributing to Flink Agents. * Please add the relevant components in the PR title. E.g., [api], [runtime], [java], [python], [hotfix], etc. --> <!-- Please link the PR to the relevant issue(s). Hotfix doesn't need this. --> Linked issue: #970 ### Purpose of change <!-- What is the purpose of this change? --> An Agent Trace Event Log (`event-log.trace.enabled: true`) from a Java or Python agent can now be exported to any OTLP backend as OpenTelemetry GenAI traces: ```text EventLogOTelExporter [--endpoint URL] [--protocol grpc|http/protobuf] [--service-name NAME] <file-or-dir>... ``` This is a new optional module, `flink-agents-integrations-observability-otel`, and is not bundled into `dist`. The goal is to make the execution hierarchy already recorded in Agent Trace visible in standard tracing tools without adding runtime overhead. #### Runtime flow 1. `exportFiles` expands directories to their `events-*.log` files and streams JSON records into `TraceRecord` (unknown fields are ignored). 2. `AgentTraceSpans.assemble` keeps lifecycle records with `executionId`, `inputRunId`, and `timestamp`, grouped by execution and input run. 3. Each input run becomes a trace with a synthesized `invoke_agent` root. Each execution becomes a span under its `parentExecutionId`, or under the root if no parent is present, based on `entityType`. 4. All spans are passed to the OTLP `SpanExporter` in one call. Diagnostics are logged and returned in `ExportSummary`. #### Key decisions * **Batch conversion from the Event Log**, rather than an in-job exporter: no runtime overhead, and the same input format works for both Java and Python. Continuous consumption is out of scope. * **Deterministic IDs** (SHA-256 of `inputRunId` / `executionId`): re-exporting the same records produces the same IDs; delivery remains at-least-once. * **Incomplete executions use status UNSET**, not ERROR: a missing terminal record may come from a crash, dropped write, or recovery, and the log alone cannot distinguish them. * `gen_ai.request.model` comes from `entityMetadata.model` (`entityName` is the ChatModel resource). `gen_ai.provider.name` is left unset because the record does not contain it. * `execute_tool` spans remain INTERNAL regardless of the tool transport. ### Behavioral Semantics <!-- For a non-trivial code change whose implementation is largely AI-assisted: interaction decisions, behavioral contracts, and failure behavior. See `contribution-guides/ai-assisted-pr.md`. Remove this heading and this comment otherwise. --> #### Interaction decisions `started` is used as the start record when present, otherwise `created`; `created` is never treated as a terminal record. | Records for one execution | Span | Status | Diagnostic | | --------------------------------- | ------------------------------ | -------------------- | ---------------------- | | start + `finished` | start → terminal | UNSET | none | | start + `failed` | start → terminal | ERROR + `error.type` | none | | start only (incl. `created` only) | zero-length, `incomplete=true` | UNSET | `INCOMPLETE_EXECUTION` | | terminal only, not `reused` | zero-length, `incomplete=true` | from terminal | `MISSING_START` | | `reused` only | zero-length, `status=reused` | UNSET | none | #### Behavioral contracts 1. One trace per `inputRunId`, rooted at `invoke_agent {agentName}` (INTERNAL), with `gen_ai.agent.name` and, when present, `gen_ai.conversation.id = businessKey`. 2. An execution span's parent is its `parentExecutionId` span; otherwise it is attached to the run root. 3. `llm` → `chat {model}` (CLIENT): `gen_ai.request.model` from `entityMetadata.model`, and `gen_ai.usage.*_tokens` from `promptTokens` / `completionTokens`. Without a model, the span is named `chat` and no request model is set. 4. `tool` → `execute_tool {name}` (INTERNAL): `gen_ai.tool.name`; `gen_ai.tool.call.id` = `externalId`, otherwise `toolCallId`; `gen_ai.tool.type` = `function`, `extension` (`remote_function`, `mcp`), or unset (`model_built_in`). The raw value is kept in `flink_agents.tool.type`. 5. `action` → `action {name}`; `parser` → `parse {name}` with `gen_ai.operation.name=parse`; both are INTERNAL. 6. A failed execution has ERROR status with the recorded error message, and `error.type` = the recorded error type, falling back to `problemCategory`. 7. Execution spans carry `flink_agents.{input_run_id, execution_id, entity_type, entity_name, execution.status}`. 8. Exporting the same records again, in any order, produces the same spans and IDs. 9. Non-lifecycle records, and records missing `executionId`, `inputRunId`, or `timestamp`, are skipped. #### Failure behavior * An undecodable record produces a `MALFORMED_RECORD` diagnostic naming the file. Reading continues, but a broken stream or 1,000 malformed records stop that file. * OTLP export that does not succeed within 30 s throws `IllegalStateException`; there is no retry. * Unsupported `--protocol` throws `IllegalArgumentException` at build time. * CLI: a flag without a value, or no input, prints usage and exits 2; a missing input path throws `NoSuchFileException`. ### Tests <!-- How is this change verified? --> | Contract | Tests | | -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | 1 | `AgentTraceSpansTest.testRunBecomesSingleTraceWithRootSpan` | | 2 | `testParenting`, `testDiagnosticsForIncompleteAndMissingStart` | | 3 | `testGenAiAttributes`, `testChatWithoutModel` | | 4 | `testGenAiAttributes`, `testToolTypeMapping` | | 5 | `testParserMapping`, `testParenting` | | 6 | `testGenAiAttributes`, `testErrorTypeFallsBackToProblemCategory`, `testTruncatedErrorMessage` | | 7 | `testCorrelationAttributes` | | 8 | `testDeterministicIds`, `testRecordOrderDoesNotMatter` | | 9 | `testNonLifecycleRecordsIgnored`, `EventLogOTelExporterTest.testExportJsonlFile` | | Interaction table | `testCreatedStartedTerminal`, `testCreatedOnly`, `testCreatedThenFailedWithoutStart`, `testIncompleteExecution`, `testDiagnosticsForIncompleteAndMissingStart`, `testReusedExecution` | | Malformed input, protocol, directories | `testMalformedRecordDiagnostic`, `testUnsupportedProtocol`, `testDirectoryDiscovery` | Span mapping is covered for each entity type and lifecycle case; input handling uses real files. Not verified: * A real OTLP endpoint: tests use an in-memory `SpanExporter`, so transport, the 30 s timeout, and export failures are not covered. * The 1,000-malformed cap, broken-stream abort, and CLI exit codes (manual only). * A Python-produced log (same format, no fixture). * Very large logs: one invocation currently holds all records in memory. <details> <summary>Implementation invariants and supporting evidence</summary> * Test fixtures match the shape written by `EventLogRecordJsonSerializer`: `entityMetadata` is an object, `status` / `problemCategory` are top-level fields, and `errorType` / `errorMessage` are inside `eventAttributes`. A failed-tool record and an LLM record produced by the real serializer (`EventLogRecord` + `ExecutionLifecycleEvents`) assemble into the same spans as the fixtures. * `OTelIds`: trace id = first 16 bytes of SHA-256(`inputRunId`), span id = first 8 bytes of SHA-256(`executionId`); the run root hashes `inputRunId` with a domain suffix so it cannot collide with an execution span id; an all-zero id is adjusted to remain valid. * Attribute keys are pinned as literals rather than taken from the incubating semconv artifact, so the exported schema changes only with a Flink Agents release. * OpenTelemetry artifacts are aligned through `opentelemetry-bom` (all resolve to 1.51.0). * Verified locally: module tests 22/22, `spotless:check`, and `verify` for the module and its upstream modules. </details> ### API <!-- Does this change touches any public APIs? --> No existing API, record format, or runtime behavior changes. `dist` is unchanged, so nothing is added to a job's classpath. New public entry points: `EventLogOTelExporter` (builder and `main`), `AgentTraceSpans`, and `TraceRecord`. Existing trace-enabled Event Logs are consumed as-is. The GenAI conventions are still at development stability, so the exported attribute set is pinned per release. ### Documentation <!-- Do not remove this section. Check the proper box only. --> * [ ] `doc-needed` <!-- Your PR changes impact docs --> * [ ] `doc-not-needed` <!-- Your PR changes do not impact docs --> * [x] `doc-included` <!-- Your PR already contains the necessary documentation updates --> ### Was this patch authored or co-authored using generative AI tooling? <!-- Do not remove this section. Check the proper box only. --> * [x] Yes * [ ] No If yes, include a `Generated-by: <tool name and version> (<model name and version>)` line, for example `Generated-by: Claude Code 2.1.226 (Claude Opus 4.6)`, in the commit message so it reaches Git history. Repeat the same line here for reviewer visibility. See the [[ASF generative tooling guidance](https://www.apache.org/legal/generative-tooling.html)](https://www.apache.org/legal/generative-tooling.html). Generated-by: Claude Code 2.1.259 (Claude Fable 5, Claude Opus 5.5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
