[Discussion][Observability 2/2] Built-in Operational Metrics for Flink Agents

[joeyutong]

This proposal is the second in a two-part Flink Agents observability series:

1. Recording Agent Traces in the Event Log
2. Built-in Operational Metrics for Flink Agents (this proposal)

Part 1 defines the per-run recording needed to reconstruct an Agent Trace. This proposal defines bounded-cardinality metrics for observing aggregate health and behavior across runs.

The two capabilities share runtime observation boundaries but produce independent outputs:

Agent Trace         = per-run recording
Operational metrics = aggregate measurement

1. Context and Relationship to Part 1

Flink already provides job, operator, checkpoint, backpressure, and resource metrics. These metrics describe the Flink execution layer, but they do not directly answer Agent-specific operational questions:

• Are input runs succeeding, and is their end-to-end latency increasing?
• Are inputs waiting behind other inputs for the same key?
• Is an Action waiting to be scheduled or spending too long on asynchronous work?
• Are failures or latency concentrated in a particular model resource, Tool, Skill, or MCP Server?
• Are model retries or token consumption increasing unexpectedly?

Part 1 proposes two runtime Event categories for per-run observation:

• Business Events describe workflow occurrences and Event-to-Action causality.
• Execution Events describe the lifecycle of Action, LLM, Parser, and Tool executions.

Those Events also provide natural boundaries for aggregate metrics. Trace logging and metric aggregation can consume the same Execution Events independently, so Action, LLM, and Tool metrics follow the lifecycle semantics recorded in each trace.

Execution Events cover only work represented as executions. Queue depth, ActionTask scheduling, token usage, and retry statistics instead arise from runtime state or call results and are recorded at their own boundaries. The complete signal model is therefore:

flowchart LR
    BE["Business Event"] --> TRACE["Event Log<br/>per-run Agent Trace"]
    BE --> EVENT_METRICS["Event-driven metrics<br/>Business Event / Action / LLM / Tool"]

    EE["Execution Event"] --> TRACE
    EE --> EVENT_METRICS

    RUN_STATE["Run and queue state"] --> RUN_METRICS["Input Run and scheduling metrics"]
    MODEL_RESULT["Model call result"] --> MODEL_METRICS["Token usage and retry metrics"]

Loading

This model aligns Agent Trace and execution metrics on lifecycle, entity, and metadata semantics while keeping the two outputs independent. Terminal outcomes use the existing contract of each runtime.

2. Goals and Non-goals

Goals

• Define built-in metrics for Input Run health, Action scheduling, and Agent execution entities.
• Align Action, LLM, and Tool metric boundaries with the Execution Event model from Part 1.
• Define stable metric names, types, scopes, and outcome semantics.
• Keep metric dimensions bounded and suitable for aggregation.
• Preserve the distinction between configured resources and provider-returned model names.
• Use the same metric names, scopes, and lifecycle outcomes across Java and Python while making language-specific Tool outcome mappings explicit.
• Define metric behavior across task attempts and recovery.

Non-goals

• Dashboard layout, visualization, or alert thresholds.
• Reporter-specific queries or Prometheus expressions.
• Per-run diagnosis through metric labels.
• Exactly-once metric values across task attempts.
• Provider-attempt-level LLM metrics.
• Evaluation datasets, experiments, or model-quality metrics.

3. Metric Model

3.1 Scope Hierarchy

The Agent name identifies the Flink operator that runs the Agent. Agent-specific dimensions are registered as key-value MetricGroup scopes. Metric reporters that support dimensions, such as Prometheus, can expose these scopes as labels; other reporters may encode them in the metric identifier.

Agent operator
├─ Agent-level metrics
└─ action=<action_name>
   ├─ Action-level metrics
   ├─ model_resource=<resource_name>
   ├─ tool=<tool_name>
   ├─ skill=<skill_name>
   ├─ mcp_server=<server_name>
   └─ model=<model_name>

