Repository: kestra-io/kestra (java)

Issue:
Extend `errors` with a failed task ID(s) and, if possible, also the stack trace(s) of the failed task(s)
### Feature description

## Problem

Users would like to access the stack trace of a failed task within the notification message.

## Example

Using the following flow:

```yaml
id: onFailureAlert
namespace: blueprint

tasks:
  - id: fail
    type: io.kestra.plugin.scripts.shell.Commands
    runner: PROCESS
    commands:
      - exit 1

errors:
  - id: slack
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#alerts",
        "text": "Failure alert for flow {{ flow.namespace }}.{{ flow.id }} with ID {{ execution.id }}"
      }
```

the Slack task configured within the `errors` section should be able to retrieve the stack trace of the task `fail` to send a more informative failure notification

## Possible implementation

Because it may not be technically feasible for a failed task to emit the stack trace as an output variable, we should attempt to at least output the failed task ID. 

Pull request description:
feat(core): add error message and stacktrace to the execution
Add **the last** error message and stack trace to the execution context.
Stacktrace is limited to the top 10 frames. It may be a little too few, but we need to limit it to avoid cluttering the execution context.

The execution object itself now contains an error and a stack trace. As a follow-up, it may be a good idea to display them in the Execution overview.

I checked the Worker, the Scheduler, and the JDBCExecutor, and most, if not all, exceptional code paths should produce this new execution error attribute.

See for example the following flow:
```yaml
id: errors
namespace: company.team

tasks:
  - id: hello
    type: io.kestra.plugin.core.log.Log
    message: Starting the workflow
  - id: fail
    type: io.kestra.plugin.core.execution.Fail

errors:
  - id: error-log
    type: io.kestra.plugin.core.log.Log
    message:
      - "It's the fault of '{{ error.taskId }}'"
      - "See the message: {{ error.message }}"
      - "{{ error.stackTrace }}"
```

