From 0c0ff37c696d07270f8d9c76a3e6ed3744c79a45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Mathieu?= Date: Mon, 13 Jan 2025 13:22:43 +0100 Subject: [PATCH] fix(core, ui): send a "start" event to be sure the UI receive the SSE The UI only store a reference to the logs SSE when receive the first event. In case a flow didn't emit any log, or the logs tab is closed before any logs is emitted, the UI will not have any reference to the SSE so the SSE connection would stay alive forever. Each SSE connection starts a thread via the logs queue, creating a thread leak. Sending a first "start" event makes sure the UI has a reference to the SSE. --- .../java/io/kestra/core/services/ExecutionLogService.java | 3 +++ ui/src/components/logs/TaskRunDetails.vue | 5 ++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/io/kestra/core/services/ExecutionLogService.java b/core/src/main/java/io/kestra/core/services/ExecutionLogService.java index f91297a340d..194813e1308 100644 --- a/core/src/main/java/io/kestra/core/services/ExecutionLogService.java +++ b/core/src/main/java/io/kestra/core/services/ExecutionLogService.java @@ -40,6 +40,9 @@ public Flux> streamExecutionLogs(final String tenantId, final AtomicReference disposable = new AtomicReference<>(); return Flux.>create(emitter -> { + // send a first "empty" event so the SSE is correctly initialized in the frontend in case there are no logs + emitter.next(Event.of(LogEntry.builder().build()).id("start")); + // fetch repository first getExecutionLogs(tenantId, executionId, minLevel, List.of(), withAccessControl) .forEach(logEntry -> emitter.next(Event.of(logEntry).id("progress"))); diff --git a/ui/src/components/logs/TaskRunDetails.vue b/ui/src/components/logs/TaskRunDetails.vue index 3350beae478..1c7079a67f2 100644 --- a/ui/src/components/logs/TaskRunDetails.vue +++ b/ui/src/components/logs/TaskRunDetails.vue @@ -502,7 +502,10 @@ this.logsSSE = sse; this.logsSSE.onmessage = event => { - this.logsBuffer = this.logsBuffer.concat(JSON.parse(event.data)); + // we are receiving a first "fake" event to force initializing the connection: ignoring it + if (event.lastEventId !== "start") { + this.logsBuffer = this.logsBuffer.concat(JSON.parse(event.data)); + } clearTimeout(this.timeout); this.timeout = setTimeout(() => {