feat: implement precise worker status tracking and monitoring UI
This commit is contained in:
duorameng
@@ -0,0 +1,114 @@
|
||||
package controllers
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"runtime"
|
||||
"time"
|
||||
|
||||
"github.com/engigu/baihu-panel/internal/services/tasks"
|
||||
"github.com/engigu/baihu-panel/internal/utils"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
type MonitorController struct {
|
||||
executorService *tasks.ExecutorService
|
||||
}
|
||||
|
||||
func NewMonitorController(executorService *tasks.ExecutorService) *MonitorController {
|
||||
return &MonitorController{
|
||||
executorService: executorService,
|
||||
}
|
||||
}
|
||||
|
||||
var monitorUpgrader = websocket.Upgrader{
|
||||
CheckOrigin: func(r *http.Request) bool {
|
||||
return true // 开发环境允许所有跨域，生产环境可根据配置限制
|
||||
},
|
||||
}
|
||||
|
||||
// GetSystemMonitor 获取系统和内存监控信息 (HTTP)
|
||||
func (mc *MonitorController) GetSystemMonitor(c *gin.Context) {
|
||||
data := mc.getMonitorData()
|
||||
utils.Success(c, data)
|
||||
}
|
||||
|
||||
// MonitorWS WebSocket实时推送系统监控信息
|
||||
func (mc *MonitorController) MonitorWS(c *gin.Context) {
|
||||
ws, err := monitorUpgrader.Upgrade(c.Writer, c.Request, nil)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer ws.Close()
|
||||
|
||||
ticker := time.NewTicker(3 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
// 先立即发送一次
|
||||
mc.sendMonitorData(ws)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if err := mc.sendMonitorData(ws); err != nil {
|
||||
return // 客户端断开连接或发送失败
|
||||
}
|
||||
case <-c.Request.Context().Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (mc *MonitorController) sendMonitorData(ws *websocket.Conn) error {
|
||||
data := mc.getMonitorData()
|
||||
return ws.WriteJSON(gin.H{
|
||||
"code": 200,
|
||||
"data": data,
|
||||
"msg": "success",
|
||||
})
|
||||
}
|
||||
|
||||
func (mc *MonitorController) getMonitorData() gin.H {
|
||||
var m runtime.MemStats
|
||||
runtime.ReadMemStats(&m)
|
||||
|
||||
return gin.H{
|
||||
"env": gin.H{
|
||||
"os": runtime.GOOS,
|
||||
"arch": runtime.GOARCH,
|
||||
"go_version": runtime.Version(),
|
||||
"num_cpu": runtime.NumCPU(),
|
||||
"goroutines": runtime.NumGoroutine(),
|
||||
},
|
||||
"mem": gin.H{
|
||||
"alloc": m.Alloc,
|
||||
"total_alloc": m.TotalAlloc,
|
||||
"sys": m.Sys,
|
||||
"lookups": m.Lookups,
|
||||
"mallocs": m.Mallocs,
|
||||
"frees": m.Frees,
|
||||
},
|
||||
"heap": gin.H{
|
||||
"heap_alloc": m.HeapAlloc,
|
||||
"heap_sys": m.HeapSys,
|
||||
"heap_idle": m.HeapIdle,
|
||||
"heap_inuse": m.HeapInuse,
|
||||
"heap_released": m.HeapReleased,
|
||||
"heap_objects": m.HeapObjects,
|
||||
},
|
||||
"gc": gin.H{
|
||||
"next_gc": m.NextGC,
|
||||
"last_gc": m.LastGC,
|
||||
"pause_total_ns": m.PauseTotalNs,
|
||||
"num_gc": m.NumGC,
|
||||
},
|
||||
"scheduler": gin.H{
|
||||
"scheduled": mc.executorService.GetScheduledCount(),
|
||||
"running": mc.executorService.GetRunningCount(),
|
||||
"queue_size": mc.executorService.GetScheduler().GetQueueSize(),
|
||||
"worker_count": mc.executorService.GetScheduler().GetConfig().WorkerCount,
|
||||
"workers": mc.executorService.GetScheduler().GetWorkerStatuses(),
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -174,6 +174,16 @@ func (h *schedulerHooksAdapter) OnHeartbeat(ctx context.Context, logID string, d
|
||||
// TaskExecutor 定义任务执行函数签名
|
||||
type TaskExecutor func(ctx context.Context, req *ExecutionRequest, stdout, stderr io.Writer) (*Result, error)
|
||||
|
||||
// WorkerStatus 定义并发池中单个 Worker 的状态
|
||||
type WorkerStatus struct {
|
||||
ID int `json:"id"`
|
||||
Status string `json:"status"` // 状态: "idle" 或 "running"
|
||||
TaskID string `json:"task_id,omitempty"`
|
||||
TaskName string `json:"task_name,omitempty"`
|
||||
StartTime int64 `json:"start_time,omitempty"` // 开始时间戳 (秒)
|
||||
Duration int64 `json:"duration,omitempty"` // 已运行时长 (秒)
|
||||
}
|
||||
|
||||
// Scheduler 统一调度器（独立组件，可在主服务和 Agent 中复用）
|
||||
// 调度器本身只负责队列管理和任务调度，具体的执行逻辑和事件处理由 Handler 实现
|
||||
type Scheduler struct {
|
||||
@@ -188,6 +198,9 @@ type Scheduler struct {
|
||||
logger SchedulerLogger
|
||||
runningTasks map[string]context.CancelFunc // 记录运行中的任务，用于停止 (TaskID -> CancelFunc)
|
||||
runningExecs map[string]context.CancelFunc // 记录运行中的执行，用于停止 (LogID -> CancelFunc)
|
||||
|
||||
workers []WorkerStatus
|
||||
workerMu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewScheduler 创建调度器
|
||||
@@ -224,6 +237,14 @@ func NewScheduler(config SchedulerConfig, handler SchedulerEventHandler) *Schedu
|
||||
logger: &DefaultLogger{},
|
||||
runningTasks: make(map[string]context.CancelFunc),
|
||||
runningExecs: make(map[string]context.CancelFunc),
|
||||
workers: make([]WorkerStatus, config.WorkerCount),
|
||||
}
|
||||
|
||||
for i := 0; i < config.WorkerCount; i++ {
|
||||
s.workers[i] = WorkerStatus{
|
||||
ID: i,
|
||||
Status: "idle",
|
||||
}
|
||||
}
|
||||
|
||||
return s
|
||||
@@ -332,7 +353,32 @@ func (s *Scheduler) worker(id int) {
|
||||
}()
|
||||
// 速率限制
|
||||
<-s.rateLimiter
|
||||
s.executeTask(req)
|
||||
|
||||
func() {
|
||||
// 恢复 worker 状态为空闲
|
||||
defer func() {
|
||||
s.workerMu.Lock()
|
||||
if id >= 0 && id < len(s.workers) {
|
||||
s.workers[id].Status = "idle"
|
||||
s.workers[id].TaskID = ""
|
||||
s.workers[id].TaskName = ""
|
||||
s.workers[id].StartTime = 0
|
||||
}
|
||||
s.workerMu.Unlock()
|
||||
}()
|
||||
|
||||
// 更新 worker 状态为运行中
|
||||
s.workerMu.Lock()
|
||||
if id >= 0 && id < len(s.workers) {
|
||||
s.workers[id].Status = "running"
|
||||
s.workers[id].TaskID = req.TaskID
|
||||
s.workers[id].TaskName = req.Name
|
||||
s.workers[id].StartTime = time.Now().Unix()
|
||||
}
|
||||
s.workerMu.Unlock()
|
||||
|
||||
s.executeTask(req)
|
||||
}()
|
||||
}()
|
||||
}
|
||||
}
|
||||
@@ -615,3 +661,24 @@ func (s *Scheduler) GetConfig() SchedulerConfig {
|
||||
defer s.mu.RUnlock()
|
||||
return s.config
|
||||
}
|
||||
|
||||
// GetWorkerStatuses 获取所有 Worker 的状态
|
||||
func (s *Scheduler) GetWorkerStatuses() []WorkerStatus {
|
||||
s.workerMu.RLock()
|
||||
defer s.workerMu.RUnlock()
|
||||
// 返回副本防止外部修改
|
||||
statuses := make([]WorkerStatus, len(s.workers))
|
||||
now := time.Now().Unix()
|
||||
for i, w := range s.workers {
|
||||
statuses[i] = w
|
||||
// 在服务端计算运行时间，彻底避免客户端与服务端时钟不一致导致的计算偏差
|
||||
if w.Status == "running" && w.StartTime > 0 {
|
||||
duration := now - w.StartTime
|
||||
if duration < 0 {
|
||||
duration = 0
|
||||
}
|
||||
statuses[i].Duration = duration
|
||||
}
|
||||
}
|
||||
return statuses
|
||||
}
|
||||
|
||||
@@ -65,6 +65,7 @@ func initAuthorizedAPIRoutes(api *gin.RouterGroup, c *Controllers) {
|
||||
registerAppLogRoutes(adminOnly, c)
|
||||
registerSystemWSRoutes(adminOnly, c)
|
||||
registerWebUIRoutes(adminOnly, c)
|
||||
registerMonitorRoutes(adminOnly, c)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -268,6 +269,14 @@ func registerSystemWSRoutes(g *gin.RouterGroup, c *Controllers) {
|
||||
g.GET("/ws/events", c.SystemWS.HandleEvents)
|
||||
}
|
||||
|
||||
func registerMonitorRoutes(g *gin.RouterGroup, c *Controllers) {
|
||||
monitor := g.Group("/monitor")
|
||||
{
|
||||
monitor.GET("", c.Monitor.GetSystemMonitor)
|
||||
monitor.GET("/ws", c.Monitor.MonitorWS)
|
||||
}
|
||||
}
|
||||
|
||||
func initAgentAPIRoutes(root *gin.RouterGroup, c *Controllers) {
|
||||
// Agent API（供远程 Agent 调用，不使用 /v1 版本号）
|
||||
agentAPI := root.Group("/api/agent")
|
||||
|
||||
@@ -64,6 +64,7 @@ func RegisterControllers() *Controllers {
|
||||
AppLog: controllers.NewAppLogController(),
|
||||
SystemWS: controllers.NewSystemWSController(),
|
||||
WebUI: controllers.NewWebUIController(services.NewWebUIService(settingsService)),
|
||||
Monitor: controllers.NewMonitorController(executorService),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ type Controllers struct {
|
||||
AppLog *controllers.AppLogController
|
||||
SystemWS *controllers.SystemWSController
|
||||
WebUI *controllers.WebUIController
|
||||
Monitor *controllers.MonitorController
|
||||
}
|
||||
|
||||
func Setup(c *Controllers) *gin.Engine {
|
||||
|
||||
Reference in New Issue
Block a user