Merge branch 'main' into feat/base-service-delivery

2026-07-28 18:03:04 +08:00
parent 93981b9aad 408aa3af0d
commit f28bc99401
8 changed files with 443 additions and 55 deletions
@@ -18,8 +18,6 @@ func AuthMiddleware(cfg config.Config) gin.HandlerFunc {
 		tokenValue := ""
 		if strings.HasPrefix(value, "Bearer ") {
 			tokenValue = strings.TrimPrefix(value, "Bearer ")
-		} else {
-			tokenValue = strings.TrimSpace(c.Query("access_token"))
 		}
 		if tokenValue == "" {
 			c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing bearer token"})
@@ -1,6 +1,7 @@
 package handler

 import (
+	"context"
 	"errors"
 	"net/http"
 	"strconv"
@@ -350,6 +351,200 @@ func classForOutputLine(line string) string {
 	}
 }

+// Stream SSE 端点，用于实时推送任务日志。
+func (h *TaskLogHandler) Stream(c *gin.Context) {
+	claims, ok := CurrentClaims(c)
+	if !ok {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "missing current user"})
+		return
+	}
+
+	taskID := c.Query("task_id")
+	if taskID == "" {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "missing task_id parameter"})
+		return
+	}
+
+	// 设置 SSE 响应头
+	c.Header("Content-Type", "text/event-stream")
+	c.Header("Cache-Control", "no-cache")
+	c.Header("Connection", "keep-alive")
+	c.Header("X-Accel-Buffering", "no")
+
+	ctx := c.Request.Context()
+
+	// 发送初始日志
+	initialLines, taskStatus, err := h.fetchTaskLogLines(ctx, taskID, claims.UserID, claims.IsAdmin)
+	if err != nil {
+		c.SSEvent("message", gin.H{"type": "error", "error": err.Error()})
+		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
+	}
+
+	// 轮询循环
+	ticker := time.NewTicker(2 * time.Second)
+	defer ticker.Stop()
+
+	heartbeatTicker := time.NewTicker(15 * time.Second)
+	defer heartbeatTicker.Stop()
+
+	lastLineCount := len(initialLines)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		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)
+			}
+
+			// 如果任务完成，发送结束事件
+			if isTerminalStatus(status) {
+				c.SSEvent("message", gin.H{"type": "finished"})
+				c.Writer.Flush()
+				return
+			}
+
+		case <-heartbeatTicker.C:
+			// 发送心跳保持连接
+			c.SSEvent("heartbeat", nil)
+			c.Writer.Flush()
+		}
+	}
+}
+
+// fetchTaskLogLines 获取任务的日志行和状态。
+func (h *TaskLogHandler) fetchTaskLogLines(ctx context.Context, taskID string, userID uint64, isAdmin bool) ([]taskLogLine, string, error) {
+	switch {
+	case strings.HasPrefix(taskID, "awx:"):
+		return h.fetchAWXLogLines(ctx, strings.TrimPrefix(taskID, "awx:"), userID, isAdmin)
+	case strings.HasPrefix(taskID, "wayne:publish:"):
+		return h.fetchWayneLogLines(ctx, taskID, userID, isAdmin)
+	default:
+		return nil, "", errors.New("unknown task log source")
+	}
+}
+
+// fetchAWXLogLines 获取 AWX 任务的日志行。
+func (h *TaskLogHandler) fetchAWXLogLines(ctx context.Context, taskID string, userID uint64, isAdmin bool) ([]taskLogLine, string, error) {
+	task, events, err := h.delivery.GetTask(ctx, taskID, userID, isAdmin)
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, "", errors.New("task log not found")
+	}
+	if err != nil {
+		return nil, "", err
+	}
+
+	lines := make([]taskLogLine, 0, len(events)+16)
+	for _, event := range events {
+		lines = append(lines, taskLogLine{Time: formatTaskLogTime(event.CreatedAt), Message: "[" + event.ToState + "] " + event.Message, Class: classForTaskStatus(event.ToState)})
+	}
+
+	var execution model.ExecutionJob
+	if err := h.db.WithContext(ctx).Where("task_id = ?", task.ID).First(&execution).Error; err == nil && execution.ExecutorJobID != "" && execution.ExecutorJobID != "pending" && !strings.HasPrefix(execution.ExecutorJobID, "pending-") {
+		stdout, stdoutErr := h.delivery.AWXJobStdout(ctx, execution.ExecutorJobID)
+		if stdoutErr != nil {
+			lines = append(lines, taskLogLine{Time: formatTaskLogTime(time.Now()), Message: "[awx] stdout fetch failed: " + stdoutErr.Error(), Class: "err"})
+		} else {
+			lines = append(lines, splitStdoutLines(stdout)...)
+		}
+	}
+
+	var rollback model.RollbackJob
+	if err := h.db.WithContext(ctx).Where("task_id = ?", task.ID).First(&rollback).Error; err == nil && rollback.ExecutorJobID != "" && rollback.ExecutorJobID != "pending" && !strings.HasPrefix(rollback.ExecutorJobID, "pending-") {
+		stdout, stdoutErr := h.delivery.AWXJobStdout(ctx, rollback.ExecutorJobID)
+		if stdoutErr != nil {
+			lines = append(lines, taskLogLine{Time: formatTaskLogTime(time.Now()), Message: "[rollback awx] stdout fetch failed: " + stdoutErr.Error(), Class: "err"})
+		} else {
+			lines = append(lines, splitStdoutLines(stdout)...)
+		}
+	}
+
+	return lines, task.Status, nil
+}
+
+// fetchWayneLogLines 获取 Wayne 任务的日志行。
+func (h *TaskLogHandler) fetchWayneLogLines(ctx context.Context, id string, userID uint64, isAdmin bool) ([]taskLogLine, string, error) {
+	resourceID, historyID, ok := parseWaynePublishTaskID(id)
+	if !ok {
+		return nil, "", errors.New("invalid Wayne task log id")
+	}
+
+	namespaces, err := h.visibleWayneNamespacesFromCtx(ctx, userID, isAdmin, 0)
+	if err != nil {
+		return nil, "", err
+	}
+
+	history, err := h.wayne.GetDeploymentHistory(ctx, namespaces, resourceID, historyID)
+	if err != nil {
+		return nil, "", err
+	}
+
+	lines := []taskLogLine{
+		{Time: formatTaskLogTime(history.CreatedAt), Message: "[wayne] publish history #" + strconv.FormatInt(history.ID, 10), Class: classForWaynePublishStatus(history.Status)},
+		{Time: formatTaskLogTime(history.CreatedAt), Message: "[deployment] " + history.ResourceName + " resource_id=" + strconv.FormatInt(history.ResourceID, 10), Class: ""},
+		{Time: formatTaskLogTime(history.CreatedAt), Message: "[cluster] " + history.Cluster + " template_id=" + strconv.FormatInt(history.TemplateID, 10), Class: ""},
+		{Time: formatTaskLogTime(history.CreatedAt), Message: "[user] " + history.User, Class: ""},
+	}
+	if strings.TrimSpace(history.Message) != "" {
+		lines = append(lines, taskLogLine{Time: formatTaskLogTime(history.CreatedAt), Message: "[message] " + history.Message, Class: classForWaynePublishStatus(history.Status)})
+	}
+
+	statusText := strconv.Itoa(history.Status)
+	return lines, statusText, nil
+}
+
+// visibleWayneNamespacesFromCtx 从 context 获取可见的 Wayne 命名空间。
+func (h *TaskLogHandler) visibleWayneNamespacesFromCtx(ctx context.Context, userID uint64, isAdmin bool, businessLineID uint64) ([]model.BusinessLineWayneNamespace, error) {
+	query := h.db.WithContext(ctx).Order("business_line_id ASC, wayne_namespace_id ASC")
+	if businessLineID != 0 {
+		query = query.Where("business_line_id = ?", businessLineID)
+	}
+	if !isAdmin {
+		query = query.Where("business_line_id IN (?)", h.db.Model(&model.BusinessLineUser{}).Select("business_line_id").Where("user_id = ?", userID))
+	}
+	var namespaces []model.BusinessLineWayneNamespace
+	if err := query.Find(&namespaces).Error; err != nil {
+		return nil, err
+	}
+	return namespaces, nil
+}
+
+// isTerminalStatus 判断任务状态是否为终态。
+func isTerminalStatus(status string) bool {
+	switch status {
+	case model.TaskFinished, model.TaskCanceled, model.TaskExecutionFailed,
+		model.TaskValidationFailed, model.TaskRegisterFailed,
+		model.TaskRolledBack, model.TaskRollbackFailed, model.TaskRollbackAck:
+		return true
+	default:
+		return false
+	}
+}
+
 func formatTaskLogTime(t time.Time) string {
 	if t.IsZero() {
 		return ""