This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] 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 0071e8fa60 [Feature][Zeta] Add worker resource REST API (#11909)
0071e8fa60 is described below
commit 0071e8fa608a6b25fb0dd4bab044bc41a4e263b8
Author: Goutam Adwant <[email protected]>
AuthorDate: Sat Aug 29 08:06:20 2026 +0000
[Feature][Zeta] Add worker resource REST API (#11909)
Signed-off-by: goutamadwant <[email protected]>
---
docs/en/engines/zeta/rest-api-v1.md | 47 +++++
docs/en/engines/zeta/rest-api-v2.md | 67 +++++++
docs/zh/engines/zeta/rest-api-v1.md | 47 +++++
docs/zh/engines/zeta/rest-api-v2.md | 67 +++++++
.../seatunnel/engine/server/JettyService.java | 5 +
.../diagnostic/PendingDiagnosticsCollector.java | 91 ++++++---
.../diagnostic/WorkerResourceDiagnostic.java | 15 ++
...Diagnostic.java => WorkerResourceSnapshot.java} | 18 +-
.../opeartion/GetWorkerResourcesOperation.java | 63 +++++++
.../seatunnel/engine/server/rest/RestConstant.java | 1 +
.../server/rest/RestHttpGetCommandProcessor.java | 12 ++
.../server/rest/service/WorkerResourceService.java | 89 +++++++++
.../server/rest/servlet/WorkerResourceServlet.java | 49 +++++
.../serializable/ResourceDataSerializerHook.java | 5 +
.../PendingDiagnosticsCollectorTest.java | 205 +++++++++++++++++++++
.../opeartion/GetWorkerResourcesOperationTest.java | 61 ++++++
.../rest/service/WorkerResourceServiceTest.java | 119 ++++++++++++
.../rest/servlet/WorkerResourceServletTest.java | 59 ++++++
.../ResourceDataSerializerHookTest.java} | 43 +++--
19 files changed, 1004 insertions(+), 59 deletions(-)
diff --git a/docs/en/engines/zeta/rest-api-v1.md
b/docs/en/engines/zeta/rest-api-v1.md
index 5c353868cd..7d43d11c92 100644
--- a/docs/en/engines/zeta/rest-api-v1.md
+++ b/docs/en/engines/zeta/rest-api-v1.md
@@ -166,6 +166,53 @@ network:
------------------------------------------------------------------------------------------
+### Query Worker Resources
+
+<details>
+ <summary><code>GET</code>
<code><b>/hazelcast/rest/maps/resource/workers</b></code> <code>(Returns the
current resource snapshot for registered workers.)</code></summary>
+
+#### Parameters
+
+None.
+
+#### Responses
+
+```json
+{
+ "available": true,
+ "collectedAt": 1723017600000,
+ "workers": [
+ {
+ "address": "10.0.0.8:5801",
+ "tags": {"region": "us-west"},
+ "totalSlots": 4,
+ "freeSlots": 1,
+ "usedSlots": 3,
+ "dynamicSlot": false,
+ "totalCpuCores": 8,
+ "availableCpuCores": 2,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 4294967296,
+ "cpuUsage": 0.42,
+ "memUsage": 0.58,
+ "runningJobIds": [123456789]
+ }
+ ]
+}
+```
+
+**Notes:**
+
+- Fixed-slot workers return `totalSlots`, `usedSlots`, and `freeSlots`.
+- Dynamic-slot workers do not have a fixed slot capacity. For them,
`totalSlots` is the number of currently tracked assigned and unassigned slots,
while `freeSlots` is the currently unassigned count. Use `dynamicSlot` together
with the CPU and heap fields when interpreting capacity.
+- When `available` is `false`, `workers` is empty because the master snapshot
cannot currently be read. Clients should retry instead of interpreting the
response as an empty cluster.
+- `collectedAt` records when the master built this response. Worker values
come from the latest resource-manager heartbeat and are not an atomic sample
with `/system-monitoring-information`.
+- Resource and usage fields are omitted until the worker heartbeat contains
those values.
+
+</details>
+
+------------------------------------------------------------------------------------------
+
### Returns thread dump information for the current node.
<details>
diff --git a/docs/en/engines/zeta/rest-api-v2.md
b/docs/en/engines/zeta/rest-api-v2.md
index 4cbf697ca6..47bc667fdf 100644
--- a/docs/en/engines/zeta/rest-api-v2.md
+++ b/docs/en/engines/zeta/rest-api-v2.md
@@ -253,6 +253,68 @@ Please refer [security](security.md)
------------------------------------------------------------------------------------------
+### Query Worker Resources
+
+<details>
+ <summary><code>GET</code> <code><b>/resource/workers</b></code>
<code>(Returns the current resource snapshot for registered
workers.)</code></summary>
+
+#### Parameters
+
+None.
+
+#### Responses
+
+```json
+{
+ "available": true,
+ "collectedAt": 1723017600000,
+ "workers": [
+ {
+ "address": "10.0.0.8:5801",
+ "tags": {"region": "us-west"},
+ "totalSlots": 4,
+ "freeSlots": 1,
+ "usedSlots": 3,
+ "dynamicSlot": false,
+ "totalCpuCores": 8,
+ "availableCpuCores": 2,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 4294967296,
+ "cpuUsage": 0.42,
+ "memUsage": 0.58,
+ "runningJobIds": [123456789]
+ },
+ {
+ "address": "10.0.0.9:5801",
+ "tags": {},
+ "totalSlots": 2,
+ "freeSlots": 0,
+ "usedSlots": 2,
+ "dynamicSlot": true,
+ "totalCpuCores": 8,
+ "availableCpuCores": 4,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 8589934592,
+ "cpuUsage": 0.35,
+ "memUsage": 0.41,
+ "runningJobIds": [123456789]
+ }
+ ]
+}
+```
+
+**Notes:**
+
+- Fixed-slot workers return `totalSlots`, `usedSlots`, and `freeSlots`.
+- Dynamic-slot workers do not have a fixed slot capacity. For them,
`totalSlots` is the number of currently tracked assigned and unassigned slots,
while `freeSlots` is the currently unassigned count. Use `dynamicSlot` together
with the CPU and heap fields when interpreting capacity.
+- `available` is `false` when the master resource snapshot cannot be read,
including the master-election window. In that case, `workers` is empty and
clients should retry instead of interpreting the response as an empty cluster.
+- `collectedAt` is the timestamp in milliseconds when the master built this
response. Worker values come from the latest resource-manager heartbeat and are
not an atomic sample with `/system-monitoring-information`.
+- Resource and usage fields are omitted until the worker heartbeat contains
those values.
+
+</details>
+
+------------------------------------------------------------------------------------------
+
### Query An Overview And State Of Running Jobs
<details>
@@ -343,7 +405,12 @@ Please refer [security](security.md)
},
"totalSlots": 4,
"freeSlots": 0,
+ "usedSlots": 4,
"dynamicSlot": false,
+ "totalCpuCores": 8,
+ "availableCpuCores": 2,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 4294967296,
"cpuUsage": 0.83,
"memUsage": 0.64,
"runningJobIds": [
diff --git a/docs/zh/engines/zeta/rest-api-v1.md
b/docs/zh/engines/zeta/rest-api-v1.md
index f1d43021c9..d2f4e359a8 100644
--- a/docs/zh/engines/zeta/rest-api-v1.md
+++ b/docs/zh/engines/zeta/rest-api-v1.md
@@ -165,6 +165,53 @@ network:
------------------------------------------------------------------------------------------
+### 查询 Worker 资源
+
+<details>
+ <summary><code>GET</code>
<code><b>/hazelcast/rest/maps/resource/workers</b></code> <code>(返回已注册 Worker
的当前资源快照。)</code></summary>
+
+#### 参数
+
+无。
+
+#### 响应
+
+```json
+{
+ "available": true,
+ "collectedAt": 1723017600000,
+ "workers": [
+ {
+ "address": "10.0.0.8:5801",
+ "tags": {"region": "us-west"},
+ "totalSlots": 4,
+ "freeSlots": 1,
+ "usedSlots": 3,
+ "dynamicSlot": false,
+ "totalCpuCores": 8,
+ "availableCpuCores": 2,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 4294967296,
+ "cpuUsage": 0.42,
+ "memUsage": 0.58,
+ "runningJobIds": [123456789]
+ }
+ ]
+}
+```
+
+**说明：**
+
+- 固定 Slot 模式的 Worker 返回 `totalSlots`、`usedSlots` 和 `freeSlots`。
+- 动态 Slot 模式的 Worker 没有固定的 Slot 容量。此时，`totalSlots` 表示当前已跟踪的已分配和未分配 Slot
总数，`freeSlots` 表示当前未分配数量。解释容量时，请结合 `dynamicSlot` 以及 CPU 和堆内存字段。
+- 当 `available` 为 `false` 时，表示当前无法读取 Master 资源快照，`workers`
为空。客户端应重试，而不应将该响应解释为空集群。
+- `collectedAt` 表示 Master 构建本次响应的时间。Worker 字段来自资源管理器收到的最近一次心跳，并不与
`/system-monitoring-information` 构成原子快照。
+- 如果最近一次 Worker 心跳尚未包含资源或使用率数据，对应字段不会返回。
+
+</details>
+
+------------------------------------------------------------------------------------------
+
### 返回当前节点的线程堆栈信息。
<details>
diff --git a/docs/zh/engines/zeta/rest-api-v2.md
b/docs/zh/engines/zeta/rest-api-v2.md
index fb744e73ae..d5eda567bd 100644
--- a/docs/zh/engines/zeta/rest-api-v2.md
+++ b/docs/zh/engines/zeta/rest-api-v2.md
@@ -249,6 +249,68 @@ seatunnel:
------------------------------------------------------------------------------------------
+### 查询 Worker 资源
+
+<details>
+ <summary><code>GET</code> <code><b>/resource/workers</b></code> <code>(返回已注册
Worker 的当前资源快照。)</code></summary>
+
+#### 参数
+
+无。
+
+#### 响应
+
+```json
+{
+ "available": true,
+ "collectedAt": 1723017600000,
+ "workers": [
+ {
+ "address": "10.0.0.8:5801",
+ "tags": {"region": "us-west"},
+ "totalSlots": 4,
+ "freeSlots": 1,
+ "usedSlots": 3,
+ "dynamicSlot": false,
+ "totalCpuCores": 8,
+ "availableCpuCores": 2,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 4294967296,
+ "cpuUsage": 0.42,
+ "memUsage": 0.58,
+ "runningJobIds": [123456789]
+ },
+ {
+ "address": "10.0.0.9:5801",
+ "tags": {},
+ "totalSlots": 2,
+ "freeSlots": 0,
+ "usedSlots": 2,
+ "dynamicSlot": true,
+ "totalCpuCores": 8,
+ "availableCpuCores": 4,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 8589934592,
+ "cpuUsage": 0.35,
+ "memUsage": 0.41,
+ "runningJobIds": [123456789]
+ }
+ ]
+}
+```
+
+**说明：**
+
+- 固定 Slot 模式的 Worker 返回 `totalSlots`、`usedSlots` 和 `freeSlots`。
+- 动态 Slot 模式的 Worker 没有固定的 Slot 容量。此时，`totalSlots` 表示当前已跟踪的已分配和未分配 Slot
总数，`freeSlots` 表示当前未分配数量。解释容量时，请结合 `dynamicSlot` 以及 CPU 和堆内存字段。
+- 当无法读取 Master 端的资源快照时（包括 Master 选举期间），`available` 为 `false`。此时 `workers`
为空，客户端应重试，而不应将该响应解释为空集群。
+- `collectedAt` 是 Master 构建本次响应时的毫秒时间戳。Worker 字段来自资源管理器收到的最近一次心跳，并不与
`/system-monitoring-information` 构成原子快照。
+- 如果最近一次 Worker 心跳尚未包含资源或使用率数据，对应字段不会返回。
+
+</details>
+
+------------------------------------------------------------------------------------------
+
### 查询作业及其当前状态的概览
<details>
@@ -337,7 +399,12 @@ seatunnel:
},
"totalSlots": 4,
"freeSlots": 0,
+ "usedSlots": 4,
"dynamicSlot": false,
+ "totalCpuCores": 8,
+ "availableCpuCores": 2,
+ "totalHeapMemoryBytes": 17179869184,
+ "availableHeapMemoryBytes": 4294967296,
"cpuUsage": 0.83,
"memUsage": 0.64,
"runningJobIds": [
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java
index 0c45ada83f..8734e12ba8 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java
@@ -53,6 +53,7 @@ import
org.apache.seatunnel.engine.server.rest.servlet.SubmitJobsServlet;
import org.apache.seatunnel.engine.server.rest.servlet.SystemMonitoringServlet;
import org.apache.seatunnel.engine.server.rest.servlet.ThreadDumpServlet;
import org.apache.seatunnel.engine.server.rest.servlet.UpdateTagsServlet;
+import org.apache.seatunnel.engine.server.rest.servlet.WorkerResourceServlet;
import com.hazelcast.spi.impl.NodeEngineImpl;
import lombok.extern.slf4j.Slf4j;
@@ -94,6 +95,7 @@ import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SUBM
import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SYSTEM_MONITORING_INFORMATION;
import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_THREAD_DUMP;
import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_UPDATE_TAGS;
+import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_WORKER_RESOURCES;
/** The Jetty service for SeaTunnel engine server. */
@Slf4j
@@ -191,6 +193,8 @@ public class JettyService {
ServletHolder finishedJobsHolder = new ServletHolder(new
FinishedJobsServlet(nodeEngine));
ServletHolder systemMonitoringHolder =
new ServletHolder(new SystemMonitoringServlet(nodeEngine));
+ ServletHolder workerResourceHolder =
+ new ServletHolder(new WorkerResourceServlet(nodeEngine));
ServletHolder jobInfoHolder = new ServletHolder(new
JobInfoServlet(nodeEngine));
ServletHolder threadDumpHolder = new ServletHolder(new
ThreadDumpServlet(nodeEngine));
@@ -230,6 +234,7 @@ public class JettyService {
context.addServlet(finishedJobsHolder,
convertUrlToPath(REST_URL_FINISHED_JOBS));
context.addServlet(
systemMonitoringHolder,
convertUrlToPath(REST_URL_SYSTEM_MONITORING_INFORMATION));
+ context.addServlet(workerResourceHolder,
convertUrlToPath(REST_URL_WORKER_RESOURCES));
context.addServlet(jobInfoHolder, convertUrlToPath(REST_URL_JOB_INFO));
context.addServlet(jobInfoHolder,
convertUrlToPath(REST_URL_RUNNING_JOB));
context.addServlet(threadDumpHolder,
convertUrlToPath(REST_URL_THREAD_DUMP));
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollector.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollector.java
index 4044307477..3b36088c48 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollector.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollector.java
@@ -26,6 +26,7 @@ import
org.apache.seatunnel.engine.server.execution.PendingJobInfo;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
import org.apache.seatunnel.engine.server.master.JobMaster;
import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager;
+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.resource.SystemLoadInfo;
import org.apache.seatunnel.engine.server.resourcemanager.worker.WorkerProfile;
@@ -34,7 +35,9 @@ import com.hazelcast.cluster.Address;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -264,35 +267,47 @@ public final class PendingDiagnosticsCollector {
} catch (Exception e) {
log.warn("Collect worker count failed: {}",
ExceptionUtils.getMessage(e));
}
- snapshot.setWorkers(buildWorkerSnapshots(resourceManager, tags));
+
snapshot.setWorkers(collectWorkerResourceSnapshot(resourceManager).getWorkers());
return snapshot;
}
- private static List<WorkerResourceDiagnostic> buildWorkerSnapshots(
- ResourceManager resourceManager, Map<String, String> tagFilter) {
+ /** Collects a point-in-time projection of the resources registered by the
master. */
+ public static WorkerResourceSnapshot collectWorkerResourceSnapshot(
+ ResourceManager resourceManager) {
+ WorkerResourceSnapshot snapshot = new WorkerResourceSnapshot();
+ snapshot.setCollectedAt(System.currentTimeMillis());
if (resourceManager == null) {
- return Collections.emptyList();
+ return snapshot;
}
- Map<Address, WorkerProfile> registerWorker =
- Optional.ofNullable(resourceManager.getRegisterWorker())
- .map(HashMap::new)
- .orElseGet(HashMap::new);
- return registerWorker.values().stream()
- .map(worker -> convertWorker(worker, tagFilter))
- .collect(Collectors.toList());
+ try {
+ Map<Address, WorkerProfile> registerWorker =
resourceManager.getRegisterWorker();
+ if (registerWorker == null) {
+ return snapshot;
+ }
+ List<WorkerResourceDiagnostic> workers =
+ registerWorker.entrySet().stream()
+ .map(entry -> convertWorker(entry.getKey(),
entry.getValue()))
+ .filter(worker -> worker != null)
+
.sorted(Comparator.comparing(WorkerResourceDiagnostic::getAddress))
+ .collect(Collectors.toList());
+ snapshot.setAvailable(true);
+ snapshot.setWorkers(workers);
+ } catch (Exception e) {
+ log.warn("Collect worker resource snapshot failed: {}",
ExceptionUtils.getMessage(e));
+ }
+ return snapshot;
}
- /**
- * TODO The current tagFilter does not actually filter. When the cluster
is particularly large,
- * tagFilter filtering should be supported, and it will be supported in
the future
- */
private static WorkerResourceDiagnostic convertWorker(
- WorkerProfile workerProfile, Map<String, String> tagFilter) {
- WorkerResourceDiagnostic diagnostic = new WorkerResourceDiagnostic();
+ Address registeredAddress, WorkerProfile workerProfile) {
if (workerProfile == null) {
- return diagnostic;
+ return null;
}
+ WorkerResourceDiagnostic diagnostic = new WorkerResourceDiagnostic();
Address address = workerProfile.getAddress();
+ if (address == null) {
+ address = registeredAddress;
+ }
diagnostic.setAddress(address == null ? "UNKNOWN" :
address.toString());
if (workerProfile.getAttributes() != null) {
diagnostic.setTags(new HashMap<>(workerProfile.getAttributes()));
@@ -308,22 +323,44 @@ public final class PendingDiagnosticsCollector {
workerProfile.getUnassignedSlots() == null
? 0
: workerProfile.getUnassignedSlots().length;
+ diagnostic.setUsedSlots(assignedSlots);
diagnostic.setTotalSlots(assignedSlots + unassignedSlots);
diagnostic.setFreeSlots(unassignedSlots);
+ setResourceValues(diagnostic, workerProfile.getProfile(), true);
+ setResourceValues(diagnostic, workerProfile.getUnassignedResource(),
false);
SystemLoadInfo systemLoadInfo = workerProfile.getSystemLoadInfo();
if (systemLoadInfo != null) {
diagnostic.setCpuUsage(systemLoadInfo.getCpuPercentage());
diagnostic.setMemUsage(systemLoadInfo.getMemPercentage());
}
- if (workerProfile.getAssignedSlots() != null) {
- List<Long> runningJobs =
- java.util.Arrays.stream(workerProfile.getAssignedSlots())
- .filter(slot -> slot != null &&
slot.getOwnerJobID() > 0)
- .map(SlotProfile::getOwnerJobID)
- .distinct()
- .collect(Collectors.toList());
- diagnostic.setRunningJobIds(runningJobs);
- }
+ List<Long> runningJobs =
+ workerProfile.getAssignedSlots() == null
+ ? Collections.emptyList()
+ : Arrays.stream(workerProfile.getAssignedSlots())
+ .filter(slot -> slot != null &&
slot.getOwnerJobID() > 0)
+ .map(SlotProfile::getOwnerJobID)
+ .distinct()
+ .sorted()
+ .collect(Collectors.toList());
+ diagnostic.setRunningJobIds(runningJobs);
return diagnostic;
}
+
+ private static void setResourceValues(
+ WorkerResourceDiagnostic diagnostic, ResourceProfile resource,
boolean total) {
+ if (resource == null) {
+ return;
+ }
+ if (total) {
+ diagnostic.setTotalCpuCores(
+ resource.getCpu() == null ? null :
resource.getCpu().getCore());
+ diagnostic.setTotalHeapMemoryBytes(
+ resource.getHeapMemory() == null ? null :
resource.getHeapMemory().getBytes());
+ } else {
+ diagnostic.setAvailableCpuCores(
+ resource.getCpu() == null ? null :
resource.getCpu().getCore());
+ diagnostic.setAvailableHeapMemoryBytes(
+ resource.getHeapMemory() == null ? null :
resource.getHeapMemory().getBytes());
+ }
+ }
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
index 17905516bb..130a6aa1e4 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
@@ -25,15 +25,30 @@ import java.io.Serializable;
import java.util.List;
import java.util.Map;
+/**
+ * Read-only projection of the latest resource-manager heartbeat for one
worker.
+ *
+ * <p>For dynamic-slot workers, {@code totalSlots} is the number of currently
tracked assigned and
+ * unassigned slots rather than a fixed capacity. Callers should use {@code
dynamicSlot} to
+ * interpret the slot fields.
+ */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class WorkerResourceDiagnostic implements Serializable {
+ // Preserve compatibility with diagnostics serialized before resource
fields were added.
+ private static final long serialVersionUID = 1347570277332453777L;
+
private String address;
private Map<String, String> tags;
private int totalSlots;
private int freeSlots;
+ private int usedSlots;
private boolean dynamicSlot;
+ private Integer totalCpuCores;
+ private Integer availableCpuCores;
+ private Long totalHeapMemoryBytes;
+ private Long availableHeapMemoryBytes;
private Double cpuUsage;
private Double memUsage;
private List<Long> runningJobIds;
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceSnapshot.java
similarity index 74%
copy from
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
copy to
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceSnapshot.java
index 17905516bb..64ef086b21 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceSnapshot.java
@@ -22,19 +22,17 @@ import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
+/** Point-in-time master projection of the latest registered worker
heartbeats. */
@Data
@NoArgsConstructor
@AllArgsConstructor
-public class WorkerResourceDiagnostic implements Serializable {
- private String address;
- private Map<String, String> tags;
- private int totalSlots;
- private int freeSlots;
- private boolean dynamicSlot;
- private Double cpuUsage;
- private Double memUsage;
- private List<Long> runningJobIds;
+public class WorkerResourceSnapshot implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private boolean available;
+ private long collectedAt;
+ private List<WorkerResourceDiagnostic> workers = new ArrayList<>();
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/opeartion/GetWorkerResourcesOperation.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/opeartion/GetWorkerResourcesOperation.java
new file mode 100644
index 0000000000..7930cbab91
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/resourcemanager/opeartion/GetWorkerResourcesOperation.java
@@ -0,0 +1,63 @@
+/*
+ * 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.resourcemanager.opeartion;
+
+import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import
org.apache.seatunnel.engine.server.diagnostic.PendingDiagnosticsCollector;
+import org.apache.seatunnel.engine.server.diagnostic.WorkerResourceSnapshot;
+import
org.apache.seatunnel.engine.server.serializable.ResourceDataSerializerHook;
+
+import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
+import com.hazelcast.spi.impl.operationservice.Operation;
+
+/** Collects the current worker resource snapshot on the master node. */
+public class GetWorkerResourcesOperation extends Operation implements
IdentifiedDataSerializable {
+
+ private WorkerResourceSnapshot snapshot;
+
+ @Override
+ public void run() throws Exception {
+ SeaTunnelServer server = getService();
+ snapshot = getWorkerResourceSnapshot(server);
+ }
+
+ @Override
+ public Object getResponse() {
+ return snapshot;
+ }
+
+ @Override
+ public int getFactoryId() {
+ return ResourceDataSerializerHook.FACTORY_ID;
+ }
+
+ @Override
+ public int getClassId() {
+ return ResourceDataSerializerHook.GET_WORKER_RESOURCES_TYPE;
+ }
+
+ @Override
+ public String getServiceName() {
+ return SeaTunnelServer.SERVICE_NAME;
+ }
+
+ public static WorkerResourceSnapshot
getWorkerResourceSnapshot(SeaTunnelServer server) {
+ return PendingDiagnosticsCollector.collectWorkerResourceSnapshot(
+ server.getCoordinatorService().getResourceManager());
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java
index 883a16c5e3..ab2c05feb1 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java
@@ -96,6 +96,7 @@ public class RestConstant {
public static final String REST_URL_RUNNING_THREADS = "/running-threads";
public static final String REST_URL_SYSTEM_MONITORING_INFORMATION =
"/system-monitoring-information";
+ public static final String REST_URL_WORKER_RESOURCES = "/resource/workers";
public static final String REST_URL_SUBMIT_JOB = "/submit-job";
public static final String REST_URL_SUBMIT_JOB_BY_UPLOAD_FILE =
"/submit-job/upload";
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java
index 4d470cf430..40dce8bdd2 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java
@@ -33,6 +33,7 @@ import
org.apache.seatunnel.engine.server.rest.service.RunningThreadService;
import org.apache.seatunnel.engine.server.rest.service.SystemMonitoringService;
import org.apache.seatunnel.engine.server.rest.service.ThreadDumpService;
import org.apache.seatunnel.engine.server.rest.service.TraceTaskMappingService;
+import org.apache.seatunnel.engine.server.rest.service.WorkerResourceService;
import com.google.gson.Gson;
import com.hazelcast.internal.ascii.TextCommandService;
@@ -75,6 +76,7 @@ import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_RUNN
import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SYSTEM_MONITORING_INFORMATION;
import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_THREAD_DUMP;
import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_TRACE_TASK_MAPPING;
+import static
org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_WORKER_RESOURCES;
/**
* Dispatches Hazelcast ASCII GET requests to SeaTunnel-specific overview,
log, and trace services.
@@ -92,6 +94,7 @@ public class RestHttpGetCommandProcessor extends
HttpCommandProcessor<HttpGetCom
private LogService logService;
private TraceTaskMappingService traceTaskMappingService;
private OptionRulesService optionRulesService;
+ private WorkerResourceService workerResourceService;
public RestHttpGetCommandProcessor(TextCommandService textCommandService) {
@@ -105,6 +108,7 @@ public class RestHttpGetCommandProcessor extends
HttpCommandProcessor<HttpGetCom
this.logService = new LogService(nodeEngine);
this.traceTaskMappingService = new TraceTaskMappingService(nodeEngine);
this.optionRulesService = new OptionRulesService(nodeEngine);
+ this.workerResourceService = new WorkerResourceService(nodeEngine);
}
public RestHttpGetCommandProcessor(
@@ -123,6 +127,7 @@ public class RestHttpGetCommandProcessor extends
HttpCommandProcessor<HttpGetCom
this.logService = new LogService(nodeEngine);
this.traceTaskMappingService = new TraceTaskMappingService(nodeEngine);
this.optionRulesService = new OptionRulesService(nodeEngine);
+ this.workerResourceService = new WorkerResourceService(nodeEngine);
}
/**
@@ -144,6 +149,8 @@ public class RestHttpGetCommandProcessor extends
HttpCommandProcessor<HttpGetCom
handleJobInfoById(httpGetCommand, uri);
} else if (uri.startsWith(CONTEXT_PATH +
REST_URL_SYSTEM_MONITORING_INFORMATION)) {
getSystemMonitoringInformation(httpGetCommand);
+ } else if (uri.startsWith(CONTEXT_PATH +
REST_URL_WORKER_RESOURCES)) {
+ getWorkerResources(httpGetCommand);
} else if (uri.startsWith(CONTEXT_PATH +
REST_URL_RUNNING_THREADS)) {
getRunningThread(httpGetCommand);
} else if (uri.startsWith(CONTEXT_PATH + REST_URL_OVERVIEW)) {
@@ -230,6 +237,11 @@ public class RestHttpGetCommandProcessor extends
HttpCommandProcessor<HttpGetCom
command,
systemMonitoringService.getSystemMonitoringInformationJsonValues());
}
+ private void getWorkerResources(HttpGetCommand command) {
+ String response = new
Gson().toJson(workerResourceService.getWorkerResources());
+ this.prepareResponse(command, Json.parse(response).asObject());
+ }
+
private void handleRunningJobsInfo(HttpGetCommand command) {
this.prepareResponse(command, jobInfoService.getRunningJobsJson());
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/WorkerResourceService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/WorkerResourceService.java
new file mode 100644
index 0000000000..7f4ce8524d
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/WorkerResourceService.java
@@ -0,0 +1,89 @@
+/*
+ * 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.rest.service;
+
+import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
+import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import org.apache.seatunnel.engine.server.diagnostic.WorkerResourceSnapshot;
+import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.GetWorkerResourcesOperation;
+import org.apache.seatunnel.engine.server.utils.NodeEngineUtil;
+
+import com.hazelcast.cluster.Address;
+import com.hazelcast.core.HazelcastInstanceNotActiveException;
+import com.hazelcast.core.MemberLeftException;
+import com.hazelcast.spi.exception.TargetDisconnectedException;
+import com.hazelcast.spi.exception.TargetNotMemberException;
+import com.hazelcast.spi.impl.NodeEngineImpl;
+
+import java.util.Collections;
+
+/** Resolves the worker resource snapshot locally on the master or forwards
the read to it. */
+public class WorkerResourceService extends BaseService {
+
+ public WorkerResourceService(NodeEngineImpl nodeEngine) {
+ super(nodeEngine);
+ }
+
+ /** Returns the local master's snapshot or forwards the request to the
current master. */
+ public WorkerResourceSnapshot getWorkerResources() {
+ SeaTunnelServer seaTunnelServer = getSeaTunnelServer(true);
+ if (seaTunnelServer != null) {
+ return
GetWorkerResourcesOperation.getWorkerResourceSnapshot(seaTunnelServer);
+ }
+ Address masterAddress = nodeEngine.getMasterAddress();
+ if (masterAddress == null) {
+ return unavailableSnapshot();
+ }
+ try {
+ return invokeOnMaster(masterAddress);
+ } catch (RuntimeException e) {
+ if (isTransientMasterFailure(e)) {
+ return unavailableSnapshot();
+ }
+ throw e;
+ }
+ }
+
+ protected WorkerResourceSnapshot invokeOnMaster(Address masterAddress) {
+ return (WorkerResourceSnapshot)
+ NodeEngineUtil.sendOperationToMemberNode(
+ nodeEngine, new GetWorkerResourcesOperation(),
masterAddress)
+ .join();
+ }
+
+ private WorkerResourceSnapshot unavailableSnapshot() {
+ return new WorkerResourceSnapshot(
+ false, System.currentTimeMillis(), Collections.emptyList());
+ }
+
+ /** Identifies failures caused by a master transition while the request is
in flight. */
+ private boolean isTransientMasterFailure(Throwable error) {
+ Throwable current = error;
+ while (current != null) {
+ if (current instanceof TargetNotMemberException
+ || current instanceof TargetDisconnectedException
+ || current instanceof MemberLeftException
+ || current instanceof HazelcastInstanceNotActiveException
+ || current instanceof SeaTunnelEngineException) {
+ return true;
+ }
+ current = current.getCause();
+ }
+ return false;
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/WorkerResourceServlet.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/WorkerResourceServlet.java
new file mode 100644
index 0000000000..abc1571ff2
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/WorkerResourceServlet.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.seatunnel.engine.server.rest.servlet;
+
+import org.apache.seatunnel.engine.server.rest.service.WorkerResourceService;
+
+import com.hazelcast.spi.impl.NodeEngineImpl;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.IOException;
+
+/** Serves the read-only worker resource snapshot. */
+public class WorkerResourceServlet extends BaseServlet {
+
+ private final WorkerResourceService workerResourceService;
+
+ public WorkerResourceServlet(NodeEngineImpl nodeEngine) {
+ this(nodeEngine, new WorkerResourceService(nodeEngine));
+ }
+
+ WorkerResourceServlet(NodeEngineImpl nodeEngine, WorkerResourceService
workerResourceService) {
+ super(nodeEngine);
+ this.workerResourceService = workerResourceService;
+ }
+
+ @Override
+ protected void doGet(HttpServletRequest req, HttpServletResponse resp)
+ throws ServletException, IOException {
+ writeJson(resp, workerResourceService.getWorkerResources());
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/ResourceDataSerializerHook.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/ResourceDataSerializerHook.java
index ca205dbbfd..ef907f1b07 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/ResourceDataSerializerHook.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/serializable/ResourceDataSerializerHook.java
@@ -22,6 +22,7 @@ import
org.apache.seatunnel.engine.server.master.cleanup.JobCleanupRecord;
import org.apache.seatunnel.engine.server.master.cleanup.PipelineCleanupRecord;
import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.GetOverviewOperation;
import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.GetPendingJobsOperation;
+import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.GetWorkerResourcesOperation;
import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.ReleaseSlotOperation;
import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.RequestSlotOperation;
import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.ResetResourceOperation;
@@ -62,6 +63,8 @@ public class ResourceDataSerializerHook implements
DataSerializerHook {
public static final int JOB_CLEANUP_RECORD_TYPE = 12;
+ public static final int GET_WORKER_RESOURCES_TYPE = 13;
+
public static final int FACTORY_ID =
FactoryIdHelper.getFactoryId(
SeaTunnelFactoryIdConstant.SEATUNNEL_RESOURCE_DATA_SERIALIZER_FACTORY,
@@ -106,6 +109,8 @@ public class ResourceDataSerializerHook implements
DataSerializerHook {
return new PipelineCleanupRecord();
case JOB_CLEANUP_RECORD_TYPE:
return new JobCleanupRecord();
+ case GET_WORKER_RESOURCES_TYPE:
+ return new GetWorkerResourcesOperation();
default:
throw new IllegalArgumentException("Unknown type id " +
typeId);
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollectorTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollectorTest.java
index 1b9fd5e58a..87f663ec11 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollectorTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/diagnostic/PendingDiagnosticsCollectorTest.java
@@ -28,19 +28,198 @@ import
org.apache.seatunnel.engine.server.execution.PendingSourceState;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
import org.apache.seatunnel.engine.server.master.JobMaster;
import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager;
+import org.apache.seatunnel.engine.server.resourcemanager.resource.CPU;
+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.resource.SystemLoadInfo;
+import org.apache.seatunnel.engine.server.resourcemanager.worker.WorkerProfile;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import com.hazelcast.cluster.Address;
+
+import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
public class PendingDiagnosticsCollectorTest {
+ @Test
+ public void testCollectWorkerResourceSnapshot() throws
UnknownHostException {
+ ResourceManager resourceManager = Mockito.mock(ResourceManager.class);
+ Address dynamicAddress = new Address("localhost", 5801);
+ Address fixedAddress = new Address("localhost", 5802);
+
+ SlotProfile dynamicAssigned = slot(dynamicAddress, 0, 100L);
+ WorkerProfile dynamicWorker =
+ worker(
+ dynamicAddress,
+ true,
+ new ResourceProfile(CPU.of(8), Memory.of(8192)),
+ new ResourceProfile(CPU.of(3), Memory.of(3072)),
+ new SlotProfile[] {dynamicAssigned},
+ new SlotProfile[] {slot(dynamicAddress, 1, 0L)});
+ dynamicWorker.setSystemLoadInfo(new SystemLoadInfo(60.0, 25.0));
+
+ SlotProfile fixedAssignedA = slot(fixedAddress, 0, 200L);
+ SlotProfile fixedAssignedB = slot(fixedAddress, 1, 200L);
+ SlotProfile fixedAssignedC = slot(fixedAddress, 2, 300L);
+ WorkerProfile fixedWorker =
+ worker(
+ fixedAddress,
+ false,
+ new ResourceProfile(CPU.of(16), Memory.of(16384)),
+ new ResourceProfile(CPU.of(4), Memory.of(4096)),
+ new SlotProfile[] {fixedAssignedA, fixedAssignedB,
fixedAssignedC},
+ new SlotProfile[] {slot(fixedAddress, 3, 0L),
slot(fixedAddress, 4, 0L)});
+
+ ConcurrentMap<Address, WorkerProfile> workers = new
ConcurrentHashMap<>();
+ workers.put(fixedAddress, fixedWorker);
+ workers.put(dynamicAddress, dynamicWorker);
+ Mockito.when(resourceManager.getRegisterWorker()).thenReturn(workers);
+
+ WorkerResourceSnapshot snapshot =
+
PendingDiagnosticsCollector.collectWorkerResourceSnapshot(resourceManager);
+
+ Assertions.assertTrue(snapshot.isAvailable());
+ Assertions.assertTrue(snapshot.getCollectedAt() > 0);
+ Assertions.assertEquals(2, snapshot.getWorkers().size());
+
+ WorkerResourceDiagnostic dynamic = snapshot.getWorkers().get(0);
+ Assertions.assertEquals(dynamicAddress.toString(),
dynamic.getAddress());
+ Assertions.assertTrue(dynamic.isDynamicSlot());
+ Assertions.assertEquals(1, dynamic.getUsedSlots());
+ Assertions.assertEquals(2, dynamic.getTotalSlots());
+ Assertions.assertEquals(1, dynamic.getFreeSlots());
+ Assertions.assertEquals(8, dynamic.getTotalCpuCores());
+ Assertions.assertEquals(3, dynamic.getAvailableCpuCores());
+ Assertions.assertEquals(8192L, dynamic.getTotalHeapMemoryBytes());
+ Assertions.assertEquals(3072L, dynamic.getAvailableHeapMemoryBytes());
+ Assertions.assertEquals(25.0, dynamic.getCpuUsage());
+ Assertions.assertEquals(60.0, dynamic.getMemUsage());
+ Assertions.assertEquals(Collections.singletonList(100L),
dynamic.getRunningJobIds());
+
+ WorkerResourceDiagnostic fixed = snapshot.getWorkers().get(1);
+ Assertions.assertEquals(fixedAddress.toString(), fixed.getAddress());
+ Assertions.assertFalse(fixed.isDynamicSlot());
+ Assertions.assertEquals(3, fixed.getUsedSlots());
+ Assertions.assertEquals(5, fixed.getTotalSlots());
+ Assertions.assertEquals(2, fixed.getFreeSlots());
+ Assertions.assertEquals(Arrays.asList(200L, 300L),
fixed.getRunningJobIds());
+ }
+
+ @Test
+ public void testCollectWorkerResourceSnapshotWithIncompleteProfile()
+ throws UnknownHostException {
+ ResourceManager resourceManager = Mockito.mock(ResourceManager.class);
+ Address registeredAddress = new Address("localhost", 5801);
+ WorkerProfile worker = new WorkerProfile();
+ worker.setAddress(null);
+ worker.setProfile(null);
+ worker.setUnassignedResource(null);
+ worker.setAssignedSlots(null);
+ worker.setUnassignedSlots(null);
+ worker.setAttributes(null);
+
+ ConcurrentMap<Address, WorkerProfile> workers = new
ConcurrentHashMap<>();
+ workers.put(registeredAddress, worker);
+ Mockito.when(resourceManager.getRegisterWorker()).thenReturn(workers);
+
+ WorkerResourceDiagnostic diagnostic =
+
PendingDiagnosticsCollector.collectWorkerResourceSnapshot(resourceManager)
+ .getWorkers()
+ .get(0);
+
+ Assertions.assertEquals(registeredAddress.toString(),
diagnostic.getAddress());
+ Assertions.assertEquals(0, diagnostic.getUsedSlots());
+ Assertions.assertEquals(0, diagnostic.getTotalSlots());
+ Assertions.assertEquals(0, diagnostic.getFreeSlots());
+ Assertions.assertNull(diagnostic.getTotalCpuCores());
+ Assertions.assertNull(diagnostic.getAvailableCpuCores());
+ Assertions.assertNull(diagnostic.getTotalHeapMemoryBytes());
+ Assertions.assertNull(diagnostic.getAvailableHeapMemoryBytes());
+ Assertions.assertEquals(Collections.emptyMap(), diagnostic.getTags());
+ Assertions.assertEquals(Collections.emptyList(),
diagnostic.getRunningJobIds());
+ }
+
+ @Test
+ public void testCollectWorkerResourceSnapshotAfterSlotReleaseAndReuse()
+ throws UnknownHostException {
+ ResourceManager resourceManager = Mockito.mock(ResourceManager.class);
+ Address address = new Address("localhost", 5801);
+ SlotProfile retained = slot(address, 0, 100L);
+ SlotProfile released = slot(address, 1, 200L);
+ SlotProfile free = slot(address, 2, 0L);
+ WorkerProfile worker =
+ worker(
+ address,
+ false,
+ new ResourceProfile(CPU.of(8), Memory.of(8192)),
+ new ResourceProfile(CPU.of(4), Memory.of(4096)),
+ new SlotProfile[] {retained, released},
+ new SlotProfile[] {free});
+ ConcurrentMap<Address, WorkerProfile> workers = new
ConcurrentHashMap<>();
+ workers.put(address, worker);
+ Mockito.when(resourceManager.getRegisterWorker()).thenReturn(workers);
+
+ released.unassigned();
+ worker.setAssignedSlots(new SlotProfile[] {retained});
+ worker.setUnassignedSlots(new SlotProfile[] {released, free});
+
+ WorkerResourceDiagnostic afterRelease =
+
PendingDiagnosticsCollector.collectWorkerResourceSnapshot(resourceManager)
+ .getWorkers()
+ .get(0);
+ Assertions.assertEquals(1, afterRelease.getUsedSlots());
+ Assertions.assertEquals(3, afterRelease.getTotalSlots());
+ Assertions.assertEquals(2, afterRelease.getFreeSlots());
+ Assertions.assertEquals(Collections.singletonList(100L),
afterRelease.getRunningJobIds());
+
+ released.assign(300L);
+ worker.setAssignedSlots(new SlotProfile[] {retained, released});
+ worker.setUnassignedSlots(new SlotProfile[] {free});
+
+ WorkerResourceDiagnostic afterReuse =
+
PendingDiagnosticsCollector.collectWorkerResourceSnapshot(resourceManager)
+ .getWorkers()
+ .get(0);
+ Assertions.assertEquals(2, afterReuse.getUsedSlots());
+ Assertions.assertEquals(3, afterReuse.getTotalSlots());
+ Assertions.assertEquals(1, afterReuse.getFreeSlots());
+ Assertions.assertEquals(Arrays.asList(100L, 300L),
afterReuse.getRunningJobIds());
+ }
+
+ @Test
+ public void testCollectEmptyAndUnavailableWorkerResourceSnapshots() {
+ WorkerResourceSnapshot unavailable =
+
PendingDiagnosticsCollector.collectWorkerResourceSnapshot(null);
+ Assertions.assertFalse(unavailable.isAvailable());
+ Assertions.assertTrue(unavailable.getWorkers().isEmpty());
+
+ ResourceManager emptyResourceManager =
Mockito.mock(ResourceManager.class);
+ Mockito.when(emptyResourceManager.getRegisterWorker())
+ .thenReturn(new ConcurrentHashMap<>());
+ WorkerResourceSnapshot empty =
+
PendingDiagnosticsCollector.collectWorkerResourceSnapshot(emptyResourceManager);
+ Assertions.assertTrue(empty.isAvailable());
+ Assertions.assertTrue(empty.getWorkers().isEmpty());
+
+ ResourceManager unavailableResourceManager =
Mockito.mock(ResourceManager.class);
+
Mockito.when(unavailableResourceManager.getRegisterWorker()).thenReturn(null);
+ WorkerResourceSnapshot missing =
+ PendingDiagnosticsCollector.collectWorkerResourceSnapshot(
+ unavailableResourceManager);
+ Assertions.assertFalse(missing.isAvailable());
+ Assertions.assertTrue(missing.getWorkers().isEmpty());
+ }
+
@Test
public void testCollectJobDiagnosticWithFailures() {
JobMaster jobMaster = Mockito.mock(JobMaster.class);
@@ -111,4 +290,30 @@ public class PendingDiagnosticsCollectorTest {
Assertions.assertEquals(3,
diagnostic.getPipelines().get(0).getTotalTaskGroups());
Assertions.assertEquals(2,
diagnostic.getPipelines().get(0).getLackingTaskGroups());
}
+
+ private static WorkerProfile worker(
+ Address address,
+ boolean dynamicSlot,
+ ResourceProfile profile,
+ ResourceProfile unassignedResource,
+ SlotProfile[] assignedSlots,
+ SlotProfile[] unassignedSlots) {
+ return new WorkerProfile(
+ address,
+ profile,
+ unassignedResource,
+ dynamicSlot,
+ assignedSlots,
+ unassignedSlots,
+ Collections.singletonMap("region", "test"));
+ }
+
+ private static SlotProfile slot(Address address, int slotId, long
ownerJobId) {
+ SlotProfile slot =
+ new SlotProfile(address, slotId, new ResourceProfile(),
"slot-" + slotId);
+ if (ownerJobId > 0) {
+ slot.assign(ownerJobId);
+ }
+ return slot;
+ }
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/opeartion/GetWorkerResourcesOperationTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/opeartion/GetWorkerResourcesOperationTest.java
new file mode 100644
index 0000000000..8a6f34e09b
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/resourcemanager/opeartion/GetWorkerResourcesOperationTest.java
@@ -0,0 +1,61 @@
+/*
+ * 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.resourcemanager.opeartion;
+
+import org.apache.seatunnel.engine.server.CoordinatorService;
+import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import org.apache.seatunnel.engine.server.diagnostic.WorkerResourceSnapshot;
+import org.apache.seatunnel.engine.server.resourcemanager.ResourceManager;
+import
org.apache.seatunnel.engine.server.serializable.ResourceDataSerializerHook;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class GetWorkerResourcesOperationTest {
+
+ @Test
+ void shouldCollectWorkerResourcesFromTheMasterResourceManager() {
+ SeaTunnelServer server = mock(SeaTunnelServer.class);
+ CoordinatorService coordinatorService = mock(CoordinatorService.class);
+ ResourceManager resourceManager = mock(ResourceManager.class);
+ when(server.getCoordinatorService()).thenReturn(coordinatorService);
+
when(coordinatorService.getResourceManager()).thenReturn(resourceManager);
+ when(resourceManager.getRegisterWorker()).thenReturn(new
ConcurrentHashMap<>());
+
+ WorkerResourceSnapshot snapshot =
+ GetWorkerResourcesOperation.getWorkerResourceSnapshot(server);
+
+ assertTrue(snapshot.isAvailable());
+ assertTrue(snapshot.getWorkers().isEmpty());
+ }
+
+ @Test
+ void shouldUseTheResourceSerializerContract() {
+ GetWorkerResourcesOperation operation = new
GetWorkerResourcesOperation();
+
+ assertEquals(ResourceDataSerializerHook.FACTORY_ID,
operation.getFactoryId());
+ assertEquals(ResourceDataSerializerHook.GET_WORKER_RESOURCES_TYPE,
operation.getClassId());
+ assertEquals(SeaTunnelServer.SERVICE_NAME, operation.getServiceName());
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/WorkerResourceServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/WorkerResourceServiceTest.java
new file mode 100644
index 0000000000..96033d242c
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/service/WorkerResourceServiceTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.rest.service;
+
+import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
+import org.apache.seatunnel.engine.server.SeaTunnelServer;
+import org.apache.seatunnel.engine.server.diagnostic.WorkerResourceSnapshot;
+
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.cluster.Address;
+import com.hazelcast.spi.exception.TargetNotMemberException;
+import com.hazelcast.spi.impl.NodeEngineImpl;
+
+import java.net.UnknownHostException;
+import java.util.concurrent.CompletionException;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class WorkerResourceServiceTest {
+
+ @Test
+ void shouldReturnUnavailableSnapshotBeforeMasterElectionCompletes() {
+ NodeEngineImpl nodeEngine = mock(NodeEngineImpl.class);
+ when(nodeEngine.getMasterAddress()).thenReturn(null);
+ WorkerResourceService service = new
TestWorkerResourceService(nodeEngine, null, null);
+
+ WorkerResourceSnapshot snapshot = service.getWorkerResources();
+
+ assertFalse(snapshot.isAvailable());
+ assertTrue(snapshot.getCollectedAt() > 0);
+ assertNotNull(snapshot.getWorkers());
+ assertTrue(snapshot.getWorkers().isEmpty());
+ }
+
+ @Test
+ void shouldReturnUnavailableSnapshotWhenMasterLeavesDuringForwarding()
+ throws UnknownHostException {
+ NodeEngineImpl nodeEngine = mock(NodeEngineImpl.class);
+ when(nodeEngine.getMasterAddress()).thenReturn(new
Address("localhost", 5801));
+ RuntimeException failure =
+ new CompletionException(new TargetNotMemberException("master
left"));
+ WorkerResourceService service = new
TestWorkerResourceService(nodeEngine, null, failure);
+
+ WorkerResourceSnapshot snapshot = service.getWorkerResources();
+
+ assertFalse(snapshot.isAvailable());
+ assertTrue(snapshot.getWorkers().isEmpty());
+ }
+
+ @Test
+ void shouldPropagateUnrelatedForwardingFailure() throws
UnknownHostException {
+ NodeEngineImpl nodeEngine = mock(NodeEngineImpl.class);
+ when(nodeEngine.getMasterAddress()).thenReturn(new
Address("localhost", 5801));
+ RuntimeException failure = new IllegalStateException("unexpected
failure");
+ WorkerResourceService service = new
TestWorkerResourceService(nodeEngine, null, failure);
+
+ assertThrows(IllegalStateException.class, service::getWorkerResources);
+ }
+
+ @Test
+ void shouldReturnUnavailableSnapshotWhenTargetLosesMastership() throws
UnknownHostException {
+ NodeEngineImpl nodeEngine = mock(NodeEngineImpl.class);
+ when(nodeEngine.getMasterAddress()).thenReturn(new
Address("localhost", 5801));
+ RuntimeException failure =
+ new CompletionException(
+ new SeaTunnelEngineException("This is not a master
node now."));
+ WorkerResourceService service = new
TestWorkerResourceService(nodeEngine, null, failure);
+
+ WorkerResourceSnapshot snapshot = service.getWorkerResources();
+
+ assertFalse(snapshot.isAvailable());
+ assertTrue(snapshot.getWorkers().isEmpty());
+ }
+
+ private static class TestWorkerResourceService extends
WorkerResourceService {
+ private final SeaTunnelServer seaTunnelServer;
+ private final RuntimeException invocationFailure;
+
+ private TestWorkerResourceService(
+ NodeEngineImpl nodeEngine,
+ SeaTunnelServer seaTunnelServer,
+ RuntimeException invocationFailure) {
+ super(nodeEngine);
+ this.seaTunnelServer = seaTunnelServer;
+ this.invocationFailure = invocationFailure;
+ }
+
+ @Override
+ protected SeaTunnelServer getSeaTunnelServer(boolean shouldBeMaster) {
+ return seaTunnelServer;
+ }
+
+ @Override
+ protected WorkerResourceSnapshot invokeOnMaster(Address masterAddress)
{
+ throw invocationFailure;
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/servlet/WorkerResourceServletTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/servlet/WorkerResourceServletTest.java
new file mode 100644
index 0000000000..2952ede99d
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/rest/servlet/WorkerResourceServletTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.rest.servlet;
+
+import org.apache.seatunnel.engine.server.diagnostic.WorkerResourceSnapshot;
+import org.apache.seatunnel.engine.server.rest.service.WorkerResourceService;
+
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.spi.impl.NodeEngineImpl;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class WorkerResourceServletTest {
+
+ @Test
+ void shouldWriteWorkerResourceSnapshotAsJson() throws Exception {
+ NodeEngineImpl nodeEngine = mock(NodeEngineImpl.class);
+ WorkerResourceService service = mock(WorkerResourceService.class);
+ WorkerResourceSnapshot snapshot =
+ new WorkerResourceSnapshot(true, 1234L,
Collections.emptyList());
+ when(service.getWorkerResources()).thenReturn(snapshot);
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ HttpServletResponse response = mock(HttpServletResponse.class);
+ StringWriter output = new StringWriter();
+ when(response.getWriter()).thenReturn(new PrintWriter(output));
+
+ new WorkerResourceServlet(nodeEngine, service).doGet(request,
response);
+
+
assertEquals("{\"available\":true,\"collectedAt\":1234,\"workers\":[]}",
output.toString());
+ verify(response).setCharacterEncoding("UTF-8");
+ verify(response).setContentType("application/json; charset=UTF-8");
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/serializable/ResourceDataSerializerHookTest.java
similarity index 52%
copy from
seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
copy to
seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/serializable/ResourceDataSerializerHookTest.java
index 17905516bb..dc05e87eb6 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/diagnostic/WorkerResourceDiagnostic.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/serializable/ResourceDataSerializerHookTest.java
@@ -15,26 +15,25 @@
* limitations under the License.
*/
-package org.apache.seatunnel.engine.server.diagnostic;
-
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.Map;
-
-@Data
-@NoArgsConstructor
-@AllArgsConstructor
-public class WorkerResourceDiagnostic implements Serializable {
- private String address;
- private Map<String, String> tags;
- private int totalSlots;
- private int freeSlots;
- private boolean dynamicSlot;
- private Double cpuUsage;
- private Double memUsage;
- private List<Long> runningJobIds;
+package org.apache.seatunnel.engine.server.serializable;
+
+import
org.apache.seatunnel.engine.server.resourcemanager.opeartion.GetWorkerResourcesOperation;
+
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+
+class ResourceDataSerializerHookTest {
+
+ @Test
+ void shouldRegisterGetWorkerResourcesOperation() {
+ IdentifiedDataSerializable serializable =
+ new ResourceDataSerializerHook()
+ .createFactory()
+
.create(ResourceDataSerializerHook.GET_WORKER_RESOURCES_TYPE);
+
+ assertInstanceOf(GetWorkerResourcesOperation.class, serializable);
+ }
}
