Repository: kestra-io/kestra (java)

Issue:
Ensure that the errorLogs() function captures error logs from a script
### Feature description

```yaml
id: error_logs_demo
namespace: company.team

tasks:    
  - id: fail
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    script: |
      raise ValueError("An error occurred: This is a manually raised exception.")
errors:
  - id: alert
    type: io.kestra.plugin.core.log.Log
    message: list of error logs — {{ errorLogs() }}
```

Another example:

```yaml
id: error_logs_demo
namespace: company.team

tasks:    
  - id: fail
    type: io.kestra.plugin.scripts.python.Script
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    script: |
      def divide_numbers(a, b):
          return a / b

      result = divide_numbers(5, 0)
      print(result)

errors:
  - id: alert
    type: io.kestra.plugin.core.log.Log
    message: list of error logs — {{ errorLogs() }}
```

The only error log it captures is `Command failed with exit code 1`  - the actual log with a reason why the task failed is not captured.

```bash
list of error logs — [{"namespace":"company.team","flowId":"error_logs_demo2","taskId":"fail","executionId":"41ClpgK5tLqp2WqKsM1U3c","taskRunId":"tfBn6QL2y51KtWxUDbWf7","attemptNumber":0,"timestamp":"2024-11-25T15:40:22.226Z","level":"ERROR","thread":"worker_4","message":"Command failed with exit code 1","deleted":false},{"namespace":"company.team","flowId":"error_logs_demo2","taskId":"fail","executionId":"41ClpgK5tLqp2WqKsM1U3c","taskRunId":"tfBn6QL2y51KtWxUDbWf7","attemptNumber":0,"timestamp":"2024-11-25T15:40:22.226Z","level":"ERROR","thread":"worker_4","message":"io.kestra.core.models.tasks.runners.TaskException: Command failed with exit code 1","deleted":false}]
```

Pull request description:
fix(core): wait for error logs in the Pebble function
Fixes #6083


