Merge branch 'main' into feat/task-logs-virtual-scroll
This commit is contained in:
Wonder
@@ -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>
|
||||
<div class="panel-actions">
|
||||
<el-switch
|
||||
@@ -99,8 +100,8 @@
|
||||
</template>
|
||||
|
||||
<script setup lang="ts">
|
||||
import { computed, nextTick, onMounted, ref, watch } from 'vue'
|
||||
import { taskLogApi, type TaskLogLine, type TaskLogSummary } from '@/api/taskLog'
|
||||
import { computed, nextTick, 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'
|
||||
import { useVirtualizer } from '@tanstack/vue-virtual'
|
||||
@@ -115,6 +116,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 autoScroll = ref(true)
|
||||
const logContainerRef = ref<HTMLDivElement | null>(null)
|
||||
|
||||
@@ -160,7 +163,8 @@ async function loadLogs(taskId: string) {
|
||||
const data = await taskLogApi.get(taskId)
|
||||
logs.value = data.lines
|
||||
lastLoadedAt.value = new Date()
|
||||
scrollToBottom()
|
||||
// 启动 SSE 流式更新
|
||||
startStream(taskId)
|
||||
} catch (error) {
|
||||
logs.value = [{ time: formatTime(new Date()), message: error instanceof Error ? error.message : '任务日志加载失败', class: 'err' }]
|
||||
} finally {
|
||||
@@ -168,10 +172,70 @@ 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() {
|
||||
if (!autoScroll.value) return
|
||||
|
||||
nextTick(() => {
|
||||
if (logContainerRef.value) {
|
||||
logContainerRef.value.scrollTop = logContainerRef.value.scrollHeight
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
function handleScroll() {
|
||||
if (!logContainerRef.value) return
|
||||
|
||||
const { scrollTop, scrollHeight, clientHeight } = logContainerRef.value
|
||||
const distanceFromBottom = scrollHeight - scrollTop - clientHeight
|
||||
|
||||
// 距离底部小于 50px 时认为在底部，恢复自动滚动
|
||||
if (distanceFromBottom < 50) {
|
||||
autoScroll.value = true
|
||||
} else if (distanceFromBottom > 100) {
|
||||
// 用户向上滚动超过 100px 时暂停自动滚动
|
||||
autoScroll.value = false
|
||||
}
|
||||
}
|
||||
|
||||
function selectTask(taskId: string) {
|
||||
if (selectedTaskId.value === taskId) {
|
||||
return
|
||||
}
|
||||
stopStream() // 停止之前的流
|
||||
selectedTaskId.value = taskId
|
||||
void loadLogs(taskId)
|
||||
}
|
||||
@@ -219,6 +283,10 @@ watch(
|
||||
onMounted(() => {
|
||||
void loadTasks()
|
||||
})
|
||||
|
||||
onUnmounted(() => {
|
||||
stopStream()
|
||||
})
|
||||
</script>
|
||||
|
||||
<style scoped>
|
||||
@@ -312,6 +380,7 @@ onMounted(() => {
|
||||
|
||||
.task-log-panel .panel-head {
|
||||
gap: 14px;
|
||||
flex-wrap: wrap;
|
||||
}
|
||||
|
||||
.task-log-panel .panel-head h3 {
|
||||
@@ -321,11 +390,26 @@ onMounted(() => {
|
||||
gap: 10px;
|
||||
}
|
||||
|
||||
.task-log-panel .panel-head .panel-actions {
|
||||
margin-left: auto;
|
||||
}
|
||||
|
||||
.selected-name {
|
||||
color: var(--text-dim);
|
||||
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; }
|
||||
}
|
||||
|
||||
.panel-actions {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
@@ -336,14 +420,6 @@ onMounted(() => {
|
||||
--el-switch-on-color: var(--accent);
|
||||
}
|
||||
|
||||
.task-log-panel .panel-head {
|
||||
flex-wrap: wrap;
|
||||
}
|
||||
|
||||
.task-log-panel .panel-head .panel-actions {
|
||||
margin-left: auto;
|
||||
}
|
||||
|
||||
.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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user