admin/TaskPool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: remote agent exec errors

Browse Source
This commit is contained in:
engigu
2026-02-08 10:18:39 +08:00
parent c3c79f25bd
commit b291e3740f
6 changed files with 152 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/agent.go
+79 -13
View File
@@ -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) {
internal/controllers/agent_controller.go
+3
View File
@@ -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 读取消息
internal/executor/cron.go
+5
View File
@@ -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{
internal/executor/scheduler.go
+15 -3
View File
@@ -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)
internal/services/agent_ws_service.go
+44 -37
View File
@@ -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()
}()
}
}
internal/services/tasks/executor_service.go
+6
View File
@@ -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) {