feat(task-logs): 实现任务日志 SSE 实时推送

实现任务日志 SSE 实时推送功能，解决前端无法流式输出日志的问题变更内容： - 后端：新增 /task-logs/stream SSE 端点，支持实时日志推送 - 后端：实现 2 秒轮询间隔检测新日志，支持心跳机制 - 后端：支持通过 query 参数 access_token 传递 JWT 认证 - 前端：新增 createTaskLogStream 函数封装 EventSource - 前端：修改 TaskCenter.vue 集成 SSE，支持实时更新和自动滚动 - 前端：添加流式连接状态指示器关联 commit： - 实现后端 SSE Stream 方法 - 注册 SSE 路由 - 实现前端 SSE 客户端 - 集成 SSE 到任务中心页面
2026-07-28 17:06:44 +08:00
parent 7d6efeb3dd
commit 8904330632
4 changed files with 309 additions and 2 deletions
@@ -79,3 +79,49 @@ function parseResponseBody(text: string) {
    return { error: text }
  }
 }
+
+export interface TaskLogStreamCallbacks {
+  onInit?: (lines: TaskLogLine[]) => void
+  onUpdate?: (lines: TaskLogLine[]) => void
+  onFinished?: () => void
+  onError?: (error: string) => void
+}
+
+export function createTaskLogStream(taskId: string, callbacks: TaskLogStreamCallbacks): EventSource {
+  const token = getToken()
+  const url = `/auth/api/v1/task-logs/stream?task_id=${encodeURIComponent(taskId)}&access_token=${encodeURIComponent(token || '')}`
+
+  const es = new EventSource(url)
+
+  es.onmessage = (event) => {
+    try {
+      const data = JSON.parse(event.data)
+      switch (data.type) {
+        case 'init':
+          callbacks.onInit?.(Array.isArray(data.lines) ? data.lines : [])
+          break
+        case 'update':
+          callbacks.onUpdate?.(Array.isArray(data.lines) ? data.lines : [])
+          break
+        case 'finished':
+          callbacks.onFinished?.()
+          es.close()
+          break
+        case 'error':
+          callbacks.onError?.(data.error || 'Unknown error')
+          break
+      }
+    } catch {
+      // ignore parse errors for heartbeat etc.
+    }
+  }
+
+  es.onerror = () => {
+    // EventSource will auto-reconnect, but we report connection errors
+    if (es.readyState === EventSource.CLOSED) {
+      callbacks.onError?.('Connection closed')
+    }
+  }
+
+  return es
+}
@@ -52,6 +52,7 @@
          <h3>
            <span>任务日志</span>
            <span class="selected-name">{{ selectedTaskName }}</span>
+            <span v-if="isStreaming" class="streaming-indicator">● 实时更新中</span>
          </h3>
          <span class="meta">{{ selectedTaskMeta }}</span>
        </div>
@@ -72,8 +73,8 @@
 </template>

 <script setup lang="ts">
-import { computed, onMounted, ref, watch } from 'vue'
-import { taskLogApi, type TaskLogLine, type TaskLogSummary } from '@/api/taskLog'
+import { computed, onMounted, onUnmounted, ref, watch } from 'vue'
+import { taskLogApi, createTaskLogStream, type TaskLogLine, type TaskLogSummary } from '@/api/taskLog'
 import { useBusinessLineStore } from '@/stores/businessLine'
 import { useBusinessLineMockProfile } from '@/utils/businessLineMock'

@@ -87,6 +88,8 @@ const logs = ref<TaskLogLine[]>([])
 const loadingTasks = ref(false)
 const loadingLogs = ref(false)
 const lastLoadedAt = ref<Date | null>(null)
+const isStreaming = ref(false)
+let eventSource: EventSource | null = null

 const selectedTask = computed(() => tasks.value.find((task) => task.id === selectedTaskId.value))
 const selectedTaskName = computed(() => selectedTask.value?.name || '未选择')
@@ -122,6 +125,8 @@ async function loadLogs(taskId: string) {
    const data = await taskLogApi.get(taskId)
    logs.value = data.lines
    lastLoadedAt.value = new Date()
+    // 启动 SSE 流式更新
+    startStream(taskId)
  } catch (error) {
    logs.value = [{ time: formatTime(new Date()), message: error instanceof Error ? error.message : '任务日志加载失败', class: 'err' }]
  } finally {
@@ -129,10 +134,55 @@ async function loadLogs(taskId: string) {
  }
 }

+function startStream(taskId: string) {
+  stopStream() // 关闭之前的连接
+
+  eventSource = createTaskLogStream(taskId, {
+    onInit: (lines) => {
+      logs.value = lines
+      lastLoadedAt.value = new Date()
+      scrollToBottom()
+    },
+    onUpdate: (newLines) => {
+      logs.value = [...logs.value, ...newLines]
+      lastLoadedAt.value = new Date()
+      scrollToBottom()
+    },
+    onFinished: () => {
+      isStreaming.value = false
+    },
+    onError: (error) => {
+      console.error('SSE error:', error)
+      isStreaming.value = false
+    },
+  })
+
+  isStreaming.value = true
+}
+
+function stopStream() {
+  if (eventSource) {
+    eventSource.close()
+    eventSource = null
+    isStreaming.value = false
+  }
+}
+
+function scrollToBottom() {
+  // 延迟滚动，等待 DOM 更新
+  setTimeout(() => {
+    const logStream = document.querySelector('.log-stream')
+    if (logStream) {
+      logStream.scrollTop = logStream.scrollHeight
+    }
+  }, 50)
+}
+
 function selectTask(taskId: string) {
  if (selectedTaskId.value === taskId) {
    return
  }
+  stopStream() // 停止之前的流
  selectedTaskId.value = taskId
  void loadLogs(taskId)
 }
@@ -155,6 +205,10 @@ watch(
 onMounted(() => {
  void loadTasks()
 })
+
+onUnmounted(() => {
+  stopStream()
+})
 </script>

 <style scoped>
@@ -262,6 +316,17 @@ onMounted(() => {
  font-weight: 500;
 }

+.streaming-indicator {
+  font-size: 11px;
+  color: var(--accent);
+  animation: pulse 2s ease-in-out infinite;
+}
+
+@keyframes pulse {
+  0%, 100% { opacity: 1; }
+  50% { opacity: 0.5; }
+}
+
 .log-stream {
  max-height: calc(100vh - 210px);
  min-height: 520px;
@@ -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 ""
@@ -154,5 +154,6 @@ func registerAuthServerRoutes(r *gin.Engine, deps Dependencies) {
 		protected.POST("/delivery/tasks/:id/clouddm/retry", deliveryHandler.RetryCloudDMRegistration)
 		protected.GET("/task-logs", taskLogHandler.List)
 		protected.GET("/task-logs/:id", taskLogHandler.Get)
+		protected.GET("/task-logs/stream", taskLogHandler.Stream)
 	}
 }