This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 1aef675c42 [Improve][Zeta] Add RequestSlotOperation observability
metrics (#11409)
1aef675c42 is described below
commit 1aef675c42939140502a6be87a12f349557dfb1a
Author: Doyeon Kim <[email protected]>
AuthorDate: Sat Jul 18 23:10:12 2026 +0900
[Improve][Zeta] Add RequestSlotOperation observability metrics (#11409)
---
docs/en/engines/zeta/telemetry.md | 27 ++++++-
docs/zh/engines/zeta/telemetry.md | 22 ++++-
.../engine/server/CoordinatorService.java | 13 +++
.../resourcemanager/AbstractResourceManager.java | 40 ++++++++++
.../server/resourcemanager/ResourceManager.java | 3 +
.../resourcemanager/ResourceRequestHandler.java | 14 ++++
.../metrics/ExportsInstanceInitializer.java | 3 +
.../metrics/entity/RequestSlotOperationStats.java | 37 +++++++++
.../exports/RequestSlotOperationExports.java | 82 +++++++++++++++++++
.../resourcemanager/ResourceManagerTest.java | 41 ++++++++++
.../TelemetryCollectorCoordinatorGuardTest.java | 93 ++++++++++++++++++++++
11 files changed, 373 insertions(+), 2 deletions(-)
diff --git a/docs/en/engines/zeta/telemetry.md
b/docs/en/engines/zeta/telemetry.md
index 7c91348a1d..1ccdf0ff16 100644
--- a/docs/en/engines/zeta/telemetry.md
+++ b/docs/en/engines/zeta/telemetry.md
@@ -155,6 +155,31 @@
engine_state_store_connector_jar_total_references{backend="hazelcast"}
| report_metrics_operation_last_invocation_latency_ms | Gauge | **address**,
worker instance address,for example: "127.0.0.1:5801"
| The most recent worker-side `ReportMetricsOperation` reporting latency
in milliseconds, including local metrics collection and worker-to-master
invocation |
| report_metrics_operation_max_invocation_latency_ms | Gauge | **address**,
worker instance address,for example: "127.0.0.1:5801"
| The maximum observed worker-side `ReportMetricsOperation` reporting
latency in milliseconds since the worker started, including local metrics
collection and worker-to-master invocation |
+### Request Slot Operation
+
+These metrics expose the master-side slot allocation RPC path. When a job
needs execution
+resources, the active master selects candidate workers and sends
`RequestSlotOperation` requests to
+reserve slots on those workers. They are intended to help operators
distinguish slow or failed slot
+allocation RPCs from general resource shortage.
+
+These metrics are exported by the active master only. They are aggregate
signals and do not include
+job, worker, slot, or resource-profile labels.
+
+| MetricName | Type | Labels
| DESCRIPTION
|
+|------------------------------------------------------|---------|-----------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------|
+| request_slot_operation_total | Counter |
**address**, master instance address,for example: "127.0.0.1:5801". **result**,
one of "success" "no_slot" "failure" | The total number of
`RequestSlotOperation` invocations sent by the master to workers
|
+| request_slot_operation_last_invocation_latency_ms | Gauge |
**address**, master instance address,for example: "127.0.0.1:5801"
| The most recent master-side
`RequestSlotOperation` invocation latency in milliseconds, including
master-to-worker invocation |
+| request_slot_operation_max_invocation_latency_ms | Gauge |
**address**, master instance address,for example: "127.0.0.1:5801"
| The maximum observed master-side
`RequestSlotOperation` invocation latency in milliseconds since the master
started, including master-to-worker invocation |
+
+The `result` label has the following meanings:
+
+- `success`: the worker returned an assigned slot.
+- `no_slot`: the request reached a worker and completed, but the worker did
not return a suitable
+ slot. A sustained increase may indicate that the master's worker-resource
view and the worker's
+ current slot state are diverging, or that concurrent allocation is consuming
slots between
+ pre-check and request execution.
+- `failure`: the master-to-worker invocation failed or the operation completed
exceptionally.
+
### Job info detail
| MetricName | Type | Labels
|
DESCRIPTION |
@@ -240,4 +265,4 @@ the
[Installation](https://grafana.com/docs/grafana/latest/setup-grafana/install
- Add Prometheus DataSource on Grafana.
- Import the `Seatunnel Cluster` monitoring dashboard JSON into Grafana.
-The [effect image](../../../images/grafana.png) of the dashboard
\ No newline at end of file
+The [effect image](../../../images/grafana.png) of the dashboard
diff --git a/docs/zh/engines/zeta/telemetry.md
b/docs/zh/engines/zeta/telemetry.md
index e90a320be4..44802d2b7c 100644
--- a/docs/zh/engines/zeta/telemetry.md
+++ b/docs/zh/engines/zeta/telemetry.md
@@ -154,6 +154,26 @@
engine_state_store_connector_jar_total_references{backend="hazelcast"}
| report_metrics_operation_last_invocation_latency_ms | Gauge |
**address**，worker 实例地址，例如："127.0.0.1:5801"
| worker 侧最近一次 `ReportMetricsOperation` 上报耗时，单位为毫秒，包含本地 metrics 收集和 worker 到
master 的调用时间 |
| report_metrics_operation_max_invocation_latency_ms | Gauge |
**address**，worker 实例地址，例如："127.0.0.1:5801"
| worker 启动以来观测到的 `ReportMetricsOperation` 最大上报耗时，单位为毫秒，包含本地 metrics 收集和
worker 到 master 的调用时间 |
+### RequestSlotOperation 指标
+
+这些指标暴露 master 侧的 slot 分配 RPC 路径。当作业需要执行资源时，active master 会选择候选 worker，并向这些
+worker 发送 `RequestSlotOperation` 请求以预留 slot。这些指标用于帮助运维人员区分 slot 分配 RPC
变慢/失败与整体资源不足。
+
+这些指标仅由 active master 输出。它们是聚合信号，不包含 job、worker、slot 或 resource profile 级别的标签。
+
+| MetricName | Type | Labels
|
描述
|
+|---------------------------------------------------|---------|--------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------|
+| request_slot_operation_total | Counter |
**address**，master 实例地址，例如："127.0.0.1:5801"。**result**，取值包括："success" "no_slot"
"failure" | master 发送到 worker 的 `RequestSlotOperation` 调用总次数
|
+| request_slot_operation_last_invocation_latency_ms | Gauge |
**address**，master 实例地址，例如："127.0.0.1:5801"
| master 侧最近一次 `RequestSlotOperation` 调用耗时，单位为毫秒，包含 master 到 worker 的调用时间 |
+| request_slot_operation_max_invocation_latency_ms | Gauge |
**address**，master 实例地址，例如："127.0.0.1:5801"
| master 启动以来观测到的 `RequestSlotOperation` 最大调用耗时，单位为毫秒，包含 master 到 worker
的调用时间 |
+
+`result` 标签含义如下：
+
+- `success`：worker 返回了已分配的 slot。
+- `no_slot`：请求到达 worker 并正常完成，但 worker 未返回合适的 slot。若该结果持续增加，可能表示 master 侧的
+ worker 资源视图与 worker 当前 slot 状态存在偏差，或者在 pre-check 与请求执行之间并发分配消耗了 slot。
+- `failure`：master 到 worker 的调用失败，或 operation 异常完成。
+
### 作业信息详细
| MetricName | Type | Labels
| 描述 |
@@ -239,4 +259,4 @@ scrape_configs:
- 在 Grafana 中添加 Prometheus 数据源。
- 将 `Seatunnel Cluster` 监控仪表板 JSON 导入到 Grafana 中。
-监控[效果图](../../../images/grafana.png)
\ No newline at end of file
+监控[效果图](../../../images/grafana.png)
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
index c58a29ad20..df2134cd98 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
@@ -1166,6 +1166,19 @@ public class CoordinatorService {
return resourceManager;
}
+ /**
+ * Returns the resource manager only when it has already been initialized.
+ *
+ * <p>Unlike {@link #getResourceManager()}, this method never creates or
initializes runtime
+ * state. Read-only paths such as telemetry collection should use this
method to avoid
+ * triggering cluster RPCs.
+ *
+ * @return the initialized resource manager, or {@code null} when it has
not been initialized
+ */
+ public ResourceManager getInitializedResourceManager() {
+ return resourceManager;
+ }
+
/** call by client to submit job */
public PassiveCompletableFuture<Void> submitJob(
long jobId, Data jobImmutableInformation, boolean
isStartWithSavePoint) {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/AbstractResourceManager.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/AbstractResourceManager.java
index 32cf63e06e..dc0059af34 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/AbstractResourceManager.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/AbstractResourceManager.java
@@ -30,6 +30,7 @@ import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.SyncWorkerPr
import
org.apache.seatunnel.engine.server.resourcemanager.resource.ResourceProfile;
import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile;
import org.apache.seatunnel.engine.server.resourcemanager.worker.WorkerProfile;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.RequestSlotOperationStats;
import org.apache.seatunnel.engine.server.utils.NodeEngineUtil;
import com.hazelcast.cluster.Address;
@@ -48,6 +49,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
@Slf4j
@@ -67,6 +69,13 @@ public abstract class AbstractResourceManager implements
ResourceManager {
@Getter private final SlotAllocationStrategy slotAllocationStrategy;
+ // Track master-side slot request cost without changing allocation
behavior.
+ private final AtomicLong requestSlotOperationSuccessCount = new
AtomicLong();
+ private final AtomicLong requestSlotOperationNoSlotCount = new
AtomicLong();
+ private final AtomicLong requestSlotOperationFailureCount = new
AtomicLong();
+ private final AtomicLong requestSlotOperationLastInvocationLatencyMs = new
AtomicLong();
+ private final AtomicLong requestSlotOperationMaxInvocationLatencyMs = new
AtomicLong();
+
public AbstractResourceManager(NodeEngine nodeEngine, EngineConfig
engineConfig) {
this.registerWorker = new ConcurrentHashMap<>();
this.nodeEngine = nodeEngine;
@@ -203,6 +212,37 @@ public abstract class AbstractResourceManager implements
ResourceManager {
NodeEngineUtil.sendOperationToMemberNode(nodeEngine,
operation, address));
}
+ void recordRequestSlotOperationSuccess(long elapsedMillis) {
+ updateRequestSlotOperationLatency(elapsedMillis);
+ requestSlotOperationSuccessCount.incrementAndGet();
+ }
+
+ void recordRequestSlotOperationNoSlot(long elapsedMillis) {
+ updateRequestSlotOperationLatency(elapsedMillis);
+ requestSlotOperationNoSlotCount.incrementAndGet();
+ }
+
+ void recordRequestSlotOperationFailure(long elapsedMillis) {
+ updateRequestSlotOperationLatency(elapsedMillis);
+ requestSlotOperationFailureCount.incrementAndGet();
+ }
+
+ private void updateRequestSlotOperationLatency(long elapsedMillis) {
+ requestSlotOperationLastInvocationLatencyMs.set(elapsedMillis);
+
requestSlotOperationMaxInvocationLatencyMs.accumulateAndGet(elapsedMillis,
Math::max);
+ }
+
+ /** Returns the latest master-side RequestSlotOperation observability
snapshot. */
+ @Override
+ public RequestSlotOperationStats getRequestSlotOperationStats() {
+ return new RequestSlotOperationStats(
+ requestSlotOperationSuccessCount.get(),
+ requestSlotOperationNoSlotCount.get(),
+ requestSlotOperationFailureCount.get(),
+ requestSlotOperationLastInvocationLatencyMs.get(),
+ requestSlotOperationMaxInvocationLatencyMs.get());
+ }
+
@Override
public CompletableFuture<Void> releaseResources(long jobId,
List<SlotProfile> profiles) {
CompletableFuture<Void> completableFuture = new CompletableFuture<>();
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManager.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManager.java
index 18623873c0..aadf5b75b7 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManager.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManager.java
@@ -21,6 +21,7 @@ import
org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture;
import
org.apache.seatunnel.engine.server.resourcemanager.resource.ResourceProfile;
import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile;
import org.apache.seatunnel.engine.server.resourcemanager.worker.WorkerProfile;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.RequestSlotOperationStats;
import com.hazelcast.cluster.Address;
import com.hazelcast.internal.services.MembershipServiceEvent;
@@ -71,4 +72,6 @@ public interface ResourceManager {
int workerCount(Map<String, String> tags);
ConcurrentMap<Address, WorkerProfile> getRegisterWorker();
+
+ RequestSlotOperationStats getRequestSlotOperationStats();
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceRequestHandler.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceRequestHandler.java
index 391e2d80b2..550922f58b 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceRequestHandler.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceRequestHandler.java
@@ -44,6 +44,7 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static com.hazelcast.jet.impl.util.ExceptionUtil.withTryCatch;
@@ -230,6 +231,7 @@ public class ResourceRequestHandler {
private CompletableFuture<SlotAndWorkerProfile>
singleResourceRequestToMember(
int i, ResourceProfile r, WorkerProfile workerProfile) {
+ long invocationStartNanos = System.nanoTime();
CompletableFuture<SlotAndWorkerProfile> future =
resourceManager.sendToMember(
new RequestSlotOperation(jobId, r),
workerProfile.getAddress());
@@ -237,15 +239,27 @@ public class ResourceRequestHandler {
withTryCatch(
LOGGER,
(slotAndWorkerProfile, error) -> {
+ long elapsedMillis =
elapsedMillisSince(invocationStartNanos);
if (error != null) {
+
resourceManager.recordRequestSlotOperationFailure(elapsedMillis);
throw new RuntimeException(error);
} else {
+ if (slotAndWorkerProfile.getSlotProfile() ==
null) {
+
resourceManager.recordRequestSlotOperationNoSlot(elapsedMillis);
+ } else {
+
resourceManager.recordRequestSlotOperationSuccess(
+ elapsedMillis);
+ }
resourceManager.heartbeat(slotAndWorkerProfile.getWorkerProfile());
addSlotToCacheMap(i,
slotAndWorkerProfile.getSlotProfile());
}
}));
}
+ private long elapsedMillisSince(long startNanos) {
+ return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ }
+
@VisibleForTesting
public Optional<WorkerProfile> preCheckWorkerResource(ResourceProfile r) {
List<WorkerProfile> workerProfiles =
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
index 4b759aba5f..920aa63036 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/ExportsInstanceInitializer.java
@@ -24,6 +24,7 @@ import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobMetricExp
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobThreadPoolStatusExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.NodeMetricExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.ReportMetricsOperationExports;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.RequestSlotOperationExports;
import com.hazelcast.instance.impl.Node;
import io.prometheus.client.CollectorRegistry;
@@ -50,6 +51,8 @@ public final class ExportsInstanceInitializer {
new NodeMetricExports(node).register(collectorRegistry);
// ReportMetricsOperation metrics
new
ReportMetricsOperationExports(node).register(collectorRegistry);
+ // RequestSlotOperation metrics
+ new RequestSlotOperationExports(node).register(collectorRegistry);
// Engine state store metrics
new
EngineStateStoreMetricExports(node).register(collectorRegistry);
// Engine state store logical metrics
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/entity/RequestSlotOperationStats.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/entity/RequestSlotOperationStats.java
new file mode 100644
index 0000000000..69b2ad8e93
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/entity/RequestSlotOperationStats.java
@@ -0,0 +1,37 @@
+/*
+ * 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.seatunnel.engine.server.telemetry.metrics.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+/** Snapshot of master-side RequestSlotOperation observability state. */
+@Data
+@AllArgsConstructor
+public class RequestSlotOperationStats {
+ /** Total request-slot invocations that returned an assigned slot. */
+ private long successCount;
+ /** Total request-slot invocations that completed but returned no slot. */
+ private long noSlotCount;
+ /** Total failed request-slot invocations. */
+ private long failureCount;
+ /** Most recent master-side request-slot invocation latency in
milliseconds. */
+ private long lastInvocationLatencyMs;
+ /** Maximum observed master-side request-slot invocation latency in
milliseconds. */
+ private long maxInvocationLatencyMs;
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/RequestSlotOperationExports.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/RequestSlotOperationExports.java
new file mode 100644
index 0000000000..5e36e560a6
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/telemetry/metrics/exports/RequestSlotOperationExports.java
@@ -0,0 +1,82 @@
+/*
+ * 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.seatunnel.engine.server.telemetry.metrics.exports;
+
+import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager;
+import org.apache.seatunnel.engine.server.telemetry.metrics.AbstractCollector;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.RequestSlotOperationStats;
+
+import com.hazelcast.instance.impl.Node;
+import io.prometheus.client.CounterMetricFamily;
+import io.prometheus.client.GaugeMetricFamily;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class RequestSlotOperationExports extends AbstractCollector {
+
+ public RequestSlotOperationExports(Node node) {
+ super(node);
+ }
+
+ @Override
+ public List<MetricFamilySamples> collect() {
+ List<MetricFamilySamples> mfs = new ArrayList<>();
+ if (!isMaster() || !isCoordinatorReady()) {
+ return mfs;
+ }
+
+ ResourceManager resourceManager =
getCoordinatorService().getInitializedResourceManager();
+ if (resourceManager == null) {
+ return mfs;
+ }
+
+ String address = localAddress();
+ RequestSlotOperationStats stats =
resourceManager.getRequestSlotOperationStats();
+
+ CounterMetricFamily totalMetricFamily =
+ new CounterMetricFamily(
+ "request_slot_operation",
+ "The total number of RequestSlotOperation invocations
sent by the master",
+ clusterLabelNames(ADDRESS, "result"));
+ totalMetricFamily.addMetric(labelValues(address, "success"),
stats.getSuccessCount());
+ totalMetricFamily.addMetric(labelValues(address, "no_slot"),
stats.getNoSlotCount());
+ totalMetricFamily.addMetric(labelValues(address, "failure"),
stats.getFailureCount());
+ mfs.add(totalMetricFamily);
+
+ GaugeMetricFamily lastLatencyMetricFamily =
+ new GaugeMetricFamily(
+ "request_slot_operation_last_invocation_latency_ms",
+ "The most recent master-side RequestSlotOperation
invocation latency "
+ + "in milliseconds, including master-to-worker
invocation",
+ clusterLabelNames(ADDRESS));
+ lastLatencyMetricFamily.addMetric(labelValues(address),
stats.getLastInvocationLatencyMs());
+ mfs.add(lastLatencyMetricFamily);
+
+ GaugeMetricFamily maxLatencyMetricFamily =
+ new GaugeMetricFamily(
+ "request_slot_operation_max_invocation_latency_ms",
+ "The maximum observed master-side RequestSlotOperation
invocation "
+ + "latency in milliseconds since the master
started, including "
+ + "master-to-worker invocation",
+ clusterLabelNames(ADDRESS));
+ maxLatencyMetricFamily.addMetric(labelValues(address),
stats.getMaxInvocationLatencyMs());
+ mfs.add(maxLatencyMetricFamily);
+ return mfs;
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManagerTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManagerTest.java
index 4e4293d8a7..0e76b5114c 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManagerTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/ResourceManagerTest.java
@@ -25,6 +25,7 @@ import
org.apache.seatunnel.engine.server.resourcemanager.resource.Memory;
import
org.apache.seatunnel.engine.server.resourcemanager.resource.ResourceProfile;
import org.apache.seatunnel.engine.server.resourcemanager.resource.SlotProfile;
import org.apache.seatunnel.engine.server.resourcemanager.worker.WorkerProfile;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.RequestSlotOperationStats;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
@@ -167,6 +168,41 @@ public class ResourceManagerTest extends
AbstractSeaTunnelServerTest<ResourceMan
exception2.getCause().getMessage());
}
+ @Test
+ public void testRequestSlotOperationStatsForSuccessAndNoSlot()
+ throws ExecutionException, InterruptedException {
+ FakeResourceManagerForRequestSlotRetryTest resourceManager =
+ new FakeResourceManagerForRequestSlotRetryTest(nodeEngine, 2,
1);
+
+ List<SlotProfile> slotProfiles =
+ resourceManager
+ .applyResources(1L, Collections.singletonList(new
ResourceProfile()), null)
+ .get();
+
+ Assertions.assertEquals(1, slotProfiles.size());
+ RequestSlotOperationStats stats =
resourceManager.getRequestSlotOperationStats();
+ Assertions.assertEquals(1, stats.getSuccessCount());
+ Assertions.assertEquals(1, stats.getNoSlotCount());
+ Assertions.assertEquals(0, stats.getFailureCount());
+ }
+
+ @Test
+ public void testRequestSlotOperationStatsAggregation() {
+ FakeResourceManagerForRequestSlotRetryTest resourceManager =
+ new FakeResourceManagerForRequestSlotRetryTest(nodeEngine, 0,
0);
+
+ resourceManager.recordRequestSlotOperationSuccess(10L);
+ resourceManager.recordRequestSlotOperationNoSlot(20L);
+ resourceManager.recordRequestSlotOperationFailure(15L);
+
+ RequestSlotOperationStats stats =
resourceManager.getRequestSlotOperationStats();
+ Assertions.assertEquals(1, stats.getSuccessCount());
+ Assertions.assertEquals(1, stats.getNoSlotCount());
+ Assertions.assertEquals(1, stats.getFailureCount());
+ Assertions.assertEquals(15L, stats.getLastInvocationLatencyMs());
+ Assertions.assertEquals(20L, stats.getMaxInvocationLatencyMs());
+ }
+
@Test
public void testApplyResourcesPreserveCauseForExternalException()
throws ExecutionException, InterruptedException {
@@ -182,6 +218,11 @@ public class ResourceManagerTest extends
AbstractSeaTunnelServerTest<ResourceMan
Assertions.assertInstanceOf(
IllegalStateException.class,
((NoEnoughResourceException) exception.getCause()).getCause());
+
+ RequestSlotOperationStats stats =
resourceManager.getRequestSlotOperationStats();
+ Assertions.assertEquals(0, stats.getSuccessCount());
+ Assertions.assertEquals(0, stats.getNoSlotCount());
+ Assertions.assertEquals(1, stats.getFailureCount());
}
@Test
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
index 03edb148a4..c4e603d042 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/telemetry/metrics/TelemetryCollectorCoordinatorGuardTest.java
@@ -20,13 +20,16 @@ package
org.apache.seatunnel.engine.server.telemetry.metrics;
import org.apache.seatunnel.engine.server.CoordinatorService;
import org.apache.seatunnel.engine.server.SeaTunnelServer;
import org.apache.seatunnel.engine.server.TaskExecutionService;
+import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager;
import org.apache.seatunnel.engine.server.telemetry.metrics.entity.JobCounter;
import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.ReportMetricsOperationStats;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.RequestSlotOperationStats;
import
org.apache.seatunnel.engine.server.telemetry.metrics.entity.ThreadPoolStatus;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.ClusterMetricExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobMetricExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.JobThreadPoolStatusExports;
import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.ReportMetricsOperationExports;
+import
org.apache.seatunnel.engine.server.telemetry.metrics.exports.RequestSlotOperationExports;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -254,6 +257,96 @@ public class TelemetryCollectorCoordinatorGuardTest {
assertSingleMetricSample(maxLatencyMetric, 27D);
}
+ @Test
+ void testRequestSlotOperationExportsReturnsEmptyWhenNotMaster() {
+ Mockito.when(mockNode.isMaster()).thenReturn(false);
+
+ RequestSlotOperationExports exports = new
RequestSlotOperationExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(result.isEmpty(), "collect() must return empty
on non-master node");
+ Mockito.verify(mockServer, Mockito.never()).isCoordinatorActive();
+ }
+
+ @Test
+ void testRequestSlotOperationExportsReturnsEmptyWhenCoordinatorNotReady() {
+ Mockito.when(mockNode.isMaster()).thenReturn(true);
+ Mockito.when(mockServer.isCoordinatorActive()).thenReturn(false);
+
+ RequestSlotOperationExports exports = new
RequestSlotOperationExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(
+ result.isEmpty(),
+ "collect() must return empty when coordinator is not ready"
+ + " to avoid initializing resource manager from scrape
path");
+ Mockito.verify(mockServer, Mockito.never()).getCoordinatorService();
+ }
+
+ @Test
+ void testRequestSlotOperationExportsDoesNotInitializeResourceManager() {
+ Mockito.when(mockNode.isMaster()).thenReturn(true);
+ Mockito.when(mockServer.isCoordinatorActive()).thenReturn(true);
+
+ RequestSlotOperationExports exports = new
RequestSlotOperationExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertTrue(
+ result.isEmpty(),
+ "collect() must return empty when resource manager has not
been initialized");
+ Mockito.verify(mockCoordinatorService).getInitializedResourceManager();
+ Mockito.verify(mockCoordinatorService,
Mockito.never()).getResourceManager();
+ }
+
+ @Test
+ void testRequestSlotOperationExportsReturnsMetricsWhenCoordinatorReady() {
+ Mockito.when(mockNode.isMaster()).thenReturn(true);
+ Mockito.when(mockServer.isCoordinatorActive()).thenReturn(true);
+ ResourceManager resourceManager = Mockito.mock(ResourceManager.class);
+ Mockito.when(mockCoordinatorService.getInitializedResourceManager())
+ .thenReturn(resourceManager);
+ Mockito.when(resourceManager.getRequestSlotOperationStats())
+ .thenReturn(new RequestSlotOperationStats(4L, 2L, 1L, 18L,
44L));
+
+ RequestSlotOperationExports exports = new
RequestSlotOperationExports(mockNode);
+ List<Collector.MetricFamilySamples> result = exports.collect();
+
+ Assertions.assertFalse(
+ result.isEmpty(), "collect() must return metrics when
coordinator is ready");
+ Collector.MetricFamilySamples totalMetric =
+ result.stream()
+ .filter(s -> "request_slot_operation".equals(s.name))
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(totalMetric);
+ Assertions.assertEquals(3, totalMetric.samples.size());
+ assertMetricSample(totalMetric, "request_slot_operation_total",
"success", 4D);
+ assertMetricSample(totalMetric, "request_slot_operation_total",
"no_slot", 2D);
+ assertMetricSample(totalMetric, "request_slot_operation_total",
"failure", 1D);
+
+ Collector.MetricFamilySamples lastLatencyMetric =
+ result.stream()
+ .filter(
+ s ->
+
"request_slot_operation_last_invocation_latency_ms"
+ .equals(s.name))
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(lastLatencyMetric);
+ assertSingleMetricSample(lastLatencyMetric, 18D);
+
+ Collector.MetricFamilySamples maxLatencyMetric =
+ result.stream()
+ .filter(
+ s ->
+
"request_slot_operation_max_invocation_latency_ms"
+ .equals(s.name))
+ .findFirst()
+ .orElse(null);
+ Assertions.assertNotNull(maxLatencyMetric);
+ assertSingleMetricSample(maxLatencyMetric, 44D);
+ }
+
//
-------------------------------------------------------------------------
// ClusterMetricExports
//
-------------------------------------------------------------------------
