weiqingy commented on code in PR #955:
URL: https://github.com/apache/flink-agents/pull/955#discussion_r3891684985
##########
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";
Review Comment:
These four names land in the same group users get for their own metrics.
`RunnerContextImpl.java:248-250` returns
`agentMetricGroup.getSubGroup("action", actionName)`, the same instance
`BuiltInMetrics.java:190` hands to `BuiltInActionMetrics`, and
`FlinkAgentsMetricGroupImpl.java:77-82` returns the cached metric when the name
already exists. So
`ctx.getActionMetricGroup().getGauge("numOfPendingActionTasks")` returns the
framework gauge, and whichever writer runs last wins.
`monitoring.md:106` points users at this group for custom metrics, and this
PR adds four reserved names to it. Is one sentence reserving the built-in names
enough, or would you rather the group reject a collision outright?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInExecutionMetrics.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * 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.Event;
+import org.apache.flink.agents.api.trace.ExecutionLifecycleEvents;
+import org.apache.flink.agents.api.trace.ExecutionTraceContext;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+import java.util.function.Predicate;
+
+/** Derives built-in LLM and Tool metrics from execution lifecycle events. */
+final class BuiltInExecutionMetrics {
+
+ private final FlinkAgentsMetricGroupImpl agentMetricGroup;
+ private final LongSupplier nanoTime;
+ private final Map<String, ExecutionMetricRecorder>
metricRecordersByEntityType;
+ private final Map<String, Long> activeExecutionStartNanos = new
HashMap<>();
Review Comment:
`activeExecutionStartNanos` is filled at `:67` and drained only by a
matching `EXECUTION_FINISHED` or `EXECUTION_FAILED` at `:81`. No cap and no
sweep, and the object is built once per operator (`BuiltInMetrics.java:83-84`),
so a start that never gets its terminal stays for the operator's lifetime.
The map one layer up is swept. `RunnerContextImpl.java:417-430` pairs the
same starts and terminals in `activeReportedExecutions`, and
`ActionTaskContextManager.java:306-308` drops that map per action execution.
The same orphan is cleaned up there and kept here.
Framework code still looks balanced on every path, so this needs a user
action to trigger. `ExecutionReporters` is public and dispatches on `ctx
instanceof ExecutionReporter` (`ExecutionReporters.java:99-101`), so an action
that reports an `llm` or `tool` start and no terminal leaks one entry per call.
Is the operator-lifetime map deliberate, or should it be swept when the
action execution completes?
##########
runtime/src/main/java/org/apache/flink/agents/runtime/metrics/BuiltInMetrics.java:
##########
@@ -41,9 +48,27 @@ public class BuiltInMetrics {
private final Counter eventLogWriteFailures;
- private final HashMap<String, BuiltInActionMetrics> actionMetricGroups;
+ private final BuiltInInputRunMetrics inputRunMetrics;
+
+ private final BuiltInExecutionMetrics executionMetrics;
+
+ private final Map<String, BuiltInActionMetrics> actionMetricGroups;
public BuiltInMetrics(FlinkAgentsMetricGroupImpl parentMetricGroup,
AgentPlan agentPlan) {
+ this(
+ parentMetricGroup,
+ agentPlan,
+ toolName -> {
+ Map<String, ?> tools =
agentPlan.getResourceProviders().get(ResourceType.TOOL);
Review Comment:
nit: this defines a registered tool as a key in
`agentPlan.getResourceProviders().get(ResourceType.TOOL)`. Production passes
something wider: `toolName -> resourceCache.hasResource(toolName,
ResourceType.TOOL)` (`ActionExecutionOperator.java:193-197`), and
`ResourceCache.java:101-108` checks the cache first, so it also matches
resources put in with no provider.
Both callers of this constructor are tests (`EventRouterTest.java:291`,
`BuiltInMetricsTest.java:39`), so the definition that is easiest to read is the
one production never runs. `BuiltInMetrics` has no access to the cache, so is
dropping the convenience constructor and letting the two tests pass their own
predicate the simpler end state?
##########
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:
Makes sense, and #956 is the right home for the error-result change.
--
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]
