diff --git a/agent/agent.go b/agent/agent.go index 114dfff..9334981 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -81,6 +81,7 @@ type TaskResult struct { AgentID uint `json:"agent_id"` // 仅用于 HTTP 上报时后端补充 Command string `json:"command"` Output string `json:"output"` + Error string `json:"error"` Status string `json:"status"` Duration int64 `json:"duration"` ExitCode int `json:"exit_code"` @@ -175,6 +176,7 @@ func (h *AgentHandler) OnTaskCompleted(req *executor.ExecutionRequest, result *e LogID: result.LogID, Command: req.Command, Output: result.Output, + Error: result.Error, Status: result.Status, Duration: result.Duration, ExitCode: result.ExitCode, @@ -203,7 +205,8 @@ func (h *AgentHandler) OnTaskFailed(req *executor.ExecutionRequest, err error) { TaskID: taskID, LogID: req.LogID, Command: req.Command, - Output: errMsg, + Output: "", + Error: err.Error(), Status: "failed", Duration: 0, ExitCode: 1, diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 8d7a5c5..096e294 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -36,6 +36,7 @@ type Request struct { // Result 任务执行结果 type Result struct { Output string + Error string Status string // success, failed Duration int64 // 毫秒 ExitCode int @@ -160,6 +161,7 @@ func ExecuteWithHooks(ctx context.Context, req Request, stdout, stderr io.Writer if err != nil { result.Status = "failed" + result.Error = err.Error() if exitErr, ok := err.(*exec.ExitError); ok { result.ExitCode = exitErr.ExitCode() } else { diff --git a/internal/executor/scheduler.go b/internal/executor/scheduler.go index e5c1530..77c7b59 100644 --- a/internal/executor/scheduler.go +++ b/internal/executor/scheduler.go @@ -10,6 +10,24 @@ import ( "time" ) +// safeBuffer 一个线程安全的字节缓冲区,用于合并 stdout 和 stderr +type safeBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (s *safeBuffer) Write(p []byte) (n int, err error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.Write(p) +} + +func (s *safeBuffer) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.String() +} + // SchedulerConfig 调度器配置 type SchedulerConfig struct { WorkerCount int // Worker 数量 @@ -319,20 +337,20 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) } } - // 2. 准备输出缓冲区 - var stdoutBuf, stderrBuf bytes.Buffer + // 2. 准备输出缓冲区(使用合并缓冲区保证顺序) + var combinedBuf safeBuffer var stdoutWriter, stderrWriter io.Writer if stdout != nil { - stdoutWriter = io.MultiWriter(&stdoutBuf, stdout) + stdoutWriter = io.MultiWriter(&combinedBuf, stdout) } else { - stdoutWriter = &stdoutBuf + stdoutWriter = &combinedBuf } if stderr != nil { - stderrWriter = io.MultiWriter(&stderrBuf, stderr) + stderrWriter = io.MultiWriter(&combinedBuf, stderr) } else { - stderrWriter = &stderrBuf + stderrWriter = &combinedBuf } // 3. 实际开始执行事件 (经过队列和速率限制之后) @@ -369,7 +387,7 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) if execResult != nil { result.Success = execResult.Status == "success" - result.Output = stdoutBuf.String() + result.Output = combinedBuf.String() result.Status = execResult.Status result.Duration = execResult.Duration result.ExitCode = execResult.ExitCode @@ -381,17 +399,11 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) result.StartTime = start result.EndTime = time.Now() result.Duration = result.EndTime.Sub(result.StartTime).Milliseconds() + result.Output = combinedBuf.String() } if execErr != nil { result.Error = execErr.Error() - if result.Output == "" { - result.Output = execErr.Error() - } - errOutput := stderrBuf.String() - if errOutput != "" { - result.Output += "\n[ERROR]\n" + errOutput - } if ctx.Err() == context.Canceled { result.Status = "cancelled" } else if ctx.Err() == context.DeadlineExceeded { @@ -401,10 +413,12 @@ func (s *Scheduler) executeTask(req *ExecutionRequest) (*ExecutionResult, error) // 6. 执行后事件 if s.handler != nil { - if execErr != nil { - s.handler.OnTaskFailed(req, execErr) - } else { + if execResult != nil { + // 只要有执行结果(即使执行失败),都认为是任务完成了(包含输出) s.handler.OnTaskCompleted(req, result) + } else if execErr != nil { + // 只有在完全没有结果的情况下(如无法启动、Panic等),才认为是任务失败 + s.handler.OnTaskFailed(req, execErr) } } diff --git a/internal/models/agent.go b/internal/models/agent.go index 7100262..5049537 100644 --- a/internal/models/agent.go +++ b/internal/models/agent.go @@ -69,6 +69,7 @@ type AgentTaskResult struct { AgentID uint `json:"agent_id"` Command string `json:"command"` Output string `json:"output"` + Error string `json:"error"` // 额外的系统错误信息 Status string `json:"status"` // success, failed Duration int64 `json:"duration"` // milliseconds ExitCode int `json:"exit_code"` diff --git a/internal/models/task.go b/internal/models/task.go index 667f43d..7ebc6f0 100644 --- a/internal/models/task.go +++ b/internal/models/task.go @@ -85,6 +85,7 @@ type TaskLog struct { AgentID *uint `json:"agent_id" gorm:"index"` // Agent ID,为空表示本地执行 Command string `json:"command" gorm:"type:text"` Output string `json:"-" gorm:"type:longtext"` // gzip+base64 compressed + Error string `json:"error" gorm:"type:text"` // 额外的系统错误信息 Status string `json:"status" gorm:"size:20"` // success, failed Duration int64 `json:"duration"` // milliseconds ExitCode int `json:"exit_code"` diff --git a/internal/services/tasks/executor_service.go b/internal/services/tasks/executor_service.go index 77234cf..0e15332 100644 --- a/internal/services/tasks/executor_service.go +++ b/internal/services/tasks/executor_service.go @@ -211,6 +211,7 @@ func (h *ServerSchedulerHandler) OnTaskCompleted(req *executor.ExecutionRequest, TaskID: task.ID, Command: req.Command, Output: output, + Error: result.Error, Status: result.Status, Duration: result.Duration, ExitCode: result.ExitCode, @@ -262,13 +263,16 @@ func (h *ServerSchedulerHandler) OnTaskFailed(req *executor.ExecutionRequest, er now := models.LocalTime(time.Now()) taskLog := &models.TaskLog{ - ID: req.LogID, - TaskID: taskID, - Output: output, - Status: "failed", - Duration: 0, - ExitCode: 1, - EndTime: &now, + ID: req.LogID, + TaskID: taskID, + Command: req.Command, + Output: output, + Error: err.Error(), + Status: "failed", + Duration: 0, + ExitCode: 1, + StartTime: &now, + EndTime: &now, } // 补充 AgentID @@ -645,6 +649,7 @@ func (es *ExecutorService) ExecuteRemoteForScheduler(task *models.Task, logID ui case agentResult := <-resultChan: return &executor.Result{ Output: agentResult.Output, + Error: agentResult.Error, Status: agentResult.Status, Duration: agentResult.Duration, ExitCode: agentResult.ExitCode, @@ -655,11 +660,12 @@ func (es *ExecutorService) ExecuteRemoteForScheduler(task *models.Task, logID ui end := time.Now() return &executor.Result{ Status: "failed", + Error: "等待 Agent 结果超时", Duration: end.Sub(start).Milliseconds(), ExitCode: -1, StartTime: start, EndTime: end, - }, fmt.Errorf("远程执行超时") + }, fmt.Errorf("等待 Agent 结果超时") } } diff --git a/internal/services/tasks/task_log_service.go b/internal/services/tasks/task_log_service.go index eb59072..e459151 100644 --- a/internal/services/tasks/task_log_service.go +++ b/internal/services/tasks/task_log_service.go @@ -156,6 +156,7 @@ func (s *TaskLogService) CreateTaskLogFromAgentResult(result *models.AgentTaskRe AgentID: &result.AgentID, Command: result.Command, Output: compressed, + Error: result.Error, Status: result.Status, Duration: result.Duration, ExitCode: result.ExitCode, @@ -175,7 +176,7 @@ func (s *TaskLogService) CreateTaskLogFromAgentResult(result *models.AgentTaskRe } // CreateTaskLogFromLocalExecution 从本地执行结果创建任务日志 -func (s *TaskLogService) CreateTaskLogFromLocalExecution(taskID uint, command, output, status string, duration int64, exitCode int, start, end time.Time, isCompressed bool) (*models.TaskLog, error) { +func (s *TaskLogService) CreateTaskLogFromLocalExecution(taskID uint, command, output, systemErr, status string, duration int64, exitCode int, start, end time.Time, isCompressed bool) (*models.TaskLog, error) { var compressed string var err error @@ -197,6 +198,7 @@ func (s *TaskLogService) CreateTaskLogFromLocalExecution(taskID uint, command, o TaskID: taskID, Command: command, Output: compressed, + Error: systemErr, Status: status, Duration: duration, ExitCode: exitCode, diff --git a/web/src/api/index.ts b/web/src/api/index.ts index 62f30e7..0d35c8d 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -319,6 +319,7 @@ export interface TaskLog { command: string status: string duration: number + error: string | null start_time: string | null end_time: string | null created_at: string @@ -336,6 +337,7 @@ export interface LogDetail { task_id: number command: string output: string + error: string | null status: string duration: number start_time: string | null diff --git a/web/src/views/history/History.vue b/web/src/views/history/History.vue index 4cf2c18..8787185 100644 --- a/web/src/views/history/History.vue +++ b/web/src/views/history/History.vue @@ -340,6 +340,15 @@ watch(() => route.query.task_id, (newTaskId) => {
+
+
+ + 系统错误 +
+ + {{ selectedLog.error }} + +
输出