<feature>[core]: add APIShowPoolStatus API for thread pool and task queue monitoring

Resolves: ZSTAC-68709

Change-Id: Iee14ef5d21c216e91312ac40a46be80222244fbb

Author: AlanJager

core/src/main/java/org/zstack/core/CoreManagerImpl.java

@@ -8,7 +8,7 @@
 import org.zstack.core.singleflight.ExternalSingleFlightMsg;
 import org.zstack.core.singleflight.ExternalSingleFlightReply;
 import org.zstack.core.singleflight.MultiNodeSingleFlightImpl;
-import org.zstack.core.thread.ThreadFacade;
+import org.zstack.core.thread.*;
 import org.zstack.header.AbstractService;
 import org.zstack.header.core.*;
 import org.zstack.header.core.progress.ChainInfo;
@@ -41,6 +41,8 @@ public class CoreManagerImpl extends AbstractService implements CoreManager {
     @Autowired
     private ThreadFacade thdf;
     @Autowired
+    private DispatchQueue dpq;
+    @Autowired
     private MultiNodeSingleFlightImpl singleFlight;
 
     @Override
@@ -55,6 +57,8 @@ public void handleMessage(Message msg) {
     private void handleApiMessage(APIMessage msg) {
         if (msg instanceof APIGetChainTaskMsg) {
             handleMessage((APIGetChainTaskMsg) msg);
+        } else if (msg instanceof APIShowPoolStatusMsg) {
+            handleMessage((APIShowPoolStatusMsg) msg);
         }
     }
 
@@ -134,6 +138,43 @@ public void done(ErrorCodeList errorCodeList) {
         });
     }
 
+    private void handleMessage(APIShowPoolStatusMsg msg) {
+        APIShowPoolStatusReply reply = new APIShowPoolStatusReply();
+
+        // Thread pool stats
+        if (thdf instanceof ThreadFacadeMXBean) {
+            ThreadPoolStatistic tps = ((ThreadFacadeMXBean) thdf).getThreadPoolStatistic();
+            reply.addEntry(new PoolStatusEntry("main-thread-pool", "ThreadPool",
+                    tps.getActiveThreadNum(), (int) tps.getQueuedTaskNum(), (int) tps.getMaxPoolSize()));
+        }
+
+        // Chain task stats (top 20 by pending)
+        Map<String, ChainTaskStatistic> chainStats = dpq.getChainTaskStatistics();
+        chainStats.entrySet().stream()
+                .sorted((a, b) -> Long.compare(b.getValue().getPendingTaskNum(), a.getValue().getPendingTaskNum()))
+                .limit(20)
+                .forEach(e -> {
+                    ChainTaskStatistic s = e.getValue();
+                    reply.addEntry(new PoolStatusEntry(
+                            "chain:" + e.getKey(), "ChainTask",
+                            s.getCurrentRunningThreadNum(), (int) s.getPendingTaskNum(), s.getSyncLevel()));
+                });
+
+        // Sync task stats (top 20 by pending)
+        Map<String, SyncTaskStatistic> syncStats = dpq.getSyncTaskStatistics();
+        syncStats.entrySet().stream()
+                .sorted((a, b) -> Long.compare(b.getValue().getPendingTaskNum(), a.getValue().getPendingTaskNum()))
+                .limit(20)
+                .forEach(e -> {
+                    SyncTaskStatistic s = e.getValue();
+                    reply.addEntry(new PoolStatusEntry(
+                            "sync:" + e.getKey(), "SyncTask",
+                            s.getCurrentRunningThreadNum(), (int) s.getPendingTaskNum(), s.getSyncLevel()));
+                });
+
+        bus.reply(msg, reply);
+    }
+
     private void handle(GetLocalTaskMsg msg) {
         GetLocalTaskReply reply = new GetLocalTaskReply();
         Map<String, ChainInfo> results = msg.getSyncSignatures().stream()

header/src/main/java/org/zstack/header/core/APIShowPoolStatusMsg.java

@@ -0,0 +1,15 @@
+package org.zstack.header.core;
+
+import org.springframework.http.HttpMethod;
+import org.zstack.header.identity.Action;
+import org.zstack.header.message.APISyncCallMessage;
+import org.zstack.header.rest.RestRequest;
+
+@Action(category = CoreConstant.ACTION_CATEGORY, adminOnly = true)
+@RestRequest(
+        path = "/core/pool-status",
+        method = HttpMethod.GET,
+        responseClass = APIShowPoolStatusReply.class
+)
+public class APIShowPoolStatusMsg extends APISyncCallMessage {
+}

header/src/main/java/org/zstack/header/core/APIShowPoolStatusReply.java

@@ -0,0 +1,24 @@
+package org.zstack.header.core;
+
+import org.zstack.header.message.APIReply;
+import org.zstack.header.rest.RestResponse;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@RestResponse(fieldsTo = {"all"})
+public class APIShowPoolStatusReply extends APIReply {
+    private List<PoolStatusEntry> entries = new ArrayList<>();
+
+    public List<PoolStatusEntry> getEntries() {
+        return entries;
+    }
+
+    public void setEntries(List<PoolStatusEntry> entries) {
+        this.entries = entries;
+    }
+
+    public void addEntry(PoolStatusEntry entry) {
+        entries.add(entry);
+    }
+}

header/src/main/java/org/zstack/header/core/PoolStatusEntry.java

@@ -0,0 +1,80 @@
+package org.zstack.header.core;
+
+public class PoolStatusEntry {
+    private String name;
+    private String poolType;
+    private int active;
+    private int pending;
+    private int max;
+    private double utilization;
+    private String status;
+
+    public PoolStatusEntry() {
+    }
+
+    public PoolStatusEntry(String name, String poolType, int active, int pending, int max) {
+        this.name = name;
+        this.poolType = poolType;
+        this.active = active;
+        this.pending = pending;
+        this.max = max;
+        this.utilization = max > 0 ? (double) active * 100 / max : 0;
+        this.status = utilization >= 80 ? "CRITICAL" : utilization >= 60 ? "WARNING" : "OK";
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getPoolType() {
+        return poolType;
+    }
+
+    public void setPoolType(String poolType) {
+        this.poolType = poolType;
+    }
+
+    public int getActive() {
+        return active;
+    }
+
+    public void setActive(int active) {
+        this.active = active;
+    }
+
+    public int getPending() {
+        return pending;
+    }
+
+    public void setPending(int pending) {
+        this.pending = pending;
+    }
+
+    public int getMax() {
+        return max;
+    }
+
+    public void setMax(int max) {
+        this.max = max;
+    }
+
+    public double getUtilization() {
+        return utilization;
+    }
+
+    public void setUtilization(double utilization) {
+        this.utilization = utilization;
+    }
+
+    public String getStatus() {
+        return status;
+    }
+
+    public void setStatus(String status) {
+        this.status = status;
+    }
+}