joeyutong commented on code in PR #955:
URL: https://github.com/apache/flink-agents/pull/955#discussion_r3891535248
##########
runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInActionMetrics.java:
##########
@@ -19,26 +19,124 @@
package org.apache.flink.agents.runtime.metrics;
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.trace.ExecutionLifecycleEvents;
+import org.apache.flink.agents.api.trace.ExecutionTraceContext;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
import org.apache.flink.metrics.Meter;
-/**
- * ActionMetricGroup class extends FlinkAgentsMetricGroupImpl and is used to
monitor and measure the
- * performance metrics of executing actions. It provides metrics for the total
number of actions
- * executed and the number of actions executed per second.
- */
+import java.util.HashMap;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+
+/** Tracks execution rate, scheduling latency, and current task/execution
counts for one Action. */
public class BuiltInActionMetrics {
+ static final String ACTION_SCHEDULING_LATENCY_MS =
"actionSchedulingLatencyMs";
+ static final String ACTION_EXECUTION_LATENCY_MS =
"actionExecutionLatencyMs";
+ static final String NUM_PENDING_ACTION_TASKS = "numOfPendingActionTasks";
+ static final String NUM_ACTIVE_ACTION_EXECUTIONS =
"numOfActiveActionExecutions";
+
private final Meter numOfActionsExecutedPerSec;
+ private final Histogram schedulingLatencyHistogram;
+ private final Histogram executionLatencyHistogram;
+ private final CurrentCountGauge pendingActionTasks;
+ private final CurrentCountGauge activeActionExecutions;
+ private final LongSupplier nanoTime;
+
+ private final Map<String, Long> initialTaskEnqueueNanos = new HashMap<>();
+ private final Map<String, OptionalLong> activeExecutions = new HashMap<>();
public BuiltInActionMetrics(FlinkAgentsMetricGroupImpl parentMetricGroup) {
+ this(parentMetricGroup, System::nanoTime);
+ }
+
+ BuiltInActionMetrics(FlinkAgentsMetricGroupImpl parentMetricGroup,
LongSupplier nanoTime) {
Counter numOfActionsExecuted =
parentMetricGroup.getCounter("numOfActionsExecuted");
this.numOfActionsExecutedPerSec =
parentMetricGroup.getMeter("numOfActionsExecutedPerSec",
numOfActionsExecuted);
+ this.schedulingLatencyHistogram =
+ parentMetricGroup.getHistogram(ACTION_SCHEDULING_LATENCY_MS);
+ this.executionLatencyHistogram =
+ parentMetricGroup.getHistogram(ACTION_EXECUTION_LATENCY_MS);
+ this.pendingActionTasks =
+ new CurrentCountGauge(parentMetricGroup,
NUM_PENDING_ACTION_TASKS);
+ this.activeActionExecutions =
+ new CurrentCountGauge(parentMetricGroup,
NUM_ACTIVE_ACTION_EXECUTIONS);
+ this.nanoTime = nanoTime;
}
/** Marks that an action has finished executing. */
public void markActionExecuted() {
numOfActionsExecutedPerSec.markEvent();
}
+
+ void actionTaskEnqueued(String executionId, boolean executionStarted) {
+ pendingActionTasks.increment();
+ if (!executionStarted && !isBlank(executionId)) {
+ initialTaskEnqueueNanos.putIfAbsent(executionId,
nanoTime.getAsLong());
+ }
+ }
+
+ void actionTaskDequeued(String executionId, boolean executionStarted) {
+ pendingActionTasks.decrement();
+ if (executionStarted || isBlank(executionId)) {
+ return;
+ }
+
+ Long enqueueNanos = initialTaskEnqueueNanos.remove(executionId);
+ if (enqueueNanos != null) {
+ schedulingLatencyHistogram.update(
+ TimeUnit.NANOSECONDS.toMillis(
+ Math.max(0L, nanoTime.getAsLong() -
enqueueNanos)));
+ }
+ }
+
+ void restoreActionTask(String executionId, boolean executionStarted) {
+ pendingActionTasks.increment();
+ if (executionStarted
+ && !isBlank(executionId)
+ && activeExecutions.putIfAbsent(executionId,
OptionalLong.empty()) == null) {
+ activeActionExecutions.increment();
+ }
+ }
+
+ void executionEventObserved(Event event, ExecutionTraceContext
traceContext) {
+ String executionId = traceContext.getExecutionId();
+ if (isBlank(executionId)) {
+ return;
+ }
+
+ if
(ExecutionLifecycleEvents.EXECUTION_STARTED_EVENT_TYPE.equals(event.getType()))
{
+ if (activeExecutions.putIfAbsent(executionId,
OptionalLong.of(nanoTime.getAsLong()))
+ == null) {
+ activeActionExecutions.increment();
+ }
+ return;
+ }
+
+ if
(!ExecutionLifecycleEvents.EXECUTION_FINISHED_EVENT_TYPE.equals(event.getType())
+ &&
!ExecutionLifecycleEvents.EXECUTION_FAILED_EVENT_TYPE.equals(event.getType())
+ &&
!ExecutionLifecycleEvents.EXECUTION_REUSED_EVENT_TYPE.equals(event.getType())) {
Review Comment:
Agreed. This is the existing Agent Trace durable-replay limitation: child
cache reuse is not exposed, so these metrics currently inherit the
fresh-success and near-zero-latency behavior during fine-grained recovery. I
documented the metric impact explicitly; distinguishing reused child executions
remains follow-up work.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/metrics/ToolExecutionMetricRecorder.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.runtime.metrics;
+
+import org.apache.flink.agents.api.trace.ExecutionReporter;
+import org.apache.flink.agents.api.trace.ExecutionTraceContext;
+import org.apache.flink.agents.api.trace.ToolExecutionMetadataKeys;
+import org.apache.flink.metrics.Histogram;
+
+import java.util.Objects;
+import java.util.function.Predicate;
+
+/** Records Tool metrics and additional Skill and MCP projections. */
+final class ToolExecutionMetricRecorder implements ExecutionMetricRecorder {
+
+ static final String UNKNOWN_TOOL_NAME = "unknown";
+
+ static final String NUM_TOOL_CALLS_SUCCEEDED = "numOfToolCallsSucceeded";
+ static final String NUM_TOOL_CALLS_FAILED = "numOfToolCallsFailed";
+ static final String TOOL_CALL_LATENCY_MS = "toolCallLatencyMs";
+
+ static final String NUM_SKILL_LOADS = "numOfSkillLoads";
+ static final String SKILL_LOAD_LATENCY_MS = "skillLoadLatencyMs";
+
+ static final String NUM_MCP_TOOL_CALLS_SUCCEEDED =
"numOfMcpToolCallsSucceeded";
+ static final String NUM_MCP_TOOL_CALLS_FAILED = "numOfMcpToolCallsFailed";
+ static final String MCP_TOOL_CALL_LATENCY_MS = "mcpToolCallLatencyMs";
+
+ private final Predicate<String> isRegisteredTool;
+
+ ToolExecutionMetricRecorder(Predicate<String> isRegisteredTool) {
+ this.isRegisteredTool = Objects.requireNonNull(isRegisteredTool);
+ }
+
+ @Override
+ public String entityType() {
+ return ExecutionReporter.EntityTypes.TOOL;
+ }
+
+ @Override
+ public void record(
+ FlinkAgentsMetricGroupImpl actionMetricGroup,
+ ExecutionTraceContext traceContext,
+ Outcome outcome,
+ Long latencyMs) {
+ String requestedToolName = traceContext.getEntityName();
+ String metricToolName =
+ !isBlank(requestedToolName) &&
isRegisteredTool.test(requestedToolName)
+ ? requestedToolName
+ : UNKNOWN_TOOL_NAME;
+ recordOutcome(
+ actionMetricGroup.getSubGroup("tool", metricToolName),
+ outcome,
+ NUM_TOOL_CALLS_SUCCEEDED,
+ NUM_TOOL_CALLS_FAILED,
+ TOOL_CALL_LATENCY_MS,
+ latencyMs);
+
+ String skillName = metadataValue(traceContext,
ToolExecutionMetadataKeys.SKILL_NAME);
+ if (!isBlank(skillName)) {
+ FlinkAgentsMetricGroupImpl skillMetricGroup =
+ actionMetricGroup.getSubGroup("skill", skillName);
+ skillMetricGroup.getCounter(NUM_SKILL_LOADS).inc();
Review Comment:
Kept the existing Tool outcome contract in this PR. numOfSkillLoads is now
documented as counting terminal load_skill calls regardless of outcome,
including the current not-found normal-return behavior. Explicit failure-result
alignment is tracked in #956.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInMetrics.java:
##########
@@ -86,4 +172,12 @@ public Counter getEventLogTruncatedEventsCounter() {
public Counter getEventLogWriteFailuresCounter() {
return eventLogWriteFailures;
}
+
+ private BuiltInActionMetrics actionMetrics(String actionName) {
+ BuiltInActionMetrics actionMetrics =
actionMetricGroups.get(actionName);
+ if (actionMetrics == null) {
+ throw new IllegalArgumentException("Unknown action: " +
actionName);
Review Comment:
Updated the metric restore path so it no longer throws solely because a
restored Action is absent from the current Plan; its metric group is created
lazily. Regular live Action lookups remain strict, and a restored-action
regression test was added.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/metrics/ToolExecutionMetricRecorder.java:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.runtime.metrics;
+
+import org.apache.flink.agents.api.trace.ExecutionReporter;
+import org.apache.flink.agents.api.trace.ExecutionTraceContext;
+import org.apache.flink.agents.api.trace.ToolExecutionMetadataKeys;
+import org.apache.flink.metrics.Histogram;
+
+import java.util.Objects;
+import java.util.function.Predicate;
+
+/** Records Tool metrics and additional Skill and MCP projections. */
+final class ToolExecutionMetricRecorder implements ExecutionMetricRecorder {
+
+ static final String UNKNOWN_TOOL_NAME = "unknown";
+
+ static final String NUM_TOOL_CALLS_SUCCEEDED = "numOfToolCallsSucceeded";
+ static final String NUM_TOOL_CALLS_FAILED = "numOfToolCallsFailed";
+ static final String TOOL_CALL_LATENCY_MS = "toolCallLatencyMs";
+
+ static final String NUM_SKILL_LOADS = "numOfSkillLoads";
+ static final String SKILL_LOAD_LATENCY_MS = "skillLoadLatencyMs";
+
+ static final String NUM_MCP_TOOL_CALLS_SUCCEEDED =
"numOfMcpToolCallsSucceeded";
+ static final String NUM_MCP_TOOL_CALLS_FAILED = "numOfMcpToolCallsFailed";
+ static final String MCP_TOOL_CALL_LATENCY_MS = "mcpToolCallLatencyMs";
+
+ private final Predicate<String> isRegisteredTool;
+
+ ToolExecutionMetricRecorder(Predicate<String> isRegisteredTool) {
+ this.isRegisteredTool = Objects.requireNonNull(isRegisteredTool);
+ }
+
+ @Override
+ public String entityType() {
+ return ExecutionReporter.EntityTypes.TOOL;
+ }
+
+ @Override
+ public void record(
+ FlinkAgentsMetricGroupImpl actionMetricGroup,
+ ExecutionTraceContext traceContext,
+ Outcome outcome,
+ Long latencyMs) {
+ String requestedToolName = traceContext.getEntityName();
+ String metricToolName =
+ !isBlank(requestedToolName) &&
isRegisteredTool.test(requestedToolName)
+ ? requestedToolName
+ : UNKNOWN_TOOL_NAME;
+ recordOutcome(
+ actionMetricGroup.getSubGroup("tool", metricToolName),
+ outcome,
+ NUM_TOOL_CALLS_SUCCEEDED,
+ NUM_TOOL_CALLS_FAILED,
+ TOOL_CALL_LATENCY_MS,
+ latencyMs);
+
+ String skillName = metadataValue(traceContext,
ToolExecutionMetadataKeys.SKILL_NAME);
+ if (!isBlank(skillName)) {
+ FlinkAgentsMetricGroupImpl skillMetricGroup =
+ actionMetricGroup.getSubGroup("skill", skillName);
Review Comment:
Fixed. Java and Python now record whether the requested Skill is registered.
Registered names keep their own scope; all other names are aggregated under
skill=unknown, while Agent Trace retains the requested name. Regression
coverage was added.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/metrics/CurrentCountGauge.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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.runtime.metrics;
+
+/** Maintains a non-negative current-count gauge on the operator mailbox
thread. */
+final class CurrentCountGauge {
+
+ private final UpdatableGaugeImpl<Long> gauge;
+ private long value;
+
+ @SuppressWarnings("unchecked")
+ CurrentCountGauge(FlinkAgentsMetricGroupImpl metricGroup, String name) {
+ this.gauge = (UpdatableGaugeImpl<Long>) metricGroup.getGauge(name);
+ update(0L);
+ }
+
+ void increment() {
+ update(value + 1L);
+ }
+
+ void decrement() {
+ update(Math.max(0L, value - 1L));
Review Comment:
Added a focused CurrentCountGaugeTest covering decrement() at zero and
set(-1), so both clamping branches are exercised directly.
##########
docs/content/docs/operations/monitoring.md:
##########
@@ -36,11 +36,58 @@ We offer data monitoring for built-in metrics, which
includes events, actions, a
| **Agent** | numOfEventProcessedPerSec | The number of
Events this operator has processed per second. | Meter |
| **Agent** | numOfActionsExecuted | The total
number of actions this operator has executed. | Count |
| **Agent** | numOfActionsExecutedPerSec | The number of
actions this operator has executed per second. | Meter |
+| **Agent** | numOfInputRunsSucceeded | The number of
input runs that reached the run-completion boundary. | Count |
+| **Agent** | numOfInputRunsFailed | The number of
input runs terminated by an unhandled exception. | Count |
+| **Agent** | inputRunLatencyMs | End-to-end
input-run latency from entering the agent operator to completion or failure,
including time queued behind another input with the same key. | Histogram |
+| **Agent** | inputRunQueueLatencyMs | Time from
entering the agent operator until the input run starts processing. | Histogram |
+| **Agent** | inputRunProcessingLatencyMs | Time from the
input-run start boundary until completion or failure. | Histogram |
+| **Agent** | numOfPendingInputEvents | Current
number of input Events buffered behind an active run with the same key. | Gauge
|
+| **Agent** | numOfActiveInputRuns | Current
number of logical input runs that are processing or waiting for asynchronous
work. | Gauge |
| **Action** | action.\<action_name\>.numOfActionsExecuted | The total number
of actions this operator has executed for a specific action name. | Count |
| **Action** | action.\<action_name\>.numOfActionsExecutedPerSec | The number
of actions this operator has executed per second for a specific action name. |
Meter |
+| **Action** | action.\<action_name\>.actionSchedulingLatencyMs | Time from
enqueuing the initial Action task until it is selected for execution. |
Histogram |
+| **Action** | action.\<action_name\>.actionExecutionLatencyMs | End-to-end
latency of one logical Action execution, including asynchronous waits and
continuations. | Histogram |
+| **Action** | action.\<action_name\>.numOfPendingActionTasks | Current number
of physical Action task segments waiting to run, including continuations. |
Gauge |
+| **Action** | action.\<action_name\>.numOfActiveActionExecutions | Current
number of logical Action executions that have started but have not reached a
terminal state. | Gauge |
| **Agent** | eventLogTruncatedEvents | Number of
event log records whose payload was truncated at `STANDARD` level. Increments
once per event, regardless of how many fields inside it were truncated. Use
this to decide whether to raise truncation thresholds or move specific event
types to `VERBOSE`. | Count |
| **Agent** | eventLogWriteFailures | Number of
Event Log write attempts for which `append`, `flush`, or both failed. Event Log
writes are best-effort and do not fail the job. | Count |
+For a locally observed input run, `inputRunLatencyMs` is split into queueing
and processing time at the input-run start boundary. `numOfPendingInputEvents`
counts buffered inputs, while `numOfActiveInputRuns` counts logical runs; an
asynchronous run remains active while it is waiting for its continuation.
+
+An Action execution can be active while one of its continuation tasks is
pending, so `numOfActiveActionExecutions` and `numOfPendingActionTasks` are
independent. Action scheduling latency is recorded only for the initial task;
continuation queueing does not create another scheduling sample.
+
+Input-run outcomes and all latency samples are process-local. Runs or Action
executions already in flight when a task is restored do not produce latency
samples because their original timestamps are unavailable. An input Event
restored from the pending queue can still produce an outcome and
processing-latency sample after it starts in the new task attempt, but it does
not produce queue or end-to-end latency. Current-count gauges are rebuilt from
Flink state after restore.
+
+#### Execution Metrics
+
+Execution metrics are derived from LLM and Tool execution lifecycle events.
The `model_resource`, `tool`, `skill`, and `mcp_server` scopes are independent
key-value scopes directly under an Action; none is nested under another. The
existing `model` scope remains dedicated to model usage metrics.
+
+| Scope | Metrics | Description | Type |
+|-------|---------|-------------|------|
+| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.numOfLlmCallsSucceeded
| The number of framework-observed model invocations that returned
successfully. | Count |
+| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.numOfLlmCallsFailed |
The number of framework-observed model invocations that failed. | Count |
+| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.llmCallLatencyMs |
Latency of each framework-observed model invocation, excluding
structured-output parsing and retry wait time. | Histogram |
+| **Model Resource** |
action.\<action_name\>.model_resource.\<resource_name\>.retryCount | The number
of additional model invocations initiated by framework retry logic. Only
recorded when at least one retry occurs. See [retry-wait-interval]({{< ref
"docs/operations/configuration#core-options" >}}). | Count |
Review Comment:
Updated the docs. Both retry metrics now state the
ErrorHandlingStrategy.RETRY gate, and the migration from
model.<connection_name> to model_resource.<resource_name> is called out
explicitly.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/CompileUtils.java:
##########
@@ -85,10 +88,13 @@ private static <K, IN, OUT> DataStream<OUT> connectToAgent(
TypeInformation<OUT> outTypeInformation,
boolean inputIsJava,
boolean pythonKeyIsPickled) {
+ String agentName = agentPlan.getAgentName();
+ String operatorName =
Review Comment:
Updated monitoring.md with the migration behavior: operator_name now uses
the Agent name, falls back to action-execute-operator only when unavailable,
and dashboards filtering the old value must be updated. The metric hierarchy is
unchanged.
--
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]
