sangkyoonnam commented on code in PR #1160: URL: https://github.com/apache/flink-agents/pull/1160#discussion_r4307811582
########## docs/content/docs/operations/monitoring.md: ########## @@ -401,6 +401,62 @@ 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` | Review Comment: If the branch above goes in, the mapping table needs the row too. ```suggestion | `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` | ``` ########## integrations/observability/otel/src/main/java/org/apache/flink/agents/integrations/observability/otel/AgentTraceSpans.java: ########## @@ -0,0 +1,536 @@ +/* + * 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.ACTION.equals(entityType)) { Review Comment: Non-blocking: `subagent` has been an execution type since #1114, and `scope: ALL` records it, but it takes the fallback branch here. A delegation exports as `subagent researcher` with no `gen_ai.operation.name` or `gen_ai.agent.name`, and the `toolCallId` in its metadata is dropped. I fed the assembler a created/started/failed sequence for one delegation and got exactly that span. This predates the port. The conventions name the span `invoke_agent {gen_ai.agent.name}` (`gen_ai.invoke_agent.internal` in `model/gen-ai/spans.yaml`). With the branch below, the same records give `invoke_agent researcher` with `gen_ai.operation.name=invoke_agent`, `gen_ai.agent.name=researcher` and `gen_ai.tool.call.id=call_1`, and the 34 tests still pass. Kind is the open question: the conventions use INTERNAL for an in-process agent and CLIENT for a remote one, and the record doesn't say whether an internal sub-agent or an async remote job (`BaseAsyncSubagentSetup`) ran. I kept the fallback's INTERNAL, so the kind doesn't change. `gen_ai.tool.call.id` isn't in the `invoke_agent` table, but it keeps the model's tool-call correlation ID. If you'd rather wait until the record carries the locality, a follow-up works too. ```suggestion } 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)) { ``` ########## docs/content/docs/operations/monitoring.md: ########## @@ -401,6 +401,62 @@ 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`. Review Comment: Same for the type list here. ```suggestion 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`. ``` -- 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]
