weiqingy commented on code in PR #955:
URL: https://github.com/apache/flink-agents/pull/955#discussion_r3891281514


##########
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:
   `skillName` here comes straight from the model's tool arguments 
(`LoadSkillTool.java:72-74`, `skill_tools.py:93-95`). Nothing checks it against 
the registry.
   
   The `unknown` bucketing at `:62-68` works, but it cannot reach this line. 
`load_skill` is itself a registered `ResourceType.TOOL` 
(`AgentPlan.java:614-620`), so the tool key resolves fine and the raw string 
flows into the `skill` scope just below. Each new value allocates a sub-group, 
a counter and a `DescriptiveStatisticsHistogram(100)` 
(`FlinkAgentsMetricGroupImpl.java:105-110`) for the life of the TaskManager, 
plus a Prometheus label. One made-up skill name per run is enough to grow it 
without bound. `monitoring.md:83` says cardinality is bounded, which is true 
for `tool`.
   
   Both runtimes already have the registry to hand (`LoadSkillTool.java:102`, 
`skill_tools.py:121`), and the Trace record keeps the requested name either 
way. Would validating inside `getToolExecutionMetadata` be the cleaner spot, or 
would you rather pass a second predicate next to `isRegisteredTool`?



##########
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:
   This counter goes up whatever the outcome, and it can never see a failed 
one. `LoadSkillTool` returns `ToolResponse.success(...)` on all seven paths and 
never calls `ToolResponse.error`. Two of those paths are errors: skill not 
found at `:105`, resource not found at `:143`. Python does the same at 
`skill_tools.py:116, :125, :151`.
   
   So `load_skill(name="does-not-exist")` raises 
`numOfToolCallsSucceeded{tool=load_skill}`, raises 
`numOfSkillLoads{skill=does-not-exist}`, and records a latency sample. No 
failure shows up anywhere.
   
   Returning an error would not cost the model the "Available skills: ..." 
hint, since the failure branch already puts `response.getError()` into the TOOL 
message (`ChatModelAction.java:661-664`). This also looks separate from #956, 
which is about Python having no error-result type. Here Java has 
`ToolResponse.error` and uses it nowhere. Should the not-found paths return it, 
with a matching change on the Python side?



##########
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:
   This teaches the Action scope about replay. The LLM and Tool scopes have no 
equivalent.
   
   `durableExecuteCompletionOnly` returns a cached result at 
`RunnerContextImpl.java:552` without calling the callable, but the reports 
around it fire either way. `ChatModelInvoker.java:157` starts and `:174-175` 
succeeds around `durableExecute` at `:162-163`, and 
`ToolCallAction.java:112-113` / `:198-199` has the same shape. So a replay 
looks like a start and a finish microseconds apart: `numOfLlmCallsSucceeded` 
goes up and `llmCallLatencyMs` takes a ~0 ms sample for a call that never ran, 
which is not what `monitoring.md:81` promises. The `isCompleted()` shortcut at 
`ActionExecutionOperator.java:468` only catches the case where the whole Action 
is complete, not the recovery where just the inner call is cached.
   
   This only reaches jobs that turn on an action state store 
(`ACTION_STATE_STORE_BACKEND` defaults to null, 
`AgentConfigOptions.java:65-66`). One knock-on if you go this way: 
`BuiltInExecutionMetrics` returns on any non-terminal type at `:76-78`, so a 
reused event would need to drain `activeExecutionStartNanos` there too. Is 
emitting `executionReused()` on the cache hit, handled the way you handle it 
here, the direction you would take?



##########
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:
   Two things are riding on these rows.
   
   The scope key and the value both moved. It was `getSubGroup("model", 
chatModel.getConnectionName())` before this PR and is 
`getSubGroup("model_resource", modelResource)` now 
(`ChatModelAction.java:205-217`, and `_record_retry_metrics` matches). The 
resource name is the better identity. But every existing `retryCount` and 
`retryWaitSec` series breaks, and neither doc mentions it.
   
   The `ErrorHandlingStrategy.RETRY` qualifier the old rows carried is also 
gone (`grep -c ErrorHandlingStrategy monitoring.md` returns 0), while 
`configuration.md:133`, edited in this same PR, still has it. The gate has not 
changed and is off by default (`AgentExecutionOptions.java:24-28`, 
`ChatModelAction.java:383-394`), so on a default config both metrics stay at 
zero and only one of the two docs explains why.
   
   Row 70's "Only recorded when at least one retry occurs" half covers it, and 
`retryWaitSec` has nothing. Worth putting the qualifier back on both rows and 
adding a line about the scope move?



##########
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:
   Neither `Math.max` is exercised by the new tests. `BuiltInActionMetricsTest` 
never dequeues while the pending count is already 0. 
`duplicateTerminalNotificationDoesNotUnderflowActiveGauge` looks like the 
coverage, but the idempotency it proves sits a level up: 
`activeInputRunIds.remove(inputRunId)` short-circuits on the second call 
(`BuiltInInputRunMetrics.java:145`), so `decrement()` at `:151` never runs. 
`restoreActiveInputRuns` also pre-clamps at `:132` before it calls `set`.
   
   I dropped both `Math.max` calls and the runtime suite still passed 688/688. 
So nothing in the ~880 new test lines tells a clamped gauge apart from an 
unclamped one.
   
   Two assertions would close it: `decrement()` at 0 stays 0, and `set(-1)` 
gives 0. Where would those sit best, a small `CurrentCountGaugeTest`, or an 
unbalanced sequence routed through the higher-level classes?



##########
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:
   `monitoring.md:164` documents the prefix format, and this PR adds that 
`<operator_name>` is the agent name, so the end state is written down. What no 
doc says is that existing series *move*. Everything this operator publishes 
used to sit under `action-execute-operator` and now sits under the agent name, 
so custom `agent_metric_group` and `action_metric_group` metrics break, along 
with any dashboard keyed on the old prefix. Your own `TokenMetricsE2ETest` edit 
shows how far it reaches: a *token* metric's expected prefix had to change.
   
   The PR body calls this "the existing operator name retained as a fallback", 
which reads gentler than what it means for anyone already scraping these. How 
would you want to flag the rename, a line in `monitoring.md` or a release note?



##########
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:
   `actionMetricGroups` is built once from the current plan at `:83-89`. But 
`restoreActionTask` reaches this lookup at `:151-155` with a name read from a 
deserialized `ActionTask` (`ActionExecutionOperator.java:845-853`). The live 
callers all take their names from the current plan, so the strict throw fits 
them. The restore path is the odd one out.
   
   I could not build a restore that actually hits this, so it is a question 
rather than a claim. The case I have in mind is a savepoint holding in-flight 
action tasks, restored after an action was renamed or removed. 
`tryResumeProcessActionTasks` runs inside `open()` (`:239`), so the throw would 
fail operator startup rather than just lose a metric, and skipping an unknown 
action or logging once looks like about three lines. Does the state format 
tolerate a plan change across a restore today?



-- 
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]

Reply via email to