From 687ca21227b8b88823a4e2db81b5744debcaaea0 Mon Sep 17 00:00:00 2001 From: guoyuqi Date: Tue, 13 Oct 2026 14:40:33 +0800 Subject: [PATCH] fix: container log text --- agent/app/service/container.go | 44 +++++++------- .../src/components/log/container/index.vue | 58 +++++++++---------- 2 files changed, 48 insertions(+), 54 deletions(-) diff --git a/agent/app/service/container.go b/agent/app/service/container.go index 0e63fad33..e015076da 100644 --- a/agent/app/service/container.go +++ b/agent/app/service/container.go @@ -810,33 +810,33 @@ func (u *ContainerService) ContainerLogClean(req dto.OperationWithName) error { func (u *ContainerService) StreamLogs(ctx *gin.Context, params dto.StreamLog) { messageChan := make(chan string, 1024) errorChan := make(chan error, 1) - doneChan := make(chan struct{}) - - go func() { - <-ctx.Request.Context().Done() - close(doneChan) - }() + logCtx, cancel := context.WithCancel(ctx.Request.Context()) + defer cancel() + doneChan := logCtx.Done() go collectLogs(doneChan, params, messageChan, errorChan) ctx.Stream(func(w io.Writer) bool { - select { - case msg, ok := <-messageChan: - if !ok { - return false - } - _, err := fmt.Fprintf(w, "data: %s\n\n", msg) - if err != nil { - return false - } - return true - case err := <-errorChan: - if err != nil { - _, _ = fmt.Fprintf(w, "event: error\ndata: %v\n\n", err.Error()) - } - return false - } + return writeContainerLogMessage(w, doneChan, messageChan, errorChan) }) } +func writeContainerLogMessage(w io.Writer, done <-chan struct{}, messages <-chan string, errors <-chan error) bool { + select { + case <-done: + return false + case message, ok := <-messages: + if ok { + _, err := fmt.Fprintf(w, "data: %s\n\n", message) + return err == nil + } + // The collector closes both channels when it finishes. Drain all buffered + // logs before handling its result, or a ready error channel can drop the tail. + if err := <-errors; err != nil { + _, _ = fmt.Fprintf(w, "event: error\ndata: %v\n\n", err) + } + return false + } +} + func collectLogs(done <-chan struct{}, params dto.StreamLog, messageChan chan<- string, errorChan chan<- error) { defer close(messageChan) defer close(errorChan) diff --git a/frontend/src/components/log/container/index.vue b/frontend/src/components/log/container/index.vue index d75ed437d..2b6db25af 100644 --- a/frontend/src/components/log/container/index.vue +++ b/frontend/src/components/log/container/index.vue @@ -126,14 +126,14 @@ const styleVars = computed(() => ({ const terminalElement = ref(null); let eventSource: EventSource | null = null; +let searchVersion = 0; let term: Terminal | null = null; const fitAddon = new FitAddon(); -let onScrollDisposable: { dispose: () => void } | null = null; const MAX_VIEW_LINES = 20000; -const followBottom = ref(true); +let followBottom = true; const logSearch = reactive({ - isWatch: props.defaultFollow ? true : true, + isWatch: true, isShowTimestamp: props.defaultIsShowTimestamp, container: '', mode: 'all', @@ -168,6 +168,7 @@ const timeOptions = ref([ ]); const stopListening = () => { + searchVersion++; if (eventSource) { eventSource.close(); eventSource = null; @@ -175,26 +176,13 @@ const stopListening = () => { }; const clearTerminal = () => { - term?.reset(); - followBottom.value = true; + // Queue the reset after pending writes so old logs cannot appear in a new search. + term?.write('\x1bc'); + followBottom = true; }; const writeLogLine = (data: string) => { - if (!term) return; - term.writeln(data); - if (followBottom.value) { - term.scrollToBottom(); - } -}; - -const bindXTermEvents = () => { - if (!term) return; - onScrollDisposable?.dispose(); - onScrollDisposable = term.onScroll(() => { - if (!term) return; - const active = term.buffer.active; - followBottom.value = active.baseY + active.cursorY >= active.length - 2; - }); + term?.writeln(data); }; const showEventSourceAuthError = (message: string) => { @@ -235,11 +223,15 @@ const initTerminal = () => { term.open(terminalElement.value); term.loadAddon(fitAddon); fitAddon.fit(); - bindXTermEvents(); -}; - -const handleClose = async () => { - stopListening(); + // Terminal.dispose() also disposes these listeners. + term.onScroll(() => { + if (!term) return; + const active = term.buffer.active; + followBottom = active.viewportY >= active.baseY; + }); + term.onWriteParsed(() => { + if (followBottom) term?.scrollToBottom(); + }); }; const searchLogs = async () => { @@ -248,6 +240,7 @@ const searchLogs = async () => { return; } stopListening(); + const version = searchVersion; clearTerminal(); let currentNode = globalCurrentNode.value; @@ -261,15 +254,19 @@ const searchLogs = async () => { } const authError = await checkStreamAuth(url, currentNode); + if (version !== searchVersion) return; if (authError) { showEventSourceAuthError(authError); return; } - eventSource = new EventSource(url); - eventSource.onmessage = (event: MessageEvent) => { + const source = new EventSource(url); + eventSource = source; + source.onmessage = (event: MessageEvent) => { + if (version !== searchVersion) return; writeLogLine(event.data); }; - eventSource.onerror = (event: MessageEvent) => { + source.onerror = (event: MessageEvent) => { + if (version !== searchVersion) return; stopListening(); if (event.data && event.data != '') { MsgError(event.data); @@ -353,8 +350,6 @@ onMounted(() => { logSearch.resource = props.resource; logSearch.tail = 100; - logSearch.mode = 'all'; - logSearch.isWatch = true; nextTick(() => { initTerminal(); @@ -369,8 +364,7 @@ onMounted(() => { }); onUnmounted(() => { - handleClose(); - onScrollDisposable?.dispose(); + stopListening(); if (term) { term.dispose(); term = null;