feat(*): OpenTelemetry traces

Fixes #6149

build.gradle

@@ -123,6 +123,7 @@ allprojects {
             micronaut "io.micronaut.micrometer:micronaut-micrometer-registry-prometheus"
             micronaut "io.micronaut:micronaut-http-client"
             micronaut "io.micronaut.reactor:micronaut-reactor-http-client"
+            micronaut "io.micronaut.tracing:micronaut-tracing-opentelemetry-http"
 
             // logs
             implementation "org.slf4j:slf4j-api"
@@ -133,6 +134,9 @@ allprojects {
             implementation group: 'org.slf4j', name: 'jcl-over-slf4j'
             implementation group: 'org.fusesource.jansi', name: 'jansi'
 
+            // OTEL
+            implementation "io.opentelemetry:opentelemetry-exporter-otlp"
+
             // jackson
             implementation group: 'com.fasterxml.jackson.core', name: 'jackson-core'
             implementation group: 'com.fasterxml.jackson.core', name: 'jackson-databind'

cli/src/main/resources/application.yml

@@ -160,6 +160,9 @@ kestra:
   metrics:
     prefix: kestra
 
+  traces:
+    root: DISABLED
+
   server:
     basic-auth:
       enabled: false
@@ -194,3 +197,12 @@ kestra:
     prefixes:
       - system.
       - internal.
+
+otel:
+  exclusions:
+    - /ping
+    - /metrics
+    - /health
+    - /env
+    - /prometheus
+  propagators: tracecontext, baggage

core/src/main/java/io/kestra/core/http/HttpRequest.java

@@ -1,6 +1,7 @@
 package io.kestra.core.http;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
@@ -93,7 +94,7 @@ public static HttpRequest of(URI uri, String method, RequestBody body, Map<Strin
             .build();
     }
 
-    public HttpUriRequest to() throws IOException {
+    public HttpUriRequest to(RunContext runContext) throws IOException {
         HttpUriRequestBase builder = new HttpUriRequestBase(this.method, this.uri);
 
         // headers
@@ -104,6 +105,10 @@ public HttpUriRequest to() throws IOException {
                 );
         }
 
+        if (runContext.getTraceParent() != null) {
+            builder.addHeader("traceparent", runContext.getTraceParent());
+        }
+
         // body
         if (this.body != null) {
             builder.setEntity(this.body.to());

core/src/main/java/io/kestra/core/http/client/HttpClient.java

@@ -248,7 +248,7 @@ private <T> HttpResponse<T> request(
         HttpClientResponseHandler<HttpResponse<T>> responseHandler
     ) throws HttpClientException {
         try {
-            return this.client.execute(request.to(), httpClientContext, responseHandler);
+            return this.client.execute(request.to(runContext), httpClientContext, responseHandler);
         } catch (SocketException e) {
             throw new HttpClientRequestException(e.getMessage(), request, e);
         } catch (IOException e) {

core/src/main/java/io/kestra/core/models/executions/Execution.java

@@ -29,6 +29,7 @@
 import jakarta.validation.constraints.Pattern;
 import lombok.*;
 import lombok.experimental.FieldDefaults;
+import lombok.experimental.NonFinal;
 import lombok.extern.slf4j.Slf4j;
 
 import java.time.Instant;
@@ -106,6 +107,10 @@ public class Execution implements DeletedInterface, TenantInterface {
     @Nullable
     Instant scheduleDate;
 
+    @NonFinal
+    @Setter
+    String traceParent;
+
     /**
      * Factory method for constructing a new {@link Execution} object for the given {@link Flow}.
      *
@@ -199,7 +204,8 @@ public Execution withState(State.Type state) {
             this.trigger,
             this.deleted,
             this.metadata,
-            this.scheduleDate
+            this.scheduleDate,
+            this.traceParent
         );
     }
 
@@ -222,7 +228,8 @@ public Execution withLabels(List<Label> labels) {
             this.trigger,
             this.deleted,
             this.metadata,
-            this.scheduleDate
+            this.scheduleDate,
+            this.traceParent
         );
     }
 
@@ -258,7 +265,8 @@ public Execution withTaskRun(TaskRun taskRun) throws InternalException {
             this.trigger,
             this.deleted,
             this.metadata,
-            this.scheduleDate
+            this.scheduleDate,
+            this.traceParent
         );
     }
 
@@ -281,7 +289,8 @@ public Execution childExecution(String childExecutionId, List<TaskRun> taskRunLi
             this.trigger,
             this.deleted,
             this.metadata,
-            this.scheduleDate
+            this.scheduleDate,
+            this.traceParent
         );
     }
 

core/src/main/java/io/kestra/core/runners/AbstractWorkerCallable.java

@@ -2,6 +2,8 @@
 
 import io.kestra.core.models.WorkerJobLifecycle;
 import io.kestra.core.models.flows.State;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.StatusCode;
 import lombok.Getter;
 import lombok.Synchronized;
 import org.slf4j.Logger;
@@ -25,6 +27,9 @@ public abstract class AbstractWorkerCallable implements Callable<State.Type> {
     @Getter
     String type;
 
+    @Getter
+    String uid;
+
     @Getter
     Throwable exception;
 
@@ -34,10 +39,11 @@ public abstract class AbstractWorkerCallable implements Callable<State.Type> {
 
     private Thread currentThread;
 
-    AbstractWorkerCallable(RunContext runContext, String type, ClassLoader classLoader) {
+    AbstractWorkerCallable(RunContext runContext, String type, String uid, ClassLoader classLoader) {
         this.logger = runContext.logger();
         this.runContext = runContext;
         this.type = type;
+        this.uid = uid;
         this.classLoader = classLoader;
     }
 
@@ -100,6 +106,7 @@ protected void kill(boolean markAsKilled) {
 
     protected State.Type exceptionHandler(Throwable e) {
         this.exception = e;
+        Span.current().recordException(e).setStatus(StatusCode.ERROR);
 
         if (this.killed) {
             return KILLED;

core/src/main/java/io/kestra/core/runners/AbstractWorkerTriggerCallable.java

@@ -13,7 +13,7 @@ abstract class AbstractWorkerTriggerCallable extends AbstractWorkerCallable {
     WorkerTrigger workerTrigger;
 
     AbstractWorkerTriggerCallable(RunContext runContext, String type, WorkerTrigger workerTrigger) {
-        super(runContext, type, workerTrigger.getTrigger().getClass().getClassLoader());
+        super(runContext, type, workerTrigger.uid(), workerTrigger.getTrigger().getClass().getClassLoader());
         this.workerTrigger = workerTrigger;
     }
 

core/src/main/java/io/kestra/core/runners/DefaultRunContext.java

@@ -60,6 +60,7 @@ public class DefaultRunContext extends RunContext {
     private Storage storage;
     private Map<String, Object> pluginConfiguration;
     private List<String> secretInputs;
+    private String traceParent;
 
     // those are only used to validate dynamic properties inside the RunContextProperty
     private Task task;
@@ -103,6 +104,20 @@ public List<String> getSecretInputs() {
         return secretInputs;
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    @JsonInclude
+    public String getTraceParent() {
+        return traceParent;
+    }
+
+    @Override
+    public void setTraceParent(String traceParent) {
+        this.traceParent = traceParent;
+    }
+
     @JsonIgnore
     public ApplicationContext getApplicationContext() {
         return applicationContext;