joeyutong commented on code in PR #1160: URL: https://github.com/apache/flink-agents/pull/1160#discussion_r4293365973
########## docs/content/docs/operations/monitoring.md: ########## @@ -401,6 +401,60 @@ Observations with the same Event ID and type share one node and deduplicated edg Malformed records, conflicting Event identities, missing upstream Events, and cycles produce warnings. Recoverable records remain available in the JSON result, and cycle edges are removed before rendering. If recording targets filtered an upstream Event, the reader cannot reconstruct that missing part of the tree. Historical formats using `eventType`/`eventAttributes` or a nested `event` object are unsupported. +### 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`, and `parser`. + +#### 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` | +| `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 responsibility; the summary reports what was read. + +**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.execution.status=reused`. When a record is repeated, the earliest one of each kind is kept. Undecodable records and records with an unparseable timestamp are skipped with a `MALFORMED_RECORD` diagnostic. + +**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 and its dependencies> \ Review Comment: I tried the CLI end to end with Jaeger. Using Maven's runtime dependency classpath gave `NoClassDefFoundError: org/slf4j/LoggerFactory`; adding `slf4j-api` allowed both gRPC and HTTP/protobuf exports to succeed. Could we mention this classpath requirement near the CLI example? ########## integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/TraceLogOTelExporter.java: ########## @@ -0,0 +1,426 @@ +/* + * 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.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 = + MAPPER.readerFor(TraceRecord.class).readValues(in)) { Review Comment: Could we also isolate parsing errors raised by `readValues(in)` to the current file? I tried a file containing `not-valid` followed by a valid file; this throws `JsonParseException` here and aborts the whole export, rather than continuing with the valid file. -- 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]
