From e5fc716b8fa16577a4c38042e4a036707510b172 Mon Sep 17 00:00:00 2001 From: duorameng <2997944583@qq.com> Date: Mon, 29 Jun 2026 22:00:09 +0800 Subject: [PATCH] refactor: migrate real-time log viewing from WebSocket to SSE with JSON wrapping --- internal/controllers/log_sse_controller.go | 99 +++++++++++++++++ internal/controllers/log_ws_controller.go | 122 --------------------- internal/router/api_routes.go | 2 +- internal/router/register.go | 2 +- internal/router/router.go | 2 +- web/src/views/history/History.vue | 56 +++++----- web/src/views/tasks/Tasks.vue | 35 +++--- 7 files changed, 147 insertions(+), 171 deletions(-) create mode 100644 internal/controllers/log_sse_controller.go delete mode 100644 internal/controllers/log_ws_controller.go diff --git a/internal/controllers/log_sse_controller.go b/internal/controllers/log_sse_controller.go new file mode 100644 index 0000000..56a43f0 --- /dev/null +++ b/internal/controllers/log_sse_controller.go @@ -0,0 +1,99 @@ +package controllers + +import ( + "fmt" + "io" + + "github.com/engigu/baihu-panel/internal/database" + "github.com/engigu/baihu-panel/internal/models" + "github.com/engigu/baihu-panel/internal/services/tasks" + "github.com/engigu/baihu-panel/internal/utils" + + "github.com/gin-gonic/gin" +) + +type LogSSEController struct{} + +func NewLogSSEController() *LogSSEController { + return &LogSSEController{} +} + +func (lc *LogSSEController) StreamLog(c *gin.Context) { + logIDStr := c.Query("log_id") + if logIDStr == "" { + c.JSON(400, gin.H{"error": "log_id is required"}) + return + } + + logID := logIDStr + + c.Header("Content-Type", "text/event-stream") + c.Header("Cache-Control", "no-cache") + c.Header("Connection", "keep-alive") + c.Header("Transfer-Encoding", "chunked") + // c.Header("Access-Control-Allow-Origin", "*") + + // 1. 检查数据库中是否已结束 + var taskLog models.TaskLog + res := database.DB.Where("id = ?", logID).Limit(1).Find(&taskLog) + if res.Error == nil && res.RowsAffected > 0 { + if taskLog.Status != "running" { + // 已结束，读取库内日志 + content, err := utils.DecompressFromBase64(string(taskLog.Output)) + if err != nil { + c.SSEvent("message", gin.H{"text": "解压日志失败: " + err.Error()}) + c.Writer.Flush() + return + } + c.SSEvent("message", gin.H{"text": content}) + c.Writer.Flush() + return + } + } + + // 2. 未结束或未找到记录，尝试从 TinyLogManager 获取 + tl := tasks.GetActiveLog(logID) + if tl == nil { + c.SSEvent("message", gin.H{"text": "未找到正在运行的任务日志"}) + c.Writer.Flush() + return + } + + // 发送系统提示 + c.SSEvent("message", gin.H{"text": fmt.Sprintf("[System] 连接成功，正在监听日志... (LogID: %s)\n", logID)}) + c.Writer.Flush() + + // 发送最后 100 行 + lastLines, err := tl.ReadLastLines(100) + if err == nil && len(lastLines) > 0 { + c.SSEvent("message", gin.H{"text": string(lastLines)}) + c.Writer.Flush() + } + + // 订阅实时更新 + sub := tl.Subscribe() + defer tl.Unsubscribe(sub) + + // 推送更新 + c.Stream(func(w io.Writer) bool { + select { + case data, ok := <-sub: + if !ok { + // 任务结束，尝试刷新最后一次库内完整内容 + var finalLog models.TaskLog + res := database.DB.Where("id = ?", logID).Limit(1).Find(&finalLog) + if res.Error == nil && res.RowsAffected > 0 { + content, _ := utils.DecompressFromBase64(string(finalLog.Output)) + if content != "" { + c.SSEvent("message", gin.H{"text": "\n--- 任务已结束 ---\n"}) + } + } + return false + } + c.SSEvent("message", gin.H{"text": string(data)}) + return true + case <-c.Request.Context().Done(): + return false + } + }) +} diff --git a/internal/controllers/log_ws_controller.go b/internal/controllers/log_ws_controller.go deleted file mode 100644 index 33e8195..0000000 --- a/internal/controllers/log_ws_controller.go +++ /dev/null @@ -1,122 +0,0 @@ -package controllers - -import ( - "fmt" - - "github.com/engigu/baihu-panel/internal/constant" - "github.com/engigu/baihu-panel/internal/database" - "github.com/engigu/baihu-panel/internal/models" - "github.com/engigu/baihu-panel/internal/services/tasks" - "github.com/engigu/baihu-panel/internal/utils" - - "github.com/gin-gonic/gin" - "github.com/gorilla/websocket" - "time" -) - -type LogWSController struct{} - -func NewLogWSController() *LogWSController { - return &LogWSController{} -} - -func (lc *LogWSController) StreamLog(c *gin.Context) { - logIDStr := c.Query("log_id") - if logIDStr == "" { - return - } - - logID := logIDStr - - conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) - if err != nil { - return - } - defer conn.Close() - - // DoS 保护与心跳设置 - conn.SetReadLimit(constant.MaxMessageSize) // 使用与终端一致的限制 - conn.SetReadDeadline(time.Now().Add(constant.PongWait)) - conn.SetPongHandler(func(string) error { - conn.SetReadDeadline(time.Now().Add(constant.PongWait)) - return nil - }) - - // 启动一个读取循环，用于处理 pong 和检测断开 - go func() { - for { - if _, _, err := conn.ReadMessage(); err != nil { - break - } - } - }() - - // 1. 检查数据库中是否已结束 - var taskLog models.TaskLog - res := database.DB.Where("id = ?", logID).Limit(1).Find(&taskLog) - if res.Error == nil && res.RowsAffected > 0 { - if taskLog.Status != "running" { - // 已结束，读取库内日志 - content, err := utils.DecompressFromBase64(string(taskLog.Output)) - if err != nil { - conn.WriteMessage(websocket.TextMessage, []byte("解压日志失败: "+err.Error())) - return - } - conn.WriteMessage(websocket.TextMessage, []byte(content)) - return - } - } - - // 2. 未结束或未找到记录，尝试从 TinyLogManager 获取 - tl := tasks.GetActiveLog(logID) - if tl == nil { - conn.WriteMessage(websocket.TextMessage, []byte("未找到正在运行的任务日志")) - return - } - - // 发送系统提示 - conn.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf("[System] 连接成功，正在监听日志... (LogID: %s)\n", logID))) - - // 发送最后 100 行 - lastLines, err := tl.ReadLastLines(100) - if err == nil && len(lastLines) > 0 { - conn.WriteMessage(websocket.TextMessage, lastLines) - } - - // 订阅实时更新 - sub := tl.Subscribe() - defer tl.Unsubscribe(sub) - - ticker := time.NewTicker(constant.PingPeriod) - defer ticker.Stop() - - // 推送更新 - for { - select { - case <-ticker.C: - conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) - if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil { - return - } - case data, ok := <-sub: - if !ok { - // 任务结束，尝试刷新最后一次库内完整内容 - var finalLog models.TaskLog - res := database.DB.Where("id = ?", logID).Limit(1).Find(&finalLog) - if res.Error == nil && res.RowsAffected > 0 { - content, _ := utils.DecompressFromBase64(string(finalLog.Output)) - if content != "" { - conn.WriteMessage(websocket.TextMessage, []byte("\n--- 任务已结束 ---\n")) - // 这里可以选择性再推一次完整版，或直接退出 - } - } - return - } - if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { - return - } - case <-c.Request.Context().Done(): - return - } - } -} diff --git a/internal/router/api_routes.go b/internal/router/api_routes.go index 001e308..f44943d 100644 --- a/internal/router/api_routes.go +++ b/internal/router/api_routes.go @@ -161,7 +161,7 @@ func registerLogRoutes(g *gin.RouterGroup, c *Controllers) { { logs.GET("", c.Log.GetLogs) logs.POST("/clear", c.Log.ClearLogs) - logs.GET("/ws", c.LogWS.StreamLog) + logs.GET("/sse", c.LogSSE.StreamLog) logs.GET("/:id", c.Log.GetLogDetail) logs.DELETE("/:id", c.Log.DeleteLog) } diff --git a/internal/router/register.go b/internal/router/register.go index a30ae1c..7d3dbee 100644 --- a/internal/router/register.go +++ b/internal/router/register.go @@ -58,7 +58,7 @@ func RegisterControllers() *Controllers { File: controllers.NewFileController(constant.ScriptsWorkDir), Dashboard: controllers.NewDashboardController(executorService), Log: controllers.NewLogController(), - LogWS: controllers.NewLogWSController(), + LogSSE: controllers.NewLogSSEController(), Terminal: controllers.NewTerminalController(envService), Settings: controllers.NewSettingsController(userService, loginLogService, executorService), Dependency: controllers.NewDependencyController(), diff --git a/internal/router/router.go b/internal/router/router.go index 9116dff..3e40837 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -21,7 +21,7 @@ type Controllers struct { File *controllers.FileController Dashboard *controllers.DashboardController Log *controllers.LogController - LogWS *controllers.LogWSController + LogSSE *controllers.LogSSEController Terminal *controllers.TerminalController Settings *controllers.SettingsController Dependency *controllers.DependencyController diff --git a/web/src/views/history/History.vue b/web/src/views/history/History.vue index cc7535a..d932c0a 100644 --- a/web/src/views/history/History.vue +++ b/web/src/views/history/History.vue @@ -80,7 +80,7 @@ const deleteLogId = ref(null) const wsContent = ref('') const isWsLoading = ref(false) -let logSocket: WebSocket | null = null +let logSource: EventSource | null = null import { decompressFromBase64 } from '@/utils/decompress' @@ -136,12 +136,9 @@ function handlePageChange(page: number) { } async function selectLog(log: TaskLog) { - if (logSocket) { - logSocket.onopen = null - logSocket.onmessage = null - logSocket.onerror = null - logSocket.onclose = null - logSocket.close() + if (logSource) { + logSource.close() + logSource = null } // 清理旧定时器 @@ -198,23 +195,28 @@ async function selectLog(log: TaskLog) { } return } - - const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:' + + const protocol = window.location.protocol const host = window.location.host const baseUrl = (window as any).__BASE_URL__ || '' const apiVersion = (window as any).__API_VERSION__ || '/api/v1' - const wsUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/ws?log_id=${log.id}` + const sseUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/sse?log_id=${log.id}` - logSocket = new WebSocket(wsUrl) + logSource = new EventSource(sseUrl) - logSocket.onopen = () => { + logSource.onopen = () => { isWsLoading.value = false - console.log('[LogWS] Connection opened') + console.log('[LogSSE] Connection opened') } - logSocket.onmessage = (event) => { + logSource.onmessage = (event) => { isWsLoading.value = false - wsContent.value += event.data + try { + const data = JSON.parse(event.data) + wsContent.value += data.text || '' + } catch { + wsContent.value += event.data + } // 自动滚动到底部 nextTick(() => { const pre = document.querySelector('.log-pre') @@ -222,15 +224,13 @@ async function selectLog(log: TaskLog) { }) } - logSocket.onerror = (e) => { + logSource.onerror = (e) => { isWsLoading.value = false - console.error('[LogWS] Connection error', e) - toast.error('日志连接异常') - } - - logSocket.onclose = (e) => { - isWsLoading.value = false - console.log('[LogWS] Connection closed', e.code, e.reason) + console.error('[LogSSE] Connection error/closed', e) + if (logSource) { + logSource.close() + logSource = null + } } } @@ -239,13 +239,9 @@ function closeDetail() { clearInterval(durationTimer) durationTimer = null } - if (logSocket) { - logSocket.onopen = null - logSocket.onmessage = null - logSocket.onerror = null - logSocket.onclose = null - logSocket.close() - logSocket = null + if (logSource) { + logSource.close() + logSource = null } selectedLog.value = null wsContent.value = '' diff --git a/web/src/views/tasks/Tasks.vue b/web/src/views/tasks/Tasks.vue index 866de81..5a11489 100644 --- a/web/src/views/tasks/Tasks.vue +++ b/web/src/views/tasks/Tasks.vue @@ -295,19 +295,13 @@ const selectedLog = ref(null) const logContent = ref('') const logEmptyTitle = ref(undefined) const logEmptyDesc = ref(undefined) -let logSocket: WebSocket | null = null +let logSource: EventSource | null = null let durationTimer: ReturnType | null = null function cleanupLogSocket() { - if (logSocket) { - logSocket.onopen = null - logSocket.onmessage = null - logSocket.onerror = null - logSocket.onclose = null - if (logSocket.readyState === WebSocket.CONNECTING || logSocket.readyState === WebSocket.OPEN) { - logSocket.close() - } - logSocket = null + if (logSource) { + logSource.close() + logSource = null } } @@ -360,17 +354,26 @@ async function viewLogs(taskId: string) { return } - // Connect WebSocket to load log content for running tasks + // Connect SSE to load log content for running tasks cleanupLogSocket() - const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:' + const protocol = window.location.protocol const host = window.location.host const baseUrl = (window as any).__BASE_URL__ || '' const apiVersion = (window as any).__API_VERSION__ || '/api/v1' - const wsUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/ws?log_id=${latestLog.id}` + const sseUrl = `${protocol}//${host}${baseUrl}${apiVersion}/logs/sse?log_id=${latestLog.id}` - logSocket = new WebSocket(wsUrl) - logSocket.onmessage = (event) => { - logContent.value += event.data + logSource = new EventSource(sseUrl) + logSource.onmessage = (event) => { + try { + const data = JSON.parse(event.data) + logContent.value += data.text || '' + } catch { + logContent.value += event.data + } + } + logSource.onerror = (e) => { + console.error('[LogSSE] Connection error/closed', e) + cleanupLogSocket() } // 启动状态轮询