Patch:
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 11f99e945..7e8598cae 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -61,6 +61,8 @@ import static io.kestra.core.utils.Rethrow.throwFunction;
 @Slf4j
 @Introspected
 public class Worker implements Service, Runnable, AutoCloseable {
+    public static final String EXECUTOR_NAME = "worker";
+
     private static final ObjectMapper MAPPER = JacksonMapper.ofJson();
     private static final String SERVICE_PROPS_WORKER_GROUP = "worker.group";
 
@@ -158,7 +160,7 @@ public class Worker implements Service, Runnable, AutoCloseable {
         this.numThreads = numThreads;
         this.workerGroup = workerGroupService.resolveGroupFromKey(workerGroupKey);
         this.eventPublisher = eventPublisher;
-        this.executorService = executorsUtils.maxCachedThreadPool(numThreads, "worker");
+        this.executorService = executorsUtils.maxCachedThreadPool(numThreads, EXECUTOR_NAME);
         this.setState(ServiceState.CREATED);
     }
 
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/PebbleUtils.java b/core/src/main/java/io/kestra/core/runners/pebble/PebbleUtils.java
new file mode 100644
index 000000000..395806725
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/pebble/PebbleUtils.java
@@ -0,0 +1,24 @@
+package io.kestra.core.runners.pebble;
+
+import io.kestra.core.runners.Worker;
+import io.micronaut.context.annotation.Value;
+import jakarta.inject.Singleton;
+
+@Singleton
+public class PebbleUtils {
+    @Value("${kestra.server-type:}") // default to empty as tests didn't set this property
+    private String serverType;
+
+    public boolean calledOnWorker() {
+        if ("WORKER".equals(serverType)) {
+            return true;
+        }
+        if ("STANDALONE".equals(serverType)) {
+            // check that it's called inside a worker thread
+            // Note: this is not ideal as we check that it starts with the name used in the Worker executor
+            return Thread.currentThread().getName().startsWith(Worker.EXECUTOR_NAME + "_");
+        }
+
+        return false;
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunction.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunction.java
index e83f9fdc7..2090cf9f0 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunction.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunction.java
@@ -1,14 +1,20 @@
 package io.kestra.core.runners.pebble.functions;
 
+import io.kestra.core.models.executions.LogEntry;
+import io.kestra.core.models.tasks.retrys.Exponential;
+import io.kestra.core.runners.pebble.PebbleUtils;
 import io.kestra.core.services.LogService;
+import io.kestra.core.utils.ListUtils;
+import io.kestra.core.utils.RetryUtils;
 import io.micronaut.context.annotation.Requires;
-import io.micronaut.core.util.StringUtils;
+import io.pebbletemplates.pebble.error.PebbleException;
 import io.pebbletemplates.pebble.extension.Function;
 import io.pebbletemplates.pebble.template.EvaluationContext;
 import io.pebbletemplates.pebble.template.PebbleTemplate;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -19,6 +25,12 @@ public class ErrorLogsFunction  implements Function {
     @Inject
     private LogService logService;
 
+    @Inject
+    private PebbleUtils pebbleUtils;
+
+    @Inject
+    private RetryUtils retryUtils;
+
     @Override
     public List<String> getArgumentNames() {
         return Collections.emptyList();
@@ -27,8 +39,27 @@ public class ErrorLogsFunction  implements Function {
     @Override
     @SuppressWarnings("unchecked")
     public Object execute(Map<String, Object> args, PebbleTemplate self, EvaluationContext context, int lineNumber) {
+        if (!pebbleUtils.calledOnWorker()) {
+            throw new PebbleException(null, "The 'errorLogs' function can only be used in the Worker as it access logs from the database.", lineNumber, self.getName());
+        }
+
         Map<String, String> flow = (Map<String, String>) context.getVariable("flow");
         Map<String, String> execution = (Map<String, String>) context.getVariable("execution");
-        return logService.errorLogs(flow.get("tenantId"), execution.get("id"));
+
+        RetryUtils.Instance<List<LogEntry>, Throwable> retry = retryUtils.of(Exponential.builder()
+            .delayFactor(2.0)
+            .interval(Duration.ofMillis(100))
+            .maxInterval(Duration.ofSeconds(1))
+            .maxAttempt(-1)
+            .maxDuration(Duration.ofSeconds(5))
+            .build());
+
+        try {
+            return retry.run( logs -> ListUtils.isEmpty(logs), () -> logService.errorLogs(flow.get("tenantId"), execution.get("id")));
+        } catch (RetryUtils.RetryFailed e) {
+            return Collections.emptyList();
+        } catch (Throwable e) {
+            throw new PebbleException(e, "Unable to fetch error logs");
+        }
     }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
index 90c3ca40b..ae3f24e11 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/ReadFileFunction.java
@@ -67,17 +67,6 @@ public class ReadFileFunction implements Function {
         }
     }
 
-    private boolean calledOnWorker() {
-        if ("WORKER".equals(serverType)) {
-            return true;
-        }
-        if ("STANDALONE".equals(serverType)) {
-            // check that it's called inside a worker thread
-            return Thread.currentThread().getClass().getName().equals("io.kestra.core.runners.Worker$WorkerThread");
-        }
-        return false;
-    }
-
     @SuppressWarnings("unchecked")
     private String readFromNamespaceFile(EvaluationContext context, String path) throws IOException {
         Map<String, String> flow = (Map<String, String>) context.getVariable("flow");


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunctionTest.java b/core/src/test/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunctionTest.java
index 23e1d8790..c1ecddf2e 100644
--- a/core/src/test/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunctionTest.java
+++ b/core/src/test/java/io/kestra/core/runners/pebble/functions/ErrorLogsFunctionTest.java
@@ -5,6 +5,7 @@ import io.kestra.core.junit.annotations.KestraTest;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.repositories.LogRepositoryInterface;
 import io.kestra.core.runners.VariableRenderer;
+import io.micronaut.context.annotation.Property;
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
@@ -18,6 +19,7 @@ import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.is;
 
 @KestraTest
+@Property(name = "kestra.server-type", value = "WORKER")
 class ErrorLogsFunctionTest {
     @Inject
     private LogRepositoryInterface logRepository;