feat(core): attach logs to dynamically-generated taskruns and nest them in the UI

brian-mulier-p · brian-mulier-p · commit d0a7fe087f69 · 2026-06-26T13:47:47.000+02:00
Plugins that synthesize dynamic taskruns via dynamicWorkerResult (e.g. the
Ansible and dbt plugins building one taskrun per unit of work) had no way to
attach logs to them, so all the output landed on the parent task's root
taskrun.

Add an overload dynamicWorkerResult(WorkerTaskResult, List&lt;DynamicTaskRunLog&gt;)
that registers the taskrun and its log lines in a single call. The lines are
emitted through the regular logging pipeline (a child logger bound to the
taskrun), so the standard appender behaviour - secret masking, long-message
splitting, level filtering, server-log forwarding - applies natively. The logs
ride with the taskrun being registered, so they can only target it:
execution/tenant/namespace/flow are taken from the run context and the attempt
is fixed to 0, so a plugin never builds (nor can forge) a LogEntry for another
execution or taskrun. The default registers the taskrun and drops the logs so
runtimes that don't support per-taskrun logs still show the taskrun.

Nest these taskruns under their parent in the execution views (they carry
parentTaskRunId): indent child rows by depth in Gantt and the Logs view, and
make the topology "Show task logs" action include a node's dynamic child
taskruns so their logs show alongside the task's own.
diff --git a/core/src/main/java/io/kestra/core/runners/DefaultRunContext.java b/core/src/main/java/io/kestra/core/runners/DefaultRunContext.java
@@ -506,6 +506,15 @@ public List<WorkerTaskResult> dynamicWorkerResults() {
         return dynamicWorkerTaskResult;
     }
 
+    @Override
+    public void dynamicWorkerResult(WorkerTaskResult workerTaskResult, List<DynamicTaskRunLog> logs) {
+        this.dynamicWorkerTaskResult.add(workerTaskResult);
+
+        if (logs != null && !logs.isEmpty() && workerTaskResult.getTaskRun() != null) {
+            this.logger.emitDynamicTaskRunLogs(workerTaskResult.getTaskRun(), logs);
+        }
+    }
+
     /**
      * {@inheritDoc}
      */
diff --git a/core/src/main/java/io/kestra/core/runners/DynamicTaskRunLog.java b/core/src/main/java/io/kestra/core/runners/DynamicTaskRunLog.java
@@ -0,0 +1,14 @@
+package io.kestra.core.runners;
+
+import org.slf4j.event.Level;
+
+/**
+ * A log line to attach to a dynamically-generated taskrun when it is registered through
+ * {@link RunContext#dynamicWorkerResult(WorkerTaskResult, java.util.List)}.
+ * <p>
+ * Callers only provide the level and message — the execution, tenant, namespace, flow, taskrun
+ * identity and attempt are filled in from the run context, and secrets are masked, so a plugin
+ * never builds (nor can tamper with) a full {@link io.kestra.core.models.executions.LogEntry}.
+ */
+public record DynamicTaskRunLog(Level level, String message) {
+}
diff --git a/core/src/main/java/io/kestra/core/runners/RunContext.java b/core/src/main/java/io/kestra/core/runners/RunContext.java
@@ -121,6 +121,22 @@ public abstract class RunContext implements PropertyContext {
 
     public abstract List<WorkerTaskResult> dynamicWorkerResults();
 
+    /**
+     * Registers a dynamically-generated taskrun together with the log lines to attach to it, in a
+     * single call — the way plugins surface logs under the sub-taskruns they create at runtime
+     * (these taskruns are synthesized after their underlying work has finished, so their logs are
+     * already available at registration time).
+     * <p>
+     * The logs ride with the taskrun being registered, so they can only ever target that taskrun;
+     * their execution, tenant, namespace, flow and attempt are taken from this context (a plugin
+     * cannot target another execution or tenant) and secrets are masked. The default registers the
+     * taskrun and drops the logs, so runtimes that don't support per-taskrun logs (and older ones)
+     * still show the taskrun rather than failing the task.
+     */
+    public void dynamicWorkerResult(WorkerTaskResult workerTaskResult, List<DynamicTaskRunLog> logs) {
+        this.dynamicWorkerResult(List.of(workerTaskResult));
+    }
+
     /**
      * Gets access to the working directory.
      *
diff --git a/core/src/main/java/io/kestra/core/runners/RunContextLogger.java b/core/src/main/java/io/kestra/core/runners/RunContextLogger.java
@@ -15,6 +15,7 @@
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.executions.LogEntry;
+import io.kestra.core.models.executions.TaskRun;
 
 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.Logger;
@@ -213,6 +214,59 @@ public void emitLogs(List<LogEntry> logEntries) {
         this.logEmitter.emits(logEntries);
     }
 
+    /**
+     * Emit the log lines attached to a dynamically-generated taskrun through the regular logging
+     * pipeline, so the standard appender behaviour (secret masking, long-message splitting, level
+     * filtering, forwarding to the server log, file vs queue routing) applies natively — the only
+     * thing that changes is the taskrun the lines are attributed to.
+     * <p>
+     * When this context logs to a file ({@code logToFile}), its logs are file-only (no inline/queue
+     * display): the lines are routed through this context's own logger so they land in the same
+     * downloadable file. A flat file has no taskrun, so there is nothing to link there.
+     * <p>
+     * Otherwise the lines go through a child logger bound to a {@link LogEntry} whose execution,
+     * tenant, namespace and flow are taken from this context's bound entry — a caller cannot target
+     * another execution or tenant — with the dynamic taskrun's id and a fixed attempt 0 (these
+     * taskruns have a single attempt and the log view groups by the 0-based attempt). The level
+     * filter and the secrets known to this context are inherited so nothing else is lost.
+     */
+    public void emitDynamicTaskRunLogs(TaskRun dynamicTaskRun, List<DynamicTaskRunLog> logs) {
+        // A logToFile task logs to a file only (no inline display): route the lines through this
+        // context's own logger so they join the same downloadable file (a flat file has no taskrun
+        // to link to). Otherwise emit through a logger bound to the dynamic taskrun, so the lines
+        // are attributed to it while still passing through the regular appender pipeline.
+        org.slf4j.Logger logger = this.logToFile ? this.logger() : deriveLoggerFor(dynamicTaskRun).logger();
+
+        for (DynamicTaskRunLog log : logs) {
+            logger.atLevel(log.level()).log(log.message());
+        }
+    }
+
+    /**
+     * Derive a logger bound to a dynamically-generated taskrun: a child of this context that shares
+     * its log emitter, level filter and known secrets, but emits under the taskrun's id with a fixed
+     * attempt 0 (these taskruns have a single attempt and the log view groups by the 0-based
+     * attempt). Execution, tenant, namespace and flow are taken from this context, so a caller can
+     * only ever target this execution's taskrun — never forge a log for another execution or tenant.
+     */
+    private RunContextLogger deriveLoggerFor(TaskRun dynamicTaskRun) {
+        LogEntry boundLogEntry = LogEntry.builder()
+            .tenantId(this.logEntry.getTenantId())
+            .executionId(this.logEntry.getExecutionId())
+            .namespace(this.logEntry.getNamespace())
+            .flowId(this.logEntry.getFlowId())
+            .executionKind(this.logEntry.getExecutionKind())
+            .taskId(dynamicTaskRun.getTaskId())
+            .taskRunId(dynamicTaskRun.getId())
+            .attemptNumber(0)
+            .build();
+
+        RunContextLogger taskRunLogger = new RunContextLogger(this.logEmitter, boundLogEntry, null, false);
+        taskRunLogger.loglevel = this.loglevel; // inherit this context's level filter as-is (logback Level)
+        taskRunLogger.useSecrets.addAll(this.useSecrets);
+        return taskRunLogger;
+    }
+
     private Logger initializeLogger() {
         LoggerContext loggerContext = new LoggerContext();
         LogbackMDCAdapter mdcAdapter = new LogbackMDCAdapter();
diff --git a/core/src/test/java/io/kestra/core/runners/RunContextLoggerTest.java b/core/src/test/java/io/kestra/core/runners/RunContextLoggerTest.java
@@ -16,6 +16,8 @@
 
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
+import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.executions.TaskRunAttempt;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.queues.BroadcastQueueInterface;
 import io.kestra.core.queues.DispatchQueueInterface;
@@ -213,6 +215,156 @@ void emitLogs_sendsToBothQueues() {
         assertThat(followQueueLogs).containsExactlyInAnyOrder(FollowLogEvent.from(e1), FollowLogEvent.from(e2));
     }
 
+    @Test
+    void emitDynamicTaskRunLogs_forcesContextAndAttemptZeroAndMasks() {
+        List<LogEntry> logs = new CopyOnWriteArrayList<>();
+        logQueue.addListener(logs::add);
+
+        Flow flow = TestsUtils.mockFlow();
+        Execution execution = TestsUtils.mockExecution(flow, Map.of());
+
+        RunContextLogger runContextLogger = new RunContextLogger(
+            logEntryEmitter,
+            LogEntry.of(execution),
+            Level.TRACE,
+            false
+        );
+        runContextLogger.usedSecret("super-secret-value");
+
+        // a dynamic taskrun that (deliberately) carries a foreign execution/tenant and one attempt:
+        // the emitted entry must NOT inherit those — execution/tenant/namespace/flow come from the context.
+        TaskRun dynamicTaskRun = TaskRun.builder()
+            .id("dyn-taskrun-id")
+            .taskId("Play | Task 1")
+            .tenantId("other-tenant")
+            .executionId("other-execution")
+            .namespace("other.namespace")
+            .flowId("other-flow")
+            .attempts(List.of(TaskRunAttempt.builder().build()))
+            .build();
+
+        runContextLogger.emitDynamicTaskRunLogs(dynamicTaskRun, List.of(new DynamicTaskRunLog(Level.ERROR, "leak super-secret-value here")));
+
+        List<LogEntry> queueLogs = TestsUtils.awaitLogs(logs, 1);
+        assertThat(queueLogs).hasSize(1);
+        LogEntry emitted = queueLogs.getFirst();
+        // taskrun identity comes from the dynamic taskrun
+        assertThat(emitted.getTaskRunId()).isEqualTo("dyn-taskrun-id");
+        assertThat(emitted.getTaskId()).isEqualTo("Play | Task 1");
+        // attempt is forced to 0 (these taskruns have one attempt; the log view groups by the 0-based attempt)
+        assertThat(emitted.getAttemptNumber()).isEqualTo(0);
+        // execution/tenant/namespace/flow are forced from the context, never from the (foreign) taskrun
+        assertThat(emitted.getExecutionId()).isEqualTo(execution.getId());
+        assertThat(emitted.getExecutionId()).isNotEqualTo("other-execution");
+        assertThat(emitted.getTenantId()).isEqualTo(execution.getTenantId());
+        assertThat(emitted.getNamespace()).isEqualTo(execution.getNamespace());
+        assertThat(emitted.getFlowId()).isEqualTo(execution.getFlowId());
+        // level preserved + secret masked
+        assertThat(emitted.getLevel()).isEqualTo(Level.ERROR);
+        assertThat(emitted.getMessage()).isEqualTo("leak ****** here");
+    }
+
+    @Test
+    void emitDynamicTaskRunLogs_inheritsLevelFilter() {
+        List<LogEntry> logs = new CopyOnWriteArrayList<>();
+        logQueue.addListener(logs::add);
+
+        Flow flow = TestsUtils.mockFlow();
+        Execution execution = TestsUtils.mockExecution(flow, Map.of());
+
+        // the context filters at WARN: an INFO dynamic line must be dropped, like any task log
+        RunContextLogger runContextLogger = new RunContextLogger(
+            logEntryEmitter,
+            LogEntry.of(execution),
+            Level.WARN,
+            false
+        );
+
+        TaskRun dynamicTaskRun = TaskRun.builder()
+            .id("dyn-taskrun-id")
+            .taskId("Play | Task 1")
+            .attempts(List.of(TaskRunAttempt.builder().build()))
+            .build();
+
+        runContextLogger.emitDynamicTaskRunLogs(dynamicTaskRun, List.of(
+            new DynamicTaskRunLog(Level.INFO, "info dropped by filter"),
+            new DynamicTaskRunLog(Level.ERROR, "error kept")
+        ));
+
+        List<LogEntry> queueLogs = TestsUtils.awaitLogs(logs, 1);
+        assertThat(queueLogs).hasSize(1);
+        assertThat(queueLogs.getFirst().getLevel()).isEqualTo(Level.ERROR);
+        assertThat(queueLogs.getFirst().getMessage()).isEqualTo("error kept");
+        assertThat(queueLogs).noneMatch(l -> l.getLevel().equals(Level.INFO));
+    }
+
+    @Test
+    void emitDynamicTaskRunLogs_underLogToFileGoesToFileNotQueue() throws Exception {
+        List<LogEntry> logs = new CopyOnWriteArrayList<>();
+        logQueue.addListener(logs::add);
+
+        Flow flow = TestsUtils.mockFlow();
+        Execution execution = TestsUtils.mockExecution(flow, Map.of());
+
+        // logToFile=true: task logs are file-only, so the dynamic lines must land in the file
+        // (with masking) and never reach the inline log queue
+        RunContextLogger runContextLogger = new RunContextLogger(
+            logEntryEmitter,
+            LogEntry.of(execution),
+            Level.TRACE,
+            true
+        );
+        runContextLogger.usedSecret("super-secret-value");
+
+        TaskRun dynamicTaskRun = TaskRun.builder()
+            .id("dyn-taskrun-id")
+            .taskId("Play | Task 1")
+            .attempts(List.of(TaskRunAttempt.builder().build()))
+            .build();
+
+        runContextLogger.emitDynamicTaskRunLogs(dynamicTaskRun, List.of(
+            new DynamicTaskRunLog(Level.INFO, "to file super-secret-value")
+        ));
+
+        runContextLogger.closeLogFile();
+        String fileContent = java.nio.file.Files.readString(runContextLogger.getLogFile().toPath());
+        assertThat(fileContent).contains("to file ******");
+        // file-only: ContextAppender is not attached, so nothing reaches the inline queue
+        assertThat(logs).isEmpty();
+    }
+
+    @Test
+    void emitDynamicTaskRunLogs_seedsMDCWithDynamicTaskRunIdentity() {
+        Flow flow = TestsUtils.mockFlow();
+        Execution execution = TestsUtils.mockExecution(flow, Map.of());
+
+        // mirror exactly how emitDynamicTaskRunLogs binds the child logger for a dynamic taskrun:
+        // execution context + the dynamic taskrun's id/taskId, attempt 0
+        LogEntry boundLogEntry = LogEntry.of(execution).toBuilder()
+            .taskId("Play | Task 1")
+            .taskRunId("dyn-taskrun-id")
+            .attemptNumber(0)
+            .build();
+
+        RunContextLogger runContextLogger = new RunContextLogger(
+            logEntryEmitter,
+            boundLogEntry,
+            Level.TRACE,
+            false
+        );
+        ch.qos.logback.classic.Logger perRunLogger =
+            (ch.qos.logback.classic.Logger) runContextLogger.logger();
+
+        // the per-run MDC carries the dynamic taskrun identity (taskRunId/taskId), not just the
+        // execution context — so forwarded server logs are attributed to the dynamic taskrun too
+        assertThat(perRunLogger.getLoggerContext().getMDCAdapter().getCopyOfContextMap())
+            .containsEntry("taskRunId", "dyn-taskrun-id")
+            .containsEntry("taskId", "Play | Task 1")
+            .containsEntry("executionId", execution.getId())
+            .containsEntry("namespace", execution.getNamespace())
+            .containsEntry("flowId", execution.getFlowId());
+    }
+
     @Test
     void transformPreservesMDC() throws Exception {
         Flow flow = TestsUtils.mockFlow();
diff --git a/ui/packages/topology/src/nodes/TaskNode.vue b/ui/packages/topology/src/nodes/TaskNode.vue
@@ -119,6 +119,7 @@
     interface TaskRun {
         id: string
         taskId: string;
+        parentTaskRunId?: string;
         state: {
             current: [string, string];
             duration?: string;
@@ -227,6 +228,16 @@
         )
     })
 
+    // The task's own taskruns plus any dynamically-generated child taskruns (e.g. Ansible
+    // plays/tasks) so "show task logs" surfaces their logs too, not just the task's root logs.
+    const taskRunsWithDynamicChildren = computed(() => {
+        const ids = new Set(taskRuns.value.map((t: TaskRun) => t.id))
+        const children = taskRunList.value.filter(
+            (t: TaskRun) => t.parentTaskRunId && ids.has(t.parentTaskRunId),
+        )
+        return [...taskRuns.value, ...children]
+    })
+
     const state = computed(() => {
         if (!taskRuns.value?.length) {
             return null
@@ -341,7 +352,7 @@
                 key: "logs",
                 label: t("show task logs"),
                 icon: TextBoxSearch,
-                onClick: () => emit(EVENTS.SHOW_LOGS, {id: taskId.value, execution: taskExecution.value, taskRuns: taskRuns.value}),
+                onClick: () => emit(EVENTS.SHOW_LOGS, {id: taskId.value, execution: taskExecution.value, taskRuns: taskRunsWithDynamicChildren.value}),
             })
         }
         if (taskExecution.value) {
diff --git a/ui/src/components/executions/Gantt.vue b/ui/src/components/executions/Gantt.vue
@@ -99,7 +99,10 @@
                                                 <ChevronRight v-if="!selectedTaskRuns.includes(item.id)" />
                                                 <ChevronDown v-else />
                                             </div>
-                                            <div class="task-label">
+                                            <div
+                                                class="task-label"
+                                                :style="{'--depth': item.depth || 0}"
+                                            >
                                                 <div v-if="taskTypeByTaskRunId[item.id]" class="task-icon-box">
                                                     <KsTaskIcon :cls="taskTypeByTaskRunId[item.id]" onlyIcon :icons="pluginsStore.icons" />
                                                 </div>
@@ -858,6 +861,7 @@
                     display: flex;
                     align-items: center;
                     gap: var(--ks-spacing-4);
+                    padding-left: calc(var(--depth, 0) * var(--ks-spacing-5));
 
                     code {
                         color: var(--ks-text-primary);
diff --git a/ui/src/components/executions/TaskRunLine.vue b/ui/src/components/executions/TaskRunLine.vue
@@ -1,5 +1,9 @@
 <template>
-    <div v-if="!hideHeader" class="taskrun-header">
+    <div
+        v-if="!hideHeader"
+        class="taskrun-header"
+        :style="{'--depth': depth}"
+    >
         <div>
             <KsIcon
                 v-if="!taskRunId && shouldDisplayChevron(currentTaskRun)"
@@ -135,6 +139,7 @@
         logs?: any[]
         filter?: string
         hideHeader?: boolean
+        depth?: number
     }
 
     const props = withDefaults(defineProps<Props>(), {
@@ -146,6 +151,7 @@
         logs: () => [],
         filter: "",
         hideHeader: false,
+        depth: 0,
     })
 
     const emit = defineEmits<{
@@ -247,6 +253,7 @@
 
     .taskrun-header {
         background-color: var(--ks-bg-surface);
+        padding-left: calc(var(--ks-spacing-4) + var(--depth, 0) * var(--ks-spacing-5));
 
         .task-icon {
             width: 36px;
diff --git a/ui/src/components/logs/TaskRunDetails.vue b/ui/src/components/logs/TaskRunDetails.vue