Scope	Meaning
Agent	Aggregate behavior of the Agent operator, without selecting an Action
Action	Scheduling and execution behavior of one Action
Model Resource	The ChatModel resource name referenced by the Agent plan; used for call health and retries
Model	The model name returned by the provider; used for token consumption
Tool	One named Tool; used for Tool-call health
Skill	The Skill name carried by an explicit load_skill Tool call
MCP Server	The MCP Server name carried by an MCP Tool execution

model_resource and model belong to independent branches:

• Model Resource metrics answer whether a configured ChatModel resource is healthy.
• Model metrics attribute token consumption to the model name returned by the provider.

When one resource is shared by multiple Actions, each Action produces a separate raw metric series. Consumers can retain the Action scope to locate the caller. Counters such as call successes and failures can also be summed across Actions to view resource-wide outcomes. Latency Histograms remain per-Action distributions; their exported percentiles are not generally composable across Action series, so this proposal does not define a resource-wide latency percentile.

3.2 Metric Types and Naming

The proposal uses standard Flink metric types:

Type	Semantics
Counter	A cumulative count since the current task attempt started
Meter	A rate derived from a Counter
Histogram	A distribution of latency samples observed in the current task attempt
Gauge	A current value that can increase or decrease

Metric names follow the existing Flink Agents style:

• Counts use numOf....
• Rates use ...PerSec.
• Millisecond latency distributions use ...LatencyMs.
• Existing domain-specific metric names for token usage and retries are retained.
• Retry metrics intentionally move from the model scope to model_resource. This is a breaking scope change; compatibility is limited to the metric names.
• Scope keys use snake_case, such as model_resource and mcp_server.

3.3 Cardinality

Operational metrics intentionally exclude per-run and per-execution identifiers. Fields such as input_run_id, execution_id, business_key, and event_id are useful for Agent Trace queries but would create unbounded metric cardinality.

Metric scopes are limited to reusable Agent entities and resources: Action, Model Resource, model name, Tool, Skill, and MCP Server. Metrics identify where aggregate behavior is concentrated; Agent Trace remains the per-run diagnostic surface.

4. Built-in Metrics

The existing Event throughput, Action throughput, and token metrics remain part of the built-in metric surface. This proposal adds Input Run, Action scheduling, and execution-entity health metrics while preserving existing names and semantics.

4.1 Existing Baseline

Scope	Metric	Type	Meaning
Agent	numOfEventProcessed	Counter	Processed business Events
Agent	numOfEventProcessedPerSec	Meter	Business Event processing rate
Agent	numOfActionsExecuted	Counter	Completed Actions
Agent	numOfActionsExecutedPerSec	Meter	Action completion rate
Action	numOfActionsExecuted	Counter	Completed invocations of one Action
Action	numOfActionsExecutedPerSec	Meter	Completion rate of one Action
Agent	eventLogTruncatedEvents	Counter	Event Log records whose payload was truncated
Model	promptTokens	Counter	Prompt tokens consumed
Model	completionTokens	Counter	Completion tokens consumed
Model	totalTokens	Counter	Total tokens consumed

4.2 Input Run

An Input Run is one attempt to process an input for a business key. Metric tracking begins when the runtime receives the Input Event. The run's processing phase begins later, when that Event leaves the pending queue. Tracking ends when all associated Actions complete or an unhandled exception terminates the run.

Scope	Metric	Type	Meaning
Agent	numOfInputRunsSucceeded	Counter	Input Runs that reached the normal completion boundary
Agent	numOfInputRunsFailed	Counter	Input Runs terminated by an unhandled exception, including failures before run context creation
Agent	inputRunLatencyMs	Histogram	Input receipt to run completion or failure
Agent	inputRunQueueLatencyMs	Histogram	Input receipt to the start of run processing, primarily reflecting same-key queueing
Agent	inputRunProcessingLatencyMs	Histogram	Start of run processing to completion or failure
Agent	numOfPendingInputEvents	Gauge	Input Events waiting in same-key queues
Agent	numOfActiveInputRuns	Gauge	Runs selected for processing but not yet complete

The latency boundaries are:

Input Event received
    │
    ├─ inputRunQueueLatencyMs
    │
Input Run starts processing
    │
    ├─ inputRunProcessingLatencyMs
    │
Input Run completed or failed

inputRunLatencyMs = received → completed or failed

Here, processing starts when the Input Event leaves the pending queue and enters Event routing. It does not imply that the first Action has already begun executing.

A child execution failure does not by itself make the Input Run fail. If the Agent handles the failure and reaches its normal completion boundary, the run is successful.

4.3 Action

Action metrics distinguish between a physical ActionTask and a logical Action execution. One logical execution may suspend while waiting for asynchronous work and later resume through another ActionTask.

Scope	Metric	Type	Meaning
Action	actionSchedulingLatencyMs	Histogram	Initial ActionTask enqueue to its first execution
Action	actionExecutionLatencyMs	Histogram	Action execution started to finished or failed, including asynchronous waiting
Action	numOfPendingActionTasks	Gauge	Physical ActionTask instances waiting for execution, including continuations
Action	numOfActiveActionExecutions	Gauge	Logical Action executions that started but have not reached a terminal state

Only the initial ActionTask contributes a scheduling-latency sample. A continuation updates the pending-task Gauge but does not create another scheduling sample.

4.4 LLM and Token Usage

LLM health and framework retry metrics use the configured ChatModel setup resource scope. The model_resource value is the AgentPlan resource name used to resolve a CHAT_MODEL resource; it is not the underlying ChatModel connection name or the model name returned by the provider. Each LLM health sample represents one framework-observed model invocation. A framework retry that calls the model again produces another sample; provider-internal attempts remain outside this metric boundary.

Scope	Metric	Type	Meaning
Model Resource	numOfLlmCallsSucceeded	Counter	Framework-observed model invocations that returned successfully
Model Resource	numOfLlmCallsFailed	Counter	Framework-observed model invocations that failed
Model Resource	llmCallLatencyMs	Histogram	Latency of each framework-observed model invocation; excludes Parser execution and retry wait time
Model Resource	retryCount	Counter	Additional model invocations initiated by framework retry logic
Model Resource	retryWaitSec	Counter	Backoff time accumulated by those framework-level retries, in seconds
Model	promptTokens	Counter	Prompt tokens attributed to the provider-returned model name
Model	completionTokens	Counter	Completion tokens attributed to the provider-returned model name
Model	totalTokens	Counter	Total tokens attributed to the provider-returned model name

The underlying connection or provider SDK may perform its own transport retries. Those attempts are a separate connection-level signal and are not included in retryCount or retryWaitSec.

Structured-output parsing is a separate execution in Agent Trace. If a model invocation succeeds but parsing fails, that LLM execution remains successful and the Parser execution fails. If framework retry logic calls the model again, the new invocation contributes another LLM outcome and latency sample, while retryCount and retryWaitSec describe the retry behavior around those invocations.

4.5 Tool, Skill, and MCP Server

Each Tool call is one Tool execution. Skill and MCP Server metrics are additional aggregate projections derived from explicit Tool execution metadata; they do not introduce separate execution levels.

The mcp_server scope uses the MCP Server resource name declared in the AgentPlan. When the plan expands an MCP Server into Tool resources, it attaches that name to each discovered MCP Tool. At execution time, the Tool exposes the name through the shared Tool execution metadata contract, and both Java and Python use the same metadata key. The runtime emits MCP Server metrics only when this declared identity is present; it does not infer the identity from an endpoint.

Scope	Metric	Type	Meaning
Tool	numOfToolCallsSucceeded	Counter	Successful Tool calls
Tool	numOfToolCallsFailed	Counter	Failed Tool calls
Tool	toolCallLatencyMs	Histogram	Tool-call latency
Skill	numOfSkillLoads	Counter	Explicit load_skill calls that reached a terminal state
Skill	skillLoadLatencyMs	Histogram	Explicit load_skill call latency
MCP Server	numOfMcpToolCallsSucceeded	Counter	Successful Tool calls exposed by one MCP Server
MCP Server	numOfMcpToolCallsFailed	Counter	Failed Tool calls exposed by one MCP Server
MCP Server	mcpToolCallLatencyMs	Histogram	Latency of Tool calls exposed by one MCP Server

Every named Tool execution contributes to Tool metrics. Only an explicit load_skill Tool execution contributes to Skill metrics. The runtime does not infer that subsequent Tool calls belong to the loaded Skill. A Tool execution carrying MCP Server metadata contributes to both its Tool and MCP Server scopes.

5. Recording Semantics

5.1 Event-Driven Metrics

Metrics derived from Business Events and Execution Events use the same semantic boundaries as Agent Trace:

• A processed business Event increments Event throughput metrics.
• Action Execution Events update Action execution latency and active-execution state.
• Terminal LLM and Tool Execution Events update outcome counters and latency.
• Tool execution metadata adds Skill and MCP Server aggregate projections where applicable.

execution_id may be used internally to pair started and terminal Events, but it is never exposed as a metric scope or label.

Metric aggregation consumes these in-process Events directly rather than reading them back from the Event Log.

5.2 Runtime State Metrics

Input Run and scheduling metrics require state transitions that are not fully represented by Execution Events:

• Input receipt, pending-queue enqueue, and pending-queue dequeue define run queue latency and pending counts.
• Run processing start and terminal boundaries define processing latency and active-run counts.
• ActionTask enqueue and dequeue define scheduling latency and pending-task counts.

These observations extend the metric model without turning queue operations into public Event types.

5.3 Model Call Metrics

Token usage and retry information are available only at the model-call boundary:

• Retry metrics record the number of additional framework invocations and their cumulative wait time.
• Token metrics use usage data and the model name returned in the final response.

These values complement the LLM Execution Event without expanding its lifecycle schema with provider-specific payloads.

5.4 Outcome Boundaries

Tool metrics consume the outcome produced by each runtime's existing Tool contract; the runtime does not inspect an arbitrary return payload to guess whether it represents an error. In both Java and Python, resource lookup, argument preparation, or invocation exceptions are failures, while a normal return is successful. Java additionally represents an explicit unsuccessful result through ToolResponse; Python Tool.call currently returns Any and has no equivalent explicit error-result type, so a normally returned Python value remains successful.

Strict alignment of explicit Tool error results is tracked in Issue #956 and should follow the parallel Tool-call work in PR #926. Until then, Tool and MCP outcome metrics retain these language-specific mappings.

Scenario	Metric outcome
An LLM invocation fails and a retry succeeds	One failed and one successful LLM call; each records its own latency; retry counters increase
An LLM call succeeds but structured-output parsing fails	LLM success; Parser failure remains visible in Agent Trace
A Tool returns normally	Successful Tool execution
Tool resource lookup, argument preparation, or invocation throws an exception	Failed Tool execution, even if the containing Action handles the result
A Java Tool returns an unsuccessful ToolResponse	Failed Tool execution
A child execution fails but the Agent handles it	Child failure metrics increase; the Input Run can still succeed
An Action suspends and resumes through a continuation	One logical Action execution; continuation affects pending tasks but not scheduling sample count

6. Task Attempt and Recovery Semantics

Flink metrics are scoped to the current task attempt. Counter, Meter, and Histogram values are not restored from checkpoints and do not provide exactly-once values across failures.

The prototype reconstructs all four current-value Gauges from checkpointed operator state before processing resumes:

Gauge	Restored state used for reconstruction
numOfActiveInputRuns	Processing keys owned by the operator at the checkpoint
numOfPendingInputEvents	Pending Input Events buffered in keyed state
numOfPendingActionTasks	Physical ActionTasks in the restored per-key task queues
numOfActiveActionExecutions	Restored ActionTasks whose execution-start marker is set, deduplicated by execution id

These Gauges are rebuilt rather than reset or approximated after failover. They represent the restored checkpoint state and then change as replayed or newly received work is processed. Historical start timestamps cannot be reconstructed:

• A latency sample is recorded only when the current attempt observes both the required start and terminal boundaries.
• Restored pending or active work contributes to current-value Gauges without fabricating a historical start time.
• LLM and Tool terminal outcomes observed in the current attempt may update counters even when no latency sample can be produced.

Consumers must aggregate Gauge values across subtasks for a job-level view. Histograms include only samples observed by the current task attempts.

7. Cross-Language and Reporter Semantics

Java and Python use the same logical execution entity types, lifecycle outcome values, metadata keys, metric names, and scope hierarchy. Each runtime maps its existing Tool API to those common lifecycle outcomes as described in section 5.4; this does not imply that Python has an equivalent to Java ToolResponse.

MetricGroup scopes are the framework-level representation of dimensions. A reporter may expose them as labels or encode them in a metric identifier, but this does not change their logical meaning. Dashboards and queries must preserve or deliberately aggregate scopes rather than assume identical physical names across reporters.

8. Prototype Status

The metric model has been prototyped in both the Java and Python runtime paths. The prototype covers:

• Input Run outcome, latency, queue, and active-state metrics.
• Action scheduling, execution latency, pending-task, and active-execution metrics.
• Event-driven LLM and Tool outcome and latency metrics.
• Model Resource, model, Tool, Skill, and MCP Server scopes.
• Token usage and retry metrics.
• Task-attempt and recovery behavior for Gauges and latency samples.
• Java and Python metric-schema and lifecycle coverage through unit and runtime tests.

The purpose of this discussion is to review the metric contracts and their relationship to Agent Trace before submitting the implementation upstream.

9. Summary

Agent Trace and operational metrics answer different questions but share the same runtime semantics. Business Events and Execution Events provide stable observation boundaries for Event throughput and execution health. Run state, queue state, and model-call results provide the additional signals required by metrics.

This proposal adds:

• Input Run health and latency metrics.
• Action scheduling and execution-state metrics.
• LLM and Tool success, failure, and latency metrics.
• Model Resource, model, Tool, Skill, and MCP Server scopes.
• Explicit cardinality and task-attempt semantics.
• A shared metric surface for Java and Python with explicit runtime-specific outcome mappings.

The result is a bounded-cardinality operational metric surface that complements per-run Agent Trace without depending on Event Log storage or introducing per-run identifiers into metrics.

[weiqingy]

Thanks for writing this up. The model_resource / provider model split in §3.1 looks like an important cleanup to me, especially because retry health and token attribution answer different questions. A few edge cases seem worth clarifying before the metric contract lands.

Retry scope and compatibility. §3.2 says existing retry names are retained for compatibility, but the proposed scope appears to move from model=... to model_resource=.... Today the retry value also looks closer to the connection name than the ChatModel setup name. Is the model → model_resource move intended as a breaking metric-scope change, with compatibility limited to the metric names? And should LLM health/retry metrics be scoped to the configured ChatModel setup, or to the underlying connection where transport failures and retries actually happen?

MCP Server identity. §4.5 adds mcp_server metrics, but it is not clear that the declared MCP Server name is available at Tool execution time in both Java and Python paths. Would Part 1 thread the declared server name through Tool execution metadata, or should the mcp_server metric scope be deferred until that identity is available consistently?

Latency aggregation. §3.1 says consumers can aggregate across the Action scope to view resource-wide health. That works for counters, but latency histograms/percentiles generally do not compose across per-Action series. Is llmCallLatencyMs intended to be per-Action latency, or resource-wide latency? Either seems reasonable, but the contract should probably avoid promising resource-wide latency aggregation from sharded histograms. The same question seems to apply to Tool, Skill, and MCP latency histograms.

Gauge recovery semantics. §6 says Gauges may be reconstructed from restored runtime state when the state identifies pending or active work. Could we spell out which gauges are actually reconstructable after failover? numOfActiveInputRuns and numOfPendingInputEvents seem to have clearer restored-state sources than numOfPendingActionTasks, where the restored action-task queue may not expose an exact size. It would help to distinguish “correctly reconstructed” gauges from “reset or best-effort after task-attempt restart.”

Tool success/failure parity. §5.4 says a Tool error response is counted as a failed Tool execution, and §7 makes Java/Python outcome alignment part of the contract. Could we define whether Tool success/failure is a framework-level contract or a best-effort interpretation of the Tool return value? Java has ToolResponse.success/error, while Python tools currently look more raw-value oriented, so numOfToolCallsFailed could mean different things unless the contract is tightened or documented as best-effort.

[joeyutong]

Thanks for raising these points. I agree that these contracts should be made more explicit.

Retry scope and compatibility. The move from model to model_resource is an intentional breaking change to the metric scope; the compatibility statement applies only to the existing metric names. The scope value identifies the configured ChatModel setup selected by ChatRequestEvent.model, not the underlying ChatModel connection. LLM health and retry metrics belong to the setup because they describe the framework-level logical call and retry loop in ChatModelAction. Connection or provider SDK transport retries are a separate lower-level signal and are not covered by these metrics.

MCP Server identity. The Part 1 prototype already propagates this identity in both Java and Python. When an MCP Server declared in the AgentPlan is expanded into Tool resources, each discovered MCP Tool retains the declared server resource name. At execution time, the Tool exposes that name through the common Tool execution metadata contract, and the metrics recorder uses it as the mcp_server scope. Both runtime paths use the same metadata key and semantics. If that identity is unavailable, the runtime does not infer it from the endpoint and does not emit the server-scoped metric.

Latency aggregation. Agreed. The current latency histograms are scoped to an Action and resource, so their exported percentiles should not be treated as generally composable across Actions. The resource-wide aggregation statement applies to counters such as call successes and failures, not to latency percentiles. The proposal now clarifies that llmCallLatencyMs, Tool latency, Skill latency, and MCP latency describe their respective per-Action resource scopes.

Gauge recovery semantics. In the prototype, all four proposed Gauges are reconstructed from restored runtime state. Active input runs come from the restored processing-key state, pending input Events from the pending-Event state, pending Action tasks from the restored ActionTask queues, and active Action executions from restored tasks whose execution-start marker is set. These Gauges are reconstructed from checkpointed state rather than reset or treated as best-effort after failover. The proposal now spells out these state sources.

Tool success/failure parity. Tool metrics follow each runtime's existing Tool contract and do not interpret arbitrary return payloads. Java treats an unsuccessful ToolResponse as failure. Python treats resource preparation or invocation exceptions as failure, but any normal Tool.call return remains successful because Python currently has no equivalent explicit error-result type. The proposal now states this limitation explicitly; strict alignment is tracked in #956 after the parallel Tool-call work in #926.

[weiqingy]

Thanks for the reply, and for folding all five answers into the proposal body.

One correction on my side: my retry question assumed the loop lives at the connection, and it does not. The framework loop is in the Action in both languages, while the connection-level max_retries passed to the provider SDK is a separate lower-level layer, so scoping the retry metrics to the setup is the right call.

Nothing blocking from me on the design. I do still have a few questions on the details, for example the Python side of the Tool outcome contract: §5.4 presumes an error result there, and Tool.call currently returns Any with no ToolResponse equivalent, so I may be missing where that normalization happens. Let me take a look at the implementation first, and I will follow up there if it still seems open.

[joeyutong]

@weiqingy You're right—there is no Python normalization equivalent to Java ToolResponse today. I corrected my earlier reply and updated the proposal and PR documentation: Java maps an unsuccessful ToolResponse to failure, while Python maps resource preparation or invocation exceptions to failure and treats any normal return as success. We are retaining the Tool/MCP outcome metrics with these explicit runtime-specific semantics. Strict alignment is tracked in #956 and will follow #926. The Metrics implementation is now available in draft #955.

[weiqingy]

Thanks for the correction, and for updating the proposal body along with it rather than leaving it in the reply. That closes everything I had open on the design.

The implementation PR looks like the better place for the remaining details, so I will go through it and follow up there once it is out of draft.