From 61eeb0c8acab3775121ec7a38725996713b88ad3 Mon Sep 17 00:00:00 2001 From: hezhaohui Date: Wed, 29 Jul 2026 14:00:40 +0800 Subject: [PATCH] =?UTF-8?q?fix(server):=20=E4=BF=AE=E5=A4=8D=20task=20log?= =?UTF-8?q?=20=E6=8E=A5=E5=8F=A3=E5=A4=9A=E4=B8=AA=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 修复 Wayne 任务状态映射，将数字状态统一为字符串状态 - 添加分页支持，支持 page/page_size 参数 - 添加 AWXJobStdout 内存缓存，减少重复 API 调用 - 修复 SubscribeTask 内存泄漏，自动清理 closed channel - 统一 Stream 端点，AWX 使用 pub/sub，Wayne 使用轮询 背景：task log 接口存在多个潜在问题，包括状态判断错误导致无限轮询、 缺少分页、重复 API 调用性能问题、内存泄漏风险等 关联 commit：fix/logs 分支 --- server/internal/handler/task_log.go | 169 +++++++++++++++++++++++++--- server/internal/service/delivery.go | 104 ++++++++++++++++- 2 files changed, 250 insertions(+), 23 deletions(-) diff --git a/server/internal/handler/task_log.go b/server/internal/handler/task_log.go index 81bf4bc..5857053 100644 --- a/server/internal/handler/task_log.go +++ b/server/internal/handler/task_log.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/1024XEngineer/xinfra/server/internal/auth" "github.com/1024XEngineer/xinfra/server/internal/model" "github.com/1024XEngineer/xinfra/server/internal/service" "github.com/gin-gonic/gin" @@ -55,6 +56,16 @@ func (h *TaskLogHandler) List(c *gin.Context) { source := strings.ToLower(strings.TrimSpace(c.DefaultQuery("source", "all"))) businessLineID, _ := strconv.ParseUint(c.Query("business_line_id"), 10, 64) + // 分页参数 + page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) + pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "20")) + if page < 1 { + page = 1 + } + if pageSize < 1 || pageSize > 100 { + pageSize = 20 + } + items := make([]taskLogSummary, 0) if source == "" || source == "all" || source == "awx" { awxItems, err := h.listAWXTasks(c, claims.UserID, claims.IsAdmin, businessLineID) @@ -73,10 +84,27 @@ func (h *TaskLogHandler) List(c *gin.Context) { items = append(items, wayneItems...) } sortTaskLogSummaries(items) - if len(items) > 100 { - items = items[:100] + + // 计算分页 + total := len(items) + start := (page - 1) * pageSize + if start >= total { + items = make([]taskLogSummary, 0) + } else { + end := start + pageSize + if end > total { + end = total + } + items = items[start:end] } - c.JSON(http.StatusOK, gin.H{"items": items}) + + c.JSON(http.StatusOK, gin.H{ + "items": items, + "total": total, + "page": page, + "page_size": pageSize, + "total_pages": (total + pageSize - 1) / pageSize, + }) } func (h *TaskLogHandler) Get(c *gin.Context) { @@ -211,7 +239,7 @@ func wayneTaskSummary(history service.WayneDeploymentHistory) taskLogSummary { Service: "wayne-deployment", Name: wayneDeploymentTaskName(history), Runner: "Wayne Native API", - Status: strconv.Itoa(history.Status), + Status: wayneStatusToString(history.Status), StatusText: textForWaynePublishStatus(history.Status), StatusClass: classForWaynePublishStatus(history.Status), BusinessLineID: history.BusinessLineID, @@ -325,6 +353,18 @@ func classForWaynePublishStatus(status int) string { } } +// wayneStatusToString 将 Wayne 数字状态映射为与 AWX 一致的字符串状态 +func wayneStatusToString(status int) string { + switch status { + case 1: + return model.TaskFinished + case 0: + return model.TaskExecutionFailed + default: + return "unknown" + } +} + func splitStdoutLines(stdout string) []taskLogLine { lines := make([]taskLogLine, 0) for _, line := range strings.Split(stdout, "\n") { @@ -352,6 +392,7 @@ func classForOutputLine(line string) string { } // Stream SSE 端点，用于实时推送任务日志。 +// AWX 任务使用 pub/sub 模式，Wayne 任务使用轮询模式。 func (h *TaskLogHandler) Stream(c *gin.Context) { claims, ok := CurrentClaims(c) if !ok { @@ -373,10 +414,94 @@ func (h *TaskLogHandler) Stream(c *gin.Context) { ctx := c.Request.Context() + // AWX 任务使用 pub/sub 模式 + if strings.HasPrefix(taskID, "awx:") { + h.streamAWXTask(c, taskID, ctx, claims) + return + } + + // Wayne 任务使用轮询模式（因为 Wayne API 不支持推送） + if strings.HasPrefix(taskID, "wayne:") { + h.streamWayneTask(c, taskID, ctx, claims) + return + } + + c.SSEvent("message", gin.H{"type": "error", "error": "invalid task_id format"}) + c.Writer.Flush() +} + +// streamAWXTask AWX 任务使用 pub/sub 模式 +func (h *TaskLogHandler) streamAWXTask(c *gin.Context, taskID string, ctx context.Context, claims *auth.Claims) { + actualTaskID := strings.TrimPrefix(taskID, "awx:") + // 发送初始日志 initialLines, taskStatus, err := h.fetchTaskLogLines(ctx, taskID, claims.UserID, claims.IsAdmin) if err != nil { c.SSEvent("message", gin.H{"type": "error", "error": err.Error()}) + c.Writer.Flush() + return + } + + // 发送初始数据 + c.SSEvent("message", gin.H{"type": "init", "lines": initialLines}) + c.Writer.Flush() + + // 如果任务已完成，直接发送结束事件 + if isTerminalStatus(taskStatus) { + c.SSEvent("message", gin.H{"type": "finished"}) + c.Writer.Flush() + return + } + + // 订阅任务更新（使用 pub/sub 模式） + updates, cancel := h.delivery.SubscribeTask(actualTaskID) + defer cancel() + + heartbeatTicker := time.NewTicker(15 * time.Second) + defer heartbeatTicker.Stop() + + for { + select { + case <-ctx.Done(): + return + case _, ok := <-updates: + if !ok { + return + } + + // 重新获取日志行 + lines, status, err := h.fetchTaskLogLines(ctx, taskID, claims.UserID, claims.IsAdmin) + if err != nil { + c.SSEvent("message", gin.H{"type": "error", "error": err.Error()}) + c.Writer.Flush() + continue + } + + // 发送完整日志 + c.SSEvent("message", gin.H{"type": "update", "lines": lines}) + c.Writer.Flush() + + // 如果任务完成，发送结束事件 + if isTerminalStatus(status) { + c.SSEvent("message", gin.H{"type": "finished"}) + c.Writer.Flush() + return + } + + case <-heartbeatTicker.C: + c.SSEvent("heartbeat", nil) + c.Writer.Flush() + } + } +} + +// streamWayneTask Wayne 任务使用轮询模式 +func (h *TaskLogHandler) streamWayneTask(c *gin.Context, taskID string, ctx context.Context, claims *auth.Claims) { + // 发送初始日志 + initialLines, taskStatus, err := h.fetchTaskLogLines(ctx, taskID, claims.UserID, claims.IsAdmin) + if err != nil { + c.SSEvent("message", gin.H{"type": "error", "error": err.Error()}) + c.Writer.Flush() return } @@ -392,14 +517,12 @@ func (h *TaskLogHandler) Stream(c *gin.Context) { } // 轮询循环 - ticker := time.NewTicker(2 * time.Second) + ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() heartbeatTicker := time.NewTicker(15 * time.Second) defer heartbeatTicker.Stop() - lastLineCount := len(initialLines) - for { select { case <-ctx.Done(): @@ -407,19 +530,14 @@ func (h *TaskLogHandler) Stream(c *gin.Context) { case <-ticker.C: lines, status, err := h.fetchTaskLogLines(ctx, taskID, claims.UserID, claims.IsAdmin) if err != nil { - // 发送错误但不关闭连接 c.SSEvent("message", gin.H{"type": "error", "error": err.Error()}) c.Writer.Flush() continue } - // 发送增量日志 - if len(lines) > lastLineCount { - newLines := lines[lastLineCount:] - c.SSEvent("message", gin.H{"type": "update", "lines": newLines}) - c.Writer.Flush() - lastLineCount = len(lines) - } + // 发送完整日志 + c.SSEvent("message", gin.H{"type": "update", "lines": lines}) + c.Writer.Flush() // 如果任务完成，发送结束事件 if isTerminalStatus(status) { @@ -429,13 +547,29 @@ func (h *TaskLogHandler) Stream(c *gin.Context) { } case <-heartbeatTicker.C: - // 发送心跳保持连接 c.SSEvent("heartbeat", nil) c.Writer.Flush() } } } +// parseActualTaskID 从 task log ID 解析出实际的任务 ID +func (h *TaskLogHandler) parseActualTaskID(taskID string) string { + switch { + case strings.HasPrefix(taskID, "awx:"): + return strings.TrimPrefix(taskID, "awx:") + case strings.HasPrefix(taskID, "wayne:publish:"): + // Wayne 任务需要使用 resourceID 作为订阅 key + parts := strings.Split(taskID, ":") + if len(parts) == 4 { + return "wayne:" + parts[2] + } + return "" + default: + return "" + } +} + // fetchTaskLogLines 获取任务的日志行和状态。 func (h *TaskLogHandler) fetchTaskLogLines(ctx context.Context, taskID string, userID uint64, isAdmin bool) ([]taskLogLine, string, error) { switch { @@ -513,7 +647,8 @@ func (h *TaskLogHandler) fetchWayneLogLines(ctx context.Context, id string, user lines = append(lines, taskLogLine{Time: formatTaskLogTime(history.CreatedAt), Message: "[message] " + history.Message, Class: classForWaynePublishStatus(history.Status)}) } - statusText := strconv.Itoa(history.Status) + // 将 Wayne 数字状态映射为与 AWX 一致的字符串状态 + statusText := wayneStatusToString(history.Status) return lines, statusText, nil } diff --git a/server/internal/service/delivery.go b/server/internal/service/delivery.go index 50f34af..94e3f9f 100644 --- a/server/internal/service/delivery.go +++ b/server/internal/service/delivery.go @@ -332,6 +332,12 @@ func allocatePort(requested int, used []int) (int, error) { return 0, fmt.Errorf("mysql port pool %d-%d is exhausted on the target host", mysqlPortPoolStart, mysqlPortPoolEnd) } +// stdoutCacheItem 缓存 AWX Job stdout 的结果 +type stdoutCacheItem struct { + stdout string + createdAt time.Time +} + type DeliveryService struct { db *gorm.DB cfg config.Config @@ -340,12 +346,21 @@ type DeliveryService struct { executionMu sync.Mutex streamMu sync.Mutex streams map[string]map[chan DeliveryTaskSnapshot]struct{} + stdoutCache map[string]*stdoutCacheItem + cacheMu sync.RWMutex } func (s *DeliveryService) DB() *gorm.DB { return s.db } func NewDeliveryService(cfg config.Config, db *gorm.DB, audit *AuditService) *DeliveryService { - return &DeliveryService{db: db, cfg: cfg, awx: NewAWXClient(cfg.AWXBaseURL, cfg.AWXToken, cfg.AWXUsername, cfg.AWXPassword), audit: audit, streams: make(map[string]map[chan DeliveryTaskSnapshot]struct{})} + return &DeliveryService{ + db: db, + cfg: cfg, + awx: NewAWXClient(cfg.AWXBaseURL, cfg.AWXToken, cfg.AWXUsername, cfg.AWXPassword), + audit: audit, + streams: make(map[string]map[chan DeliveryTaskSnapshot]struct{}), + stdoutCache: make(map[string]*stdoutCacheItem), + } } func (s *DeliveryService) ListTargets(ctx context.Context, component string) ([]DeliveryTarget, error) { @@ -942,18 +957,95 @@ func (s *DeliveryService) broadcastTask(ctx context.Context, taskID string) { if err != nil { return } + + // 任务状态变化时清除 stdout 缓存 + s.invalidateStdoutCacheForTask(taskID) + s.streamMu.Lock() defer s.streamMu.Unlock() + + // 收集需要清理的 closed channel + var closedChannels []chan DeliveryTaskSnapshot + for ch := range s.streams[taskID] { - select { - case ch <- snapshot: - default: - } + // 使用 recover 捕获 send on closed channel 的错误 + func() { + defer func() { + if r := recover(); r != nil { + // channel 已关闭，标记需要清理 + closedChannels = append(closedChannels, ch) + } + }() + select { + case ch <- snapshot: + default: + } + }() + } + + // 清理 closed channels + for _, ch := range closedChannels { + delete(s.streams[taskID], ch) + } + if len(s.streams[taskID]) == 0 { + delete(s.streams, taskID) } } +// invalidateStdoutCacheForTask 清除与任务相关的 stdout 缓存 +func (s *DeliveryService) invalidateStdoutCacheForTask(taskID string) { + var execution model.ExecutionJob + var rollback model.RollbackJob + + s.cacheMu.Lock() + defer s.cacheMu.Unlock() + + // 清除 execution job 的缓存 + if err := s.db.Where("task_id = ?", taskID).First(&execution).Error; err == nil && execution.ExecutorJobID != "" { + delete(s.stdoutCache, execution.ExecutorJobID) + } + + // 清除 rollback job 的缓存 + if err := s.db.Where("task_id = ?", taskID).First(&rollback).Error; err == nil && rollback.ExecutorJobID != "" { + delete(s.stdoutCache, rollback.ExecutorJobID) + } +} + +const stdoutCacheTTL = 30 * time.Second + func (s *DeliveryService) AWXJobStdout(ctx context.Context, jobID string) (string, error) { - return s.awx.JobStdout(ctx, jobID) + // 检查缓存 + s.cacheMu.RLock() + if item, ok := s.stdoutCache[jobID]; ok { + if time.Since(item.createdAt) < stdoutCacheTTL { + s.cacheMu.RUnlock() + return item.stdout, nil + } + } + s.cacheMu.RUnlock() + + // 缓存未命中或已过期，重新获取 + stdout, err := s.awx.JobStdout(ctx, jobID) + if err != nil { + return "", err + } + + // 更新缓存 + s.cacheMu.Lock() + s.stdoutCache[jobID] = &stdoutCacheItem{ + stdout: stdout, + createdAt: time.Now(), + } + s.cacheMu.Unlock() + + return stdout, nil +} + +// InvalidateStdoutCache 清除指定 jobID 的 stdout 缓存 +func (s *DeliveryService) InvalidateStdoutCache(jobID string) { + s.cacheMu.Lock() + delete(s.stdoutCache, jobID) + s.cacheMu.Unlock() } func (s *DeliveryService) Cancel(ctx context.Context, taskID string, userID uint64, isAdmin bool) error {