fix(server): 修复 task log 接口多个问题

- 修复 Wayne 任务状态映射，将数字状态统一为字符串状态 - 添加分页支持，支持 page/page_size 参数 - 添加 AWXJobStdout 内存缓存，减少重复 API 调用 - 修复 SubscribeTask 内存泄漏，自动清理 closed channel - 统一 Stream 端点，AWX 使用 pub/sub，Wayne 使用轮询背景：task log 接口存在多个潜在问题，包括状态判断错误导致无限轮询、缺少分页、重复 API 调用性能问题、内存泄漏风险等关联 commit：fix/logs 分支
2026-07-29 14:00:40 +08:00
parent eaba4d237a
commit 61eeb0c8ac
2 changed files with 250 additions and 23 deletions
@@ -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
 }

@@ -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 {