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

fix: logs io buffer combine

Browse Source
This commit is contained in:
engigu
2026-02-08 11:03:11 +08:00
parent b291e3740f
commit 6402943462
9 changed files with 67 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/agent.go
+4 -1
View File
@@ -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,
internal/executor/executor.go
+2
View File
@@ -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 {
internal/executor/scheduler.go
+31 -17
View File
@@ -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)
}
}
internal/models/agent.go
+1
View File
@@ -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"`
internal/models/task.go
+1
View File
@@ -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"`
internal/services/tasks/executor_service.go
+14 -8
View File
@@ -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 结果超时")
}
}
internal/services/tasks/task_log_service.go
+3 -1
View File
@@ -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,
web/src/api/index.ts
+2
View File
@@ -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
web/src/views/history/History.vue
+9
View File
@@ -340,6 +340,15 @@ watch(() => route.query.task_id, (newTaskId) => {
</div>
</div>
<div class="flex-1 flex flex-col overflow-hidden">
<div v-if="selectedLog.error" class="px-4 py-3 border-b bg-red-500/5 space-y-2 text-sm">
<div class="flex items-center gap-2 text-red-500 font-medium">
<X class="h-4 w-4" />
<span>系统错误</span>
</div>
<code class="block font-mono bg-red-500/10 text-red-600 px-2 py-1 rounded text-xs break-all">
{{ selectedLog.error }}
</code>
</div>
<div class="px-4 py-2 text-sm text-muted-foreground border-b bg-muted/50 flex items-center justify-between">
<span>输出</span>
<Button variant="ghost" size="icon" class="h-6 w-6" @click="showFullscreen = true" title="全屏查看">