diff --git a/agent/agent.go b/agent/agent.go index 24a2b4b..114dfff 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -101,17 +101,21 @@ type Agent struct { wsConn *websocket.Conn wsMu sync.Mutex stopCh chan struct{} - wsStopCh chan struct{} // 用于停止当前 WebSocket 相关的 goroutine + wsStopCh chan struct{} // 用于停止当前 WebSocket 相关的 goroutine + taskLogs map[uint][]string // 记录最近的日志行,用于失败显示 + logMu sync.Mutex // taskLogs 的锁 } func NewAgent(config *Config, configFile string) *Agent { a := &Agent{ - config: config, - configFile: configFile, - machineID: utils.GenerateMachineID(), - tasks: make(map[uint]*AgentTask), - client: &http.Client{Timeout: 30 * time.Second}, - stopCh: make(chan struct{}), + config: config, + configFile: configFile, + machineID: utils.GenerateMachineID(), + tasks: make(map[uint]*AgentTask), + client: &http.Client{Timeout: 30 * time.Second}, + stopCh: make(chan struct{}), + lastTaskCount: -1, + taskLogs: make(map[uint][]string), } // 初始化调度器 @@ -152,6 +156,12 @@ func (h *AgentHandler) OnTaskHeartbeat(req *executor.ExecutionRequest, duration "duration": duration, }) } + + // 每分钟打印一次任务还在运行的日志,提升长任务的存在感 + if duration >= 60000 && (duration/60000 > (duration-3000)/60000) { + logger.Infof("[Scheduler] 任务 #%s 仍在运行中... (已耗时: %v)", + req.TaskID, (time.Duration(duration) * time.Millisecond).Round(time.Second)) + } } func (h *AgentHandler) OnTaskStarted(req *executor.ExecutionRequest) {} @@ -171,6 +181,11 @@ func (h *AgentHandler) OnTaskCompleted(req *executor.ExecutionRequest, result *e StartTime: result.StartTime.Unix(), EndTime: result.EndTime.Unix(), }) + + if result.Status == "failed" { + h.agent.printLastLogs(result.LogID) + } + h.agent.clearTaskLog(result.LogID) } func (h *AgentHandler) OnTaskFailed(req *executor.ExecutionRequest, err error) { @@ -195,6 +210,9 @@ func (h *AgentHandler) OnTaskFailed(req *executor.ExecutionRequest, err error) { StartTime: time.Now().Unix(), EndTime: time.Now().Unix(), }) + + h.agent.printLastLogs(req.LogID) + h.agent.clearTaskLog(req.LogID) } func (h *AgentHandler) OnCronNextRun(req *executor.ExecutionRequest, nextRun time.Time) {} @@ -345,6 +363,7 @@ func (a *Agent) handleWSMessage(msg *WSMessage) { } func (a *Agent) fetchTasks() { + logger.Info("正在从服务器拉取任务列表...") if err := a.sendWSMessage(WSTypeFetchTasks, map[string]interface{}{}); err != nil { logger.Warnf("请求任务列表失败: %v", err) } @@ -431,8 +450,8 @@ func (a *Agent) handleTasks(data json.RawMessage) { json.Unmarshal(data, &resp) newCount := len(resp.Tasks) - if newCount != a.lastTaskCount { - logger.Infof("任务列表更新: %d -> %d 个任务", a.lastTaskCount, newCount) + if newCount != a.lastTaskCount || newCount == 0 { + logger.Infof("任务列表同步成功: 共获取到 %d 个任务", newCount) a.lastTaskCount = newCount } @@ -486,6 +505,9 @@ func (w *RealTimeLogWriter) Write(p []byte) (n int, err error) { return 0, nil } + // 记录到本地缓存,用于失败时显示 + w.agent.addTaskLog(w.logID, p) + // 构造消息 msg := map[string]interface{}{ "log_id": w.logID, @@ -596,7 +618,7 @@ func (a *Agent) updateTasks(tasks []AgentTask) { if _, exists := newTasks[id]; !exists { a.cronManager.RemoveTask(fmt.Sprintf("%d", id)) delete(a.tasks, id) - logger.Infof("移除任务 #%d", id) + logger.Infof("移除调度任务 #%d", id) } } @@ -607,13 +629,13 @@ func (a *Agent) updateTasks(tasks []AgentTask) { if task.Enabled { err := a.cronManager.AddTask(task) if err != nil { - logger.Errorf("调度任务 #%d 失败: %v", id, err) + logger.Errorf("添加调度任务 #%d 失败: %v", id, err) continue } - logger.Infof("已调度任务 #%d %s (%s)", id, task.Name, task.GetSchedule()) + logger.Infof("已添加调度任务 #%d %s (%s)", id, task.Name, task.GetSchedule()) } else { a.cronManager.RemoveTask(fmt.Sprintf("%d", id)) - logger.Infof("任务 #%d 已禁用", id) + logger.Infof("调度任务 #%d 已禁用", id) } a.tasks[id] = task } @@ -634,6 +656,50 @@ func (a *Agent) clearAllTasks() { logger.Info("所有任务已清空") } +func (a *Agent) addTaskLog(logID uint, p []byte) { + if logID == 0 { + return + } + a.logMu.Lock() + defer a.logMu.Unlock() + + content := string(p) + lines := strings.Split(strings.TrimSuffix(content, "\n"), "\n") + + a.taskLogs[logID] = append(a.taskLogs[logID], lines...) + if len(a.taskLogs[logID]) > 50 { + a.taskLogs[logID] = a.taskLogs[logID][len(a.taskLogs[logID])-50:] + } +} + +func (a *Agent) printLastLogs(logID uint) { + if logID == 0 { + return + } + a.logMu.Lock() + lines, ok := a.taskLogs[logID] + a.logMu.Unlock() + + if !ok || len(lines) == 0 { + return + } + + logger.Errorf("--- 任务 #%d 失败日志预览 (最近 %d 行) ---", logID, len(lines)) + for _, line := range lines { + fmt.Println(" " + line) + } + logger.Errorf("--- 任务 #%d 结束 ---", logID) +} + +func (a *Agent) clearTaskLog(logID uint) { + if logID == 0 { + return + } + a.logMu.Lock() + defer a.logMu.Unlock() + delete(a.taskLogs, logID) +} + // executeTask 已被 AgentHandler.OnTaskCompleted 代替,此处删除旧实现 func (a *Agent) doRequest(method, path string, body interface{}) (*http.Response, error) { diff --git a/internal/controllers/agent_controller.go b/internal/controllers/agent_controller.go index 7680286..d1a7e4e 100644 --- a/internal/controllers/agent_controller.go +++ b/internal/controllers/agent_controller.go @@ -439,6 +439,9 @@ func (c *AgentController) WSConnect(ctx *gin.Context) { // 启动读写协程 go c.wsWritePump(ac) go c.wsReadPump(ac, agent) + + // 主动推送任务列表 + go c.wsManager.BroadcastTasks(agent.ID) } // wsReadPump 读取消息 diff --git a/internal/executor/cron.go b/internal/executor/cron.go index e492e43..c005641 100644 --- a/internal/executor/cron.go +++ b/internal/executor/cron.go @@ -77,6 +77,11 @@ func (m *CronManager) AddTask(task CronTask) error { timeout := task.GetTimeout() entryID, err := m.cron.AddFunc(task.GetSchedule(), func() { + defer func() { + if r := recover(); r != nil { + m.logger.Errorf("[CronManager] 任务 #%s 执行过程中发生 Panic: %v", taskID, r) + } + }() m.logger.Infof("[CronManager] 触发计划任务 #%s (%s)", taskID, name) req := &ExecutionRequest{ diff --git a/internal/executor/scheduler.go b/internal/executor/scheduler.go index 12271f2..e5c1530 100644 --- a/internal/executor/scheduler.go +++ b/internal/executor/scheduler.go @@ -262,15 +262,27 @@ func (s *Scheduler) worker(id int) { case <-s.stopCh: return case req := <-s.taskQueue: - // 速率限制 - <-s.rateLimiter - s.executeTask(req) + func() { + defer func() { + if r := recover(); r != nil { + s.logger.Errorf("[Scheduler] Worker %d panic while processing task %s: %v", id, req.TaskID, r) + } + }() + // 速率限制 + <-s.rateLimiter + s.executeTask(req) + }() } } } // executeTask 执行任务(本地执行) func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) { + defer func() { + if r := recover(); r != nil { + s.logger.Errorf("[Scheduler] 任务 %s 执行过程中发生 Panic: %v", req.TaskID, r) + } + }() start := time.Now() s.logger.Infof("[Scheduler] 执行任务 %s (名称: %s, 类型: %s)", req.TaskID, req.Name, req.Type) diff --git a/internal/services/agent_ws_service.go b/internal/services/agent_ws_service.go index 1f8805c..f302b07 100644 --- a/internal/services/agent_ws_service.go +++ b/internal/services/agent_ws_service.go @@ -271,48 +271,55 @@ func (m *AgentWSManager) cleanupLoop() { NewAgentService().ResetAllAgentsToOffline() for range ticker.C { - m.mu.Lock() - now := time.Now() + func() { + defer func() { + if r := recover(); r != nil { + logger.Errorf("[AgentWS] cleanupLoop panic: %v", r) + } + }() + m.mu.Lock() + now := time.Now() - // 清理超时连接 - for agentID, conn := range m.connections { - if now.Sub(conn.LastPing) > 2*time.Minute { - // 减少 IP 连接计数 - if conn.IP != "" { - if count, ok := m.ipConnections[conn.IP]; ok && count > 0 { - m.ipConnections[conn.IP] = count - 1 + // 清理超时连接 + for agentID, conn := range m.connections { + if now.Sub(conn.LastPing) > 2*time.Minute { + // 减少 IP 连接计数 + if conn.IP != "" { + if count, ok := m.ipConnections[conn.IP]; ok && count > 0 { + m.ipConnections[conn.IP] = count - 1 + } + } + conn.Close() + delete(m.connections, agentID) + // 更新数据库状态 + database.DB.Model(&models.Agent{}).Where("id = ?", agentID).Update("status", "offline") + logger.Infof("[AgentWS] Agent #%d 心跳超时,已断开", agentID) + } + } + + // 定期清理数据库中的过期状态(处理服务重启或异常终止的情况) + // 有些 Agent 虽然没有连接,但数据库状态可能是 "online" + cutoff := now.Add(-2 * time.Minute) + database.DB.Model(&models.Agent{}). + Where("status = ? AND last_seen < ?", "online", cutoff). + Update("status", "offline") + + // 清理过期的限流记录(超过 10 分钟未活动) + + // 清理过期的限流记录(超过 10 分钟未活动) + for ip, lastAttempt := range m.ipLastAttempt { + if now.Sub(lastAttempt) > 10*time.Minute { + delete(m.ipLastAttempt, ip) + delete(m.ipFailCount, ip) + // 只清理没有活跃连接的 IP 计数 + if m.ipConnections[ip] == 0 { + delete(m.ipConnections, ip) } } - conn.Close() - delete(m.connections, agentID) - // 更新数据库状态 - database.DB.Model(&models.Agent{}).Where("id = ?", agentID).Update("status", "offline") - logger.Infof("[AgentWS] Agent #%d 心跳超时,已断开", agentID) } - } - // 定期清理数据库中的过期状态(处理服务重启或异常终止的情况) - // 有些 Agent 虽然没有连接,但数据库状态可能是 "online" - cutoff := now.Add(-2 * time.Minute) - database.DB.Model(&models.Agent{}). - Where("status = ? AND last_seen < ?", "online", cutoff). - Update("status", "offline") - - // 清理过期的限流记录(超过 10 分钟未活动) - - // 清理过期的限流记录(超过 10 分钟未活动) - for ip, lastAttempt := range m.ipLastAttempt { - if now.Sub(lastAttempt) > 10*time.Minute { - delete(m.ipLastAttempt, ip) - delete(m.ipFailCount, ip) - // 只清理没有活跃连接的 IP 计数 - if m.ipConnections[ip] == 0 { - delete(m.ipConnections, ip) - } - } - } - - m.mu.Unlock() + m.mu.Unlock() + }() } } diff --git a/internal/services/tasks/executor_service.go b/internal/services/tasks/executor_service.go index e13789d..77234cf 100644 --- a/internal/services/tasks/executor_service.go +++ b/internal/services/tasks/executor_service.go @@ -161,6 +161,12 @@ func (h *ServerSchedulerHandler) OnTaskHeartbeat(req *executor.ExecutionRequest, if req.LogID > 0 { h.es.taskLogService.UpdateTaskDuration(req.LogID, duration) } + + // 每分钟打印一次任务还在运行的日志 + if duration >= 60000 && (duration/60000 > (duration-3000)/60000) { + logger.Infof("[Scheduler] 任务 #%s 仍在运行中... (已耗时: %v)", + req.TaskID, (time.Duration(duration) * time.Millisecond).Round(time.Second)) + } } func (h *ServerSchedulerHandler) OnTaskStarted(req *executor.ExecutionRequest) {