Results in the following logs, note that the stackrace is mangled because the logs view remove the carriage return
![image](https://github.com/user-attachments/assets/ba33eb49-d7bd-422d-91c2-49c444a5ed73)

In the server log or the Execution JSON it is correctly displayed
![image](https://github.com/user-attachments/assets/f45dff54-8ab9-4456-b6ec-828843ab871b)

Fixes #2000


Patch:
diff --git a/core/src/main/java/io/kestra/core/models/executions/Execution.java b/core/src/main/java/io/kestra/core/models/executions/Execution.java
index 8d6fbf30e..890c62976 100644
--- a/core/src/main/java/io/kestra/core/models/executions/Execution.java
+++ b/core/src/main/java/io/kestra/core/models/executions/Execution.java
@@ -105,6 +105,10 @@ public class Execution implements DeletedInterface, TenantInterface {
     @Nullable
     Instant scheduleDate;
 
+    @With
+    @Nullable
+    ExecutionError error;
+
     /**
      * Factory method for constructing a new {@link Execution} object for the given {@link Flow}.
      *
@@ -196,7 +200,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.trigger,
             this.deleted,
             this.metadata,
-            this.scheduleDate
+            this.scheduleDate,
+            this.error
         );
     }
 
@@ -230,7 +235,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.trigger,
             this.deleted,
             this.metadata,
-            this.scheduleDate
+            this.scheduleDate,
+            taskRun.getError() != null ? taskRun.getError() : this.error
         );
     }
 
@@ -252,7 +258,8 @@ public class Execution implements DeletedInterface, TenantInterface {
             this.trigger,
             this.deleted,
             this.metadata,
-            this.scheduleDate
+            this.scheduleDate,
+            this.error
         );
     }
 
@@ -631,7 +638,7 @@ public class Execution implements DeletedInterface, TenantInterface {
             .map(t -> {
                 try {
                     return new FailedExecutionWithLog(
-                        this.withTaskRun(t.getTaskRun()),
+                        this.withTaskRun(t.getTaskRun()).withError(ExecutionError.from(e)),
                         t.getLogs()
                     );
                 } catch (InternalException ex) {
@@ -639,7 +646,7 @@ public class Execution implements DeletedInterface, TenantInterface {
                 }
             })
             .orElseGet(() -> new FailedExecutionWithLog(
-                    this.state.getCurrent() != State.Type.FAILED ? this.withState(State.Type.FAILED) : this,
+                    this.state.getCurrent() != State.Type.FAILED ? this.withState(State.Type.FAILED).withError(ExecutionError.from(e)) : this.withError(ExecutionError.from(e)),
                     RunContextLogger.logEntries(loggingEventFromException(e), LogEntry.of(this))
                 )
             );
diff --git a/core/src/main/java/io/kestra/core/models/executions/ExecutionError.java b/core/src/main/java/io/kestra/core/models/executions/ExecutionError.java
new file mode 100644
index 000000000..70d0055c6
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/executions/ExecutionError.java
@@ -0,0 +1,45 @@
+package io.kestra.core.models.executions;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+@Builder
+@Setter
+@Getter
+public class ExecutionError {
+    private static final int MAX_NB_FRAMES = 10;
+
+    private String message;
+    private String stacktrace;
+
+    public static ExecutionError from(Throwable throwable) {
+        if (throwable == null) {
+            return ExecutionError.builder().message("Unknown error").build();
+        }
+
+        String firstLine = throwable.getClass().getName() + ": " + throwable.getMessage() + "\n";
+        StackTraceElement[] stackTraces = throwable.getStackTrace();
+        String stackTraceStr;
+        if (stackTraces.length > 10) {
+            // keep only the top 10 frames
+            stackTraceStr = stackTraceToString(firstLine, Arrays.copyOf(stackTraces, 10)) + "\n\t[...]";
+        } else {
+            stackTraceStr = stackTraceToString(firstLine, stackTraces);
+        }
+        return ExecutionError.builder()
+            .message(throwable.getMessage())
+            .stacktrace(stackTraceStr)
+            .build();
+    }
+
+    private static String stackTraceToString(String firstLine, StackTraceElement[] stackTraces) {
+        return Stream.of(stackTraces)
+            .map(stackTraceElement -> stackTraceElement.toString())
+            .collect(Collectors.joining("\n\t", firstLine, ""));
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/models/executions/TaskRun.java b/core/src/main/java/io/kestra/core/models/executions/TaskRun.java
index 06069e5b0..bff0e3d23 100644
--- a/core/src/main/java/io/kestra/core/models/executions/TaskRun.java
+++ b/core/src/main/java/io/kestra/core/models/executions/TaskRun.java
@@ -57,6 +57,9 @@ public class TaskRun implements TenantInterface {
     @With
     Integer iteration;
 
+    @With
+    ExecutionError error;
+
     @Deprecated
     public void setItems(String items) {
         // no-op for backward compatibility
@@ -75,7 +78,8 @@ public class TaskRun implements TenantInterface {
             this.attempts,
             this.outputs,
             this.state.withState(state),
-            this.iteration
+            this.iteration,
+            this.error
         );
     }
 
@@ -92,7 +96,8 @@ public class TaskRun implements TenantInterface {
             this.attempts,
             this.outputs,
             newState,
-            this.iteration
+            this.iteration,
+            this.error
         );
     }
 
@@ -113,7 +118,8 @@ public class TaskRun implements TenantInterface {
             newAttempts,
             this.outputs,
             this.state.withState(State.Type.FAILED),
-            this.iteration
+            this.iteration,
+            this.error
         );
     }
 
diff --git a/core/src/main/java/io/kestra/core/runners/Executor.java b/core/src/main/java/io/kestra/core/runners/Executor.java
index ecff473fe..a5a08bdd1 100644
--- a/core/src/main/java/io/kestra/core/runners/Executor.java
+++ b/core/src/main/java/io/kestra/core/runners/Executor.java
@@ -1,10 +1,7 @@
 package io.kestra.core.runners;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.ExecutionKilled;
-import io.kestra.core.models.executions.ExecutionKilledExecution;
-import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowWithException;
 import io.kestra.core.models.flows.State;
@@ -104,6 +101,7 @@ public class Executor {
         this.exception = exception;
         this.from.add(from);
         this.executionUpdated = true;
+        this.execution = this.execution.withError(ExecutionError.from(exception));
 
         return this;
     }
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
index cee43efa3..04968302c 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -847,11 +847,7 @@ public class ExecutorService {
                     }
                 } catch (Exception e) {
                     WorkerTaskResult failed = WorkerTaskResult.builder()
-                        .taskRun(workerTask.getTaskRun().withState(State.Type.FAILED)
-                            .withAttempts(Collections.singletonList(
-                                TaskRunAttempt.builder().state(new State().withState(State.Type.FAILED)).build()
-                            ))
-                        )
+                        .taskRun(workerTask.getTaskRun().fail())
                         .build();
                     executor
                         .withWorkerTaskResults(List.of(failed), "handleExecutableTask")
@@ -902,11 +898,7 @@ public class ExecutorService {
                     );
                 } catch (Exception e) {
                     workerTaskResults.add(WorkerTaskResult.builder()
-                        .taskRun(workerTask.getTaskRun().withState(State.Type.FAILED)
-                            .withAttempts(Collections.singletonList(
-                                TaskRunAttempt.builder().state(new State().withState(State.Type.FAILED)).build()
-                            ))
-                        )
+                        .taskRun(workerTask.getTaskRun().fail())
                         .build());
                     executor.withException(e, "handleExecutionUpdatingTask");
                 }
diff --git a/core/src/main/java/io/kestra/core/runners/RunVariables.java b/core/src/main/java/io/kestra/core/runners/RunVariables.java
index 63a69bd94..0ac5673c1 100644
--- a/core/src/main/java/io/kestra/core/runners/RunVariables.java
+++ b/core/src/main/java/io/kestra/core/runners/RunVariables.java
@@ -247,10 +247,20 @@ public final class RunVariables {
                     builder.put("tasks", tasksMap);
 
                     // search for failures
-                    execution.getTaskRunList().reversed().stream()
+                    Map<String, Object> error = new HashMap<>();
+                    Optional<TaskRun> failedTaskRun = execution.getTaskRunList().reversed().stream()
                         .filter(taskRun -> taskRun.getState() != null && taskRun.getState().isFailed())
-                        .findFirst()
-                        .ifPresent(taskRun -> builder.put("error", Map.of("taskId", taskRun.getTaskId())));
+                        .findFirst();
+                    if (failedTaskRun.isPresent() || execution.getError() != null) {
+                        failedTaskRun.ifPresent(run -> error.put("taskId", run.getTaskId()));
+                        if (execution.getError() != null) {
+                            error.put("message", execution.getError().getMessage());
+                            error.put("stackTrace", execution.getError().getStacktrace());
+                        }
+                    }
+                    if (!error.isEmpty()) {
+                        builder.put("error", error);
+                    }
                 }
 
                 // Inputs
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index 11b580c38..03c1308ab 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -9,6 +9,7 @@ import io.kestra.core.exceptions.DeserializationException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.*;
+import io.kestra.core.models.executions.ExecutionError;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.*;
@@ -292,7 +293,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
                     workingDirectory.preExecuteTasks(workingDirectoryRunContext, workerTask.getTaskRun());
                 } catch (Exception e) {
                     workingDirectoryRunContext.logger().error("Failed preExecuteTasks on WorkingDirectory: {}", e.getMessage(), e);
-                    WorkerTask failed = workerTask.withTaskRun(workerTask.fail());
+                    WorkerTask failed = workerTask.withTaskRun(workerTask.fail(e));
                     try {
                         this.workerTaskResultQueue.emit(new WorkerTaskResult(failed.getTaskRun()));
                     } catch (QueueException ex) {
@@ -320,8 +321,6 @@ public class Worker implements Service, Runnable, AutoCloseable {
                         break;
                     }
 
-
-
                     // create the next RunContext populated with the previous WorkerTaskResult
                     runContext = runContextInitializer.forWorker(runContext.clone(), workerTaskResult, workerTask.getTaskRun());
                 }
@@ -332,7 +331,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
                 } catch (Exception e) {
                     workingDirectoryRunContext.logger().error("Failed postExecuteTasks on WorkingDirectory: {}", e.getMessage(), e);
                     try {
-                        this.workerTaskResultQueue.emit(new WorkerTaskResult(workerTask.fail()));
+                        this.workerTaskResultQueue.emit(new WorkerTaskResult(workerTask.fail(e)));
                     } catch (QueueException ex) {
                         log.error("Unable to emit the worker task result for task {} taskrun {}", workerTask.getTask().getId(), workerTask.getTaskRun().getId(), e);
                     }
@@ -413,7 +412,8 @@ public class Worker implements Service, Runnable, AutoCloseable {
         // We create a FAILED execution, so the user is aware that the realtime trigger failed to be created
         var execution = TriggerService
             .generateRealtimeExecution(workerTrigger.getTrigger(), workerTrigger.getConditionContext(), workerTrigger.getTriggerContext(), null)
-            .withState(FAILED);
+            .withState(FAILED)
+            .withError(ExecutionError.from(e));
 
         // We create an ERROR log attached to the execution
         Logger logger = workerTrigger.getConditionContext().getRunContext().logger();
@@ -613,7 +613,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
         } catch (QueueException e) {
             // If there is a QueueException it can either be caused by the message limit or another queue issue.
             // We fail the task and try to resend it.
-            TaskRun failed  = workerTask.fail();
+            TaskRun failed  = workerTask.fail(e);
             if (e instanceof MessageTooBigException) {
                 // If it's a message too big, we remove the outputs
                 failed = failed.withOutputs(Collections.emptyMap());
@@ -756,6 +756,10 @@ public class Worker implements Service, Runnable, AutoCloseable {
             logger.warn("Unable to save output on taskRun '{}'", taskRun, e);
         }
 
+        if (workerTaskCallable.exception !=null) {
+            taskRun = taskRun.withError(ExecutionError.from(workerTaskCallable.exception));
+        }
+
         return workerTask
             .withTaskRun(taskRun);
     }
diff --git a/core/src/main/java/io/kestra/core/runners/WorkerTask.java b/core/src/main/java/io/kestra/core/runners/WorkerTask.java
index cfd6e018d..f98495336 100644
--- a/core/src/main/java/io/kestra/core/runners/WorkerTask.java
+++ b/core/src/main/java/io/kestra/core/runners/WorkerTask.java
@@ -1,6 +1,7 @@
 package io.kestra.core.runners;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
+import io.kestra.core.models.executions.ExecutionError;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.Task;
@@ -51,8 +52,8 @@ public class WorkerTask extends WorkerJob {
      *
      * @return this worker task, updated
      */
-    public TaskRun fail() {
+    public TaskRun fail(Throwable cause) {
         var state = this.task.isAllowFailure() ? this.task.isAllowWarning() ? State.Type.SUCCESS : State.Type.WARNING : State.Type.FAILED;
-        return this.getTaskRun().withState(state);
+        return this.getTaskRun().withState(state).withError(ExecutionError.from(cause));
     }
 }
diff --git a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
index 4bee45f62..73e16b2ff 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -8,6 +8,7 @@ import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.conditions.Condition;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.ExecutionError;
 import io.kestra.core.models.executions.ExecutionKilled;
 import io.kestra.core.models.executions.ExecutionKilledTrigger;
 import io.kestra.core.models.flows.FlowWithException;
@@ -565,6 +566,7 @@ public abstract class AbstractScheduler implements Scheduler, Service {
                             .flowRevision(f.getFlow().getRevision())
                             .labels(f.getFlow().getLabels())
                             .state(new State().withState(State.Type.FAILED))
+                            .error(ExecutionError.from(ie))
                             .build();
                         ZonedDateTime nextExecutionDate = this.nextEvaluationDate(f.getAbstractTrigger());
                         var trigger = f.getTriggerContext().resetExecution(State.Type.FAILED, nextExecutionDate);
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
index 5c0aba449..167c34af3 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -438,7 +438,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                                     }
                                 }
                             } catch (IllegalVariableEvaluationException e) {
-                                workerTaskResultQueue.emit(new WorkerTaskResult(workerTask.getTaskRun().withState(State.Type.FAILED)));
+                                workerTaskResultQueue.emit(new WorkerTaskResult(workerTask.getTaskRun().withState(State.Type.FAILED).withError(ExecutionError.from(e))));
                                 workerTask.getRunContext().logger().error("Unable to evaluate the runIf condition for task {}", workerTask.getTask().getId(), e);
                             }
                         }));
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index 7a5b1dbdd..0f1947034 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -6,12 +6,8 @@ import io.kestra.core.events.CrudEventType;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.Label;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.executions.ExecutionKilled;
-import io.kestra.core.models.executions.ExecutionKilledExecution;
-import io.kestra.core.models.executions.ExecutionMetadata;
-import io.kestra.core.models.executions.ExecutionTrigger;
-import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.executions.*;
+import io.kestra.core.models.executions.ExecutionError;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.FlowForExecution;
 import io.kestra.core.models.flows.FlowScope;
@@ -661,8 +657,8 @@ public class ExecutionController {
         private final URI url;
 
         // This is not nice, but we cannot use @AllArgsConstructor as it would open a bunch of necessary changes on the Execution class.
-        ExecutionResponse(String tenantId, String id, String namespace, String flowId, Integer flowRevision, List<TaskRun> taskRunList, Map<String, Object> inputs, Map<String, Object> outputs, List<Label> labels, Map<String, Object> variables, State state, String parentId, String originalId, ExecutionTrigger trigger, boolean deleted, ExecutionMetadata metadata, Instant scheduleDate, URI url) {
-            super(tenantId, id, namespace, flowId, flowRevision, taskRunList, inputs, outputs, labels, variables, state, parentId, originalId, trigger, deleted, metadata, scheduleDate);
+        ExecutionResponse(String tenantId, String id, String namespace, String flowId, Integer flowRevision, List<TaskRun> taskRunList, Map<String, Object> inputs, Map<String, Object> outputs, List<Label> labels, Map<String, Object> variables, State state, String parentId, String originalId, ExecutionTrigger trigger, boolean deleted, ExecutionMetadata metadata, Instant scheduleDate, ExecutionError error, URI url) {
+            super(tenantId, id, namespace, flowId, flowRevision, taskRunList, inputs, outputs, labels, variables, state, parentId, originalId, trigger, deleted, metadata, scheduleDate, error);
 
             this.url = url;
         }
@@ -686,6 +682,7 @@ public class ExecutionController {
                 execution.isDeleted(),
                 execution.getMetadata(),
                 execution.getScheduleDate(),
+                execution.getError(),
                 url
             );
         }


Test patch:
diff --git a/core/src/test/resources/flows/valids/errors.yaml b/core/src/test/resources/flows/valids/errors.yaml
index cf61b9643..252d8ff47 100644
--- a/core/src/test/resources/flows/valids/errors.yaml
+++ b/core/src/test/resources/flows/valids/errors.yaml
@@ -7,7 +7,10 @@ tasks:
 errors:
   - id: t2
     type: io.kestra.plugin.core.log.Log
-    message: The task that that fail is '{{ error.taskId }}'
+    message:
+      - "It's the fault of '{{ error.taskId }}'"
+      - "See the message: {{ error.message }}"
+      - "See the stackTrace: {{ error.stackTrace }}"
 
   - id: t3
     type: io.kestra.plugin.core.flow.Parallel
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
index a1de7da38..978805478 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -127,9 +127,15 @@ public abstract class JdbcRunnerTest {
         assertThat(execution.getTaskRunList(), hasSize(7));
 
         receive.blockLast();
-        LogEntry logEntry = TestsUtils.awaitLog(logs, log -> log.getMessage().startsWith("The task that that fail"));
+        LogEntry logEntry = TestsUtils.awaitLog(logs, log -> log.getMessage().startsWith("It's the fault of "));
         assertThat(logEntry, notNullValue());
-        assertThat(logEntry.getMessage(), is("The task that that fail is 'failed'"));
+        assertThat(logEntry.getMessage(), is("It's the fault of 'failed'"));
+        logEntry = TestsUtils.awaitLog(logs, log -> log.getMessage().startsWith("See the message: "));
+        assertThat(logEntry, notNullValue());
+        assertThat(logEntry.getMessage(), is("See the message: Task failure"));
+        logEntry = TestsUtils.awaitLog(logs, log -> log.getMessage().startsWith("See the stackTrace: "));
+        assertThat(logEntry, notNullValue());
+        assertThat(logEntry.getMessage(), startsWith("See the stackTrace: java.lang.Exception: Task failure"));
     }
 
     @Test