Zhuoxi2000 commented on code in PR #1160: URL: https://github.com/apache/flink-agents/pull/1160#discussion_r4118413378
########## integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/EventLogOTelExporter.java: ########## @@ -0,0 +1,333 @@ +/* + * 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.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.concurrent.TimeUnit; + +/** + * Exports Agent Trace Event Logs as OpenTelemetry GenAI traces. + * + * <p>This is the out-of-band exporter agreed in the Agent Trace design discussions: it reads the + * Event Log written by the File/SLF4J event loggers (with {@code event-log.trace.enabled: true}), + * assembles spans per {@link AgentTraceSpans}, and pushes them to any OTLP endpoint. 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 Event Logger, and also + * its pretty-printed variant (the parser consumes concatenated JSON objects regardless of line + * breaks). + * + * <p>Example usage: + * + * <pre>{@code + * EventLogOTelExporter exporter = + * EventLogOTelExporter.builder() + * .setEndpoint("http://localhost:4317";) + * .setProtocol("grpc") + * .setServiceName("my-agent-job") + * .build(); + * exporter.exportFiles(List.of(Path.of("/tmp/flink-agents/events-<jobId>-<task>-0.log"))); + * exporter.shutdown(); + * }</pre> + */ +public class EventLogOTelExporter implements AutoCloseable { + + private static final Logger LOG = LoggerFactory.getLogger(EventLogOTelExporter.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; + + private final AgentTraceSpans assembler; + private final SpanExporter spanExporter; + + EventLogOTelExporter(String serviceName, SpanExporter spanExporter) { + this.assembler = new AgentTraceSpans(serviceName); + this.spanExporter = spanExporter; + } + + public static Builder builder() { + return new Builder(); + } + + /** + * Reads the given Event Log files and exports the assembled spans. + * + * <p>A directory argument is expanded to the {@code events-*.log} files it contains (the + * FileEventLogger naming contract, matching {@code trace_tree.py}); a file argument is consumed + * as-is regardless of its name, so records collected from any Event 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)) { + 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( + new ConverterDiagnostic( + ConverterDiagnostic.MALFORMED_RECORD, + null, + "Could not decode an Event Log record: " + e.getMessage(), + file.toString())); + malformedInFile++; + fileAborted = malformedInFile >= MAX_MALFORMED_PER_FILE || !canContinue(it); + } + } + } + } + return export(records, diagnostics); + } + + 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)) { + List<Path> discovered = new ArrayList<>(); + try (DirectoryStream<Path> stream = + Files.newDirectoryStream(path, "events-*.log")) { + stream.forEach(discovered::add); + } + Collections.sort(discovered); + 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); + if (!spans.isEmpty()) { + CompletableResultCode result = spanExporter.export(spans); Review Comment: Done in 36e9717: 512-span batches, 30 s per batch, and failures report delivered spans. ########## integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java: ########## @@ -0,0 +1,427 @@ +/* + * 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.ExecutionLifecycleEvents; +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 java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Assembles OpenTelemetry spans from Agent Trace Event 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} (falling back to the run root); + * <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 Event 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_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 String INSTRUMENTATION_SCOPE_NAME = "org.apache.flink.agents.otel"; + + private final Resource resource; + private final InstrumentationScopeInfo scope; + + public AgentTraceSpans(String serviceName) { + this.resource = + Resource.getDefault().toBuilder() + .put(AttributeKey.stringKey("service.name"), serviceName) + .build(); + this.scope = InstrumentationScopeInfo.create(INSTRUMENTATION_SCOPE_NAME); + } + + /** Assembles spans from Event 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) 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) { + String eventType = record.getEventType(); + if (eventType == null + || !ExecutionLifecycleEvents.isExecutionLifecycleEvent(eventType)) { + continue; + } + if (record.getExecutionId() == null + || record.getInputRunId() == null + || record.getTimestamp() == null) { + continue; + } + executions + .computeIfAbsent(record.getExecutionId(), id -> new ExecutionSpanBuilder()) + .accept(record); + runs.computeIfAbsent(record.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, 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); + } + if (run.businessKey != null) { + attributes.put(GEN_AI_CONVERSATION_ID, 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, List<ConverterDiagnostic> diagnostics) { + TraceRecord any = execution.anyRecord(); + String inputRunId = any.getInputRunId(); + String entityType = any.getEntityType(); + String entityName = any.getEntityName() != null ? any.getEntityName() : "unknown"; + + AttributesBuilder attributes = Attributes.builder(); + attributes.put(FA_INPUT_RUN_ID, inputRunId); + attributes.put(FA_EXECUTION_ID, any.getExecutionId()); + if (entityType != null) { + attributes.put(FA_ENTITY_TYPE, entityType); + } + attributes.put(FA_ENTITY_NAME, entityName); + if (any.getBusinessKey() != null) { + attributes.put(GEN_AI_CONVERSATION_ID, 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); + // Prefer the provider-issued call id, which is what the model's tool-call request + // carries; the framework-assigned id is the fallback. + String callId = stringValue(metadata, ToolExecutionMetadataKeys.EXTERNAL_ID); + if (callId == null) { + 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.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 != null ? entityType + " " : "") + entityName; + kind = SpanKind.INTERNAL; + } + + StatusData status = StatusData.unset(); + TraceRecord terminal = execution.terminal; + if (terminal != null) { + if (ExecutionLifecycleEvents.STATUS_FAILED.equals(terminal.getStatus())) { + status = + StatusData.create( + StatusCode.ERROR, + terminal.getErrorMessage() != null + ? terminal.getErrorMessage() + : ""); + String errorType = + terminal.getErrorType() != null + ? terminal.getErrorType() + : terminal.getProblemCategory(); + if (errorType != null) { + attributes.put(ERROR_TYPE, errorType); + } + } + if (terminal.getStatus() != null) { + attributes.put(FA_EXECUTION_STATUS, terminal.getStatus()); + } + putUsageIfPresent(attributes, terminal.getEventAttributes()); + } + boolean reused = + terminal != null + && ExecutionLifecycleEvents.STATUS_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; + + SpanContext parent = + any.getParentExecutionId() != null + ? spanContext(inputRunId, OTelIds.spanId(any.getParentExecutionId())) + : spanContext(inputRunId, OTelIds.runRootSpanId(inputRunId)); + + 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 static String stringValue(Map<String, Object> map, String key) { + Object value = map.get(key); + return value == null ? null : String.valueOf(value); + } + + private static void putUsageIfPresent( + AttributesBuilder attributes, Map<String, Object> eventAttributes) { + if (eventAttributes == null) { + return; + } + Object prompt = eventAttributes.get("promptTokens"); + if (prompt instanceof Number) { + attributes.put(GEN_AI_USAGE_INPUT_TOKENS, ((Number) prompt).longValue()); + } + Object completion = eventAttributes.get("completionTokens"); + if (completion instanceof Number) { + attributes.put(GEN_AI_USAGE_OUTPUT_TOKENS, ((Number) completion).longValue()); + } + } + + private static SpanContext spanContext(String inputRunId, String spanIdHex) { + return SpanContext.create( + OTelIds.traceId(inputRunId), + spanIdHex, + TraceFlags.getSampled(), + TraceState.getDefault()); + } + + private static long epochNanos(String isoTimestamp) { + Instant instant = Instant.parse(isoTimestamp); Review Comment: Done: an invalid timestamp now yields a `MALFORMED_RECORD` diagnostic and skips that record. ########## integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java: ########## @@ -0,0 +1,427 @@ +/* + * 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.ExecutionLifecycleEvents; +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 java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Assembles OpenTelemetry spans from Agent Trace Event 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} (falling back to the run root); + * <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 Event 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_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 String INSTRUMENTATION_SCOPE_NAME = "org.apache.flink.agents.otel"; + + private final Resource resource; + private final InstrumentationScopeInfo scope; + + public AgentTraceSpans(String serviceName) { + this.resource = + Resource.getDefault().toBuilder() + .put(AttributeKey.stringKey("service.name"), serviceName) + .build(); + this.scope = InstrumentationScopeInfo.create(INSTRUMENTATION_SCOPE_NAME); + } + + /** Assembles spans from Event 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) 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) { + String eventType = record.getEventType(); + if (eventType == null + || !ExecutionLifecycleEvents.isExecutionLifecycleEvent(eventType)) { + continue; + } + if (record.getExecutionId() == null + || record.getInputRunId() == null + || record.getTimestamp() == null) { + continue; + } + executions + .computeIfAbsent(record.getExecutionId(), id -> new ExecutionSpanBuilder()) + .accept(record); + runs.computeIfAbsent(record.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, 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); + } + if (run.businessKey != null) { + attributes.put(GEN_AI_CONVERSATION_ID, 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, List<ConverterDiagnostic> diagnostics) { + TraceRecord any = execution.anyRecord(); + String inputRunId = any.getInputRunId(); + String entityType = any.getEntityType(); + String entityName = any.getEntityName() != null ? any.getEntityName() : "unknown"; + + AttributesBuilder attributes = Attributes.builder(); + attributes.put(FA_INPUT_RUN_ID, inputRunId); + attributes.put(FA_EXECUTION_ID, any.getExecutionId()); + if (entityType != null) { + attributes.put(FA_ENTITY_TYPE, entityType); + } + attributes.put(FA_ENTITY_NAME, entityName); + if (any.getBusinessKey() != null) { + attributes.put(GEN_AI_CONVERSATION_ID, 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); + // Prefer the provider-issued call id, which is what the model's tool-call request + // carries; the framework-assigned id is the fallback. + String callId = stringValue(metadata, ToolExecutionMetadataKeys.EXTERNAL_ID); + if (callId == null) { + 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.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 != null ? entityType + " " : "") + entityName; + kind = SpanKind.INTERNAL; + } + + StatusData status = StatusData.unset(); + TraceRecord terminal = execution.terminal; + if (terminal != null) { + if (ExecutionLifecycleEvents.STATUS_FAILED.equals(terminal.getStatus())) { + status = + StatusData.create( + StatusCode.ERROR, + terminal.getErrorMessage() != null + ? terminal.getErrorMessage() + : ""); + String errorType = + terminal.getErrorType() != null + ? terminal.getErrorType() + : terminal.getProblemCategory(); + if (errorType != null) { + attributes.put(ERROR_TYPE, errorType); + } + } + if (terminal.getStatus() != null) { + attributes.put(FA_EXECUTION_STATUS, terminal.getStatus()); + } + putUsageIfPresent(attributes, terminal.getEventAttributes()); + } + boolean reused = + terminal != null + && ExecutionLifecycleEvents.STATUS_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; + + SpanContext parent = + any.getParentExecutionId() != null + ? spanContext(inputRunId, OTelIds.spanId(any.getParentExecutionId())) + : spanContext(inputRunId, OTelIds.runRootSpanId(inputRunId)); + + 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 static String stringValue(Map<String, Object> map, String key) { + Object value = map.get(key); + return value == null ? null : String.valueOf(value); + } + + private static void putUsageIfPresent( + AttributesBuilder attributes, Map<String, Object> eventAttributes) { + if (eventAttributes == null) { + return; + } + Object prompt = eventAttributes.get("promptTokens"); Review Comment: Good catch; the docs now say built-in reporters don't write these token attributes yet. ########## integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java: ########## @@ -0,0 +1,427 @@ +/* + * 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.ExecutionLifecycleEvents; +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 java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Assembles OpenTelemetry spans from Agent Trace Event 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} (falling back to the run root); + * <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 Event 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_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 String INSTRUMENTATION_SCOPE_NAME = "org.apache.flink.agents.otel"; + + private final Resource resource; + private final InstrumentationScopeInfo scope; + + public AgentTraceSpans(String serviceName) { + this.resource = + Resource.getDefault().toBuilder() + .put(AttributeKey.stringKey("service.name"), serviceName) + .build(); + this.scope = InstrumentationScopeInfo.create(INSTRUMENTATION_SCOPE_NAME); + } + + /** Assembles spans from Event 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) 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) { + String eventType = record.getEventType(); + if (eventType == null + || !ExecutionLifecycleEvents.isExecutionLifecycleEvent(eventType)) { + continue; + } + if (record.getExecutionId() == null + || record.getInputRunId() == null + || record.getTimestamp() == null) { + continue; + } + executions + .computeIfAbsent(record.getExecutionId(), id -> new ExecutionSpanBuilder()) + .accept(record); + runs.computeIfAbsent(record.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, 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); + } + if (run.businessKey != null) { + attributes.put(GEN_AI_CONVERSATION_ID, 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, List<ConverterDiagnostic> diagnostics) { + TraceRecord any = execution.anyRecord(); + String inputRunId = any.getInputRunId(); + String entityType = any.getEntityType(); + String entityName = any.getEntityName() != null ? any.getEntityName() : "unknown"; + + AttributesBuilder attributes = Attributes.builder(); + attributes.put(FA_INPUT_RUN_ID, inputRunId); + attributes.put(FA_EXECUTION_ID, any.getExecutionId()); + if (entityType != null) { + attributes.put(FA_ENTITY_TYPE, entityType); + } + attributes.put(FA_ENTITY_NAME, entityName); + if (any.getBusinessKey() != null) { + attributes.put(GEN_AI_CONVERSATION_ID, 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); Review Comment: Done: tool spans now carry `gen_ai.agent.name` whenever their records have one. ########## integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java: ########## @@ -0,0 +1,427 @@ +/* + * 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.ExecutionLifecycleEvents; +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 java.time.Instant; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Assembles OpenTelemetry spans from Agent Trace Event 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} (falling back to the run root); + * <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 Event 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_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 String INSTRUMENTATION_SCOPE_NAME = "org.apache.flink.agents.otel"; + + private final Resource resource; + private final InstrumentationScopeInfo scope; + + public AgentTraceSpans(String serviceName) { + this.resource = + Resource.getDefault().toBuilder() + .put(AttributeKey.stringKey("service.name"), serviceName) + .build(); + this.scope = InstrumentationScopeInfo.create(INSTRUMENTATION_SCOPE_NAME); + } + + /** Assembles spans from Event 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) 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) { + String eventType = record.getEventType(); + if (eventType == null + || !ExecutionLifecycleEvents.isExecutionLifecycleEvent(eventType)) { + continue; + } + if (record.getExecutionId() == null + || record.getInputRunId() == null + || record.getTimestamp() == null) { + continue; + } + executions + .computeIfAbsent(record.getExecutionId(), id -> new ExecutionSpanBuilder()) + .accept(record); + runs.computeIfAbsent(record.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, 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); + } + if (run.businessKey != null) { + attributes.put(GEN_AI_CONVERSATION_ID, 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, List<ConverterDiagnostic> diagnostics) { + TraceRecord any = execution.anyRecord(); + String inputRunId = any.getInputRunId(); + String entityType = any.getEntityType(); + String entityName = any.getEntityName() != null ? any.getEntityName() : "unknown"; + + AttributesBuilder attributes = Attributes.builder(); + attributes.put(FA_INPUT_RUN_ID, inputRunId); + attributes.put(FA_EXECUTION_ID, any.getExecutionId()); + if (entityType != null) { + attributes.put(FA_ENTITY_TYPE, entityType); + } + attributes.put(FA_ENTITY_NAME, entityName); + if (any.getBusinessKey() != null) { + attributes.put(GEN_AI_CONVERSATION_ID, any.getBusinessKey()); Review Comment: Agreed; it's always `flink_agents.business_key` now, and conversation id is opt-in. -- 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]
