diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
index 04c01d45a8..4fb798347f 100644
--- a/core/src/main/java/io/kestra/core/models/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -13,6 +13,7 @@
 import io.kestra.core.models.Label;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.sla.SLA;
 import io.kestra.core.models.listeners.Listener;
 import io.kestra.core.models.tasks.FlowableTask;
 import io.kestra.core.models.tasks.Task;
@@ -116,7 +117,10 @@ public List<PluginDefault> getTaskDefaults() {
     List<Output> outputs;
 
     @Valid
-    protected AbstractRetry retry;
+    AbstractRetry retry;
+
+    List<SLA> sla;
+
 
     public Logger logger() {
         return LoggerFactory.getLogger("flow." + this.id);
diff --git a/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java b/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
index 8b4d796348..efa583b135 100644
--- a/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
+++ b/core/src/main/java/io/kestra/core/models/flows/FlowWithSource.java
@@ -36,6 +36,7 @@ public Flow toFlow() {
             .deleted(this.deleted)
             .concurrency(this.concurrency)
             .retry(this.retry)
+            .sla(this.sla)
             .build();
     }
 
@@ -89,6 +90,7 @@ public static FlowWithSource of(Flow flow, String source) {
             .source(source)
             .concurrency(flow.concurrency)
             .retry(flow.retry)
+            .sla(flow.sla)
             .build();
     }
 }
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/ExecutionChangedSLA.java b/core/src/main/java/io/kestra/core/models/flows/sla/ExecutionChangedSLA.java
new file mode 100644
index 0000000000..520847be12
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/ExecutionChangedSLA.java
@@ -0,0 +1,8 @@
+package io.kestra.core.models.flows.sla;
+
+/**
+ * Marker interface to denote an SLA as evaluating on execution change.
+ * ExecutionChangedSLA will be evaluated on each execution change, a.k.a. at the beginning of the processing of the execution queue.
+ */
+public interface ExecutionChangedSLA {
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/ExecutionMonitoringSLA.java b/core/src/main/java/io/kestra/core/models/flows/sla/ExecutionMonitoringSLA.java
new file mode 100644
index 0000000000..da98fe0257
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/ExecutionMonitoringSLA.java
@@ -0,0 +1,12 @@
+package io.kestra.core.models.flows.sla;
+
+import java.time.Duration;
+
+/**
+ * Marker interface to denote an SLA as evaluating using an {@link SLAMonitor}.
+ * ExecutionMonitoringSLA will be evaluated on a deadline defined by the monitor;
+ * the monitor is created when the execution is created.
+ */
+public interface ExecutionMonitoringSLA {
+    Duration getDuration();
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/SLA.java b/core/src/main/java/io/kestra/core/models/flows/sla/SLA.java
new file mode 100644
index 0000000000..c61003fbd8
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/SLA.java
@@ -0,0 +1,57 @@
+package io.kestra.core.models.flows.sla;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.sla.types.ExecutionConditionSLA;
+import io.kestra.core.models.flows.sla.types.MaxDurationSLA;
+import io.kestra.core.runners.RunContext;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.util.Map;
+import java.util.Optional;
+
+@SuperBuilder
+@Getter
+@NoArgsConstructor
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", visible = true, include = JsonTypeInfo.As.EXISTING_PROPERTY)
+@JsonSubTypes({
+    @JsonSubTypes.Type(value = MaxDurationSLA.class, name = "MAX_DURATION"),
+    @JsonSubTypes.Type(value = ExecutionConditionSLA.class, name = "EXECUTION_CONDITION"),
+})
+public abstract class SLA {
+    @NotNull
+    @NotEmpty
+    private String id;
+
+    @NotNull
+    private SLA.Type type;
+
+    @NotNull
+    private Behavior behavior;
+
+    // TODO prevent system labels
+    private Map<String, Object> labels;
+
+    /**
+     * Evaluate a flow SLA on an execution.
+     * In case the SLA is exceeded, a violation will be returned.
+     */
+    public abstract Optional<Violation> evaluate(RunContext runContext, Execution execution) throws InternalException;
+
+    public enum Type {
+        MAX_DURATION,
+        EXECUTION_CONDITION,
+    }
+
+    public enum Behavior {
+        FAIL,
+        CANCEL,
+        NONE
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/SLAMonitor.java b/core/src/main/java/io/kestra/core/models/flows/sla/SLAMonitor.java
new file mode 100644
index 0000000000..884f563252
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/SLAMonitor.java
@@ -0,0 +1,21 @@
+package io.kestra.core.models.flows.sla;
+
+import io.kestra.core.models.HasUID;
+import io.kestra.core.utils.IdUtils;
+import lombok.Builder;
+import lombok.Getter;
+
+import java.time.Instant;
+
+@Builder
+@Getter
+public class SLAMonitor implements HasUID {
+    String executionId;
+    String slaId;
+    Instant deadline;
+
+    @Override
+    public String uid() {
+        return IdUtils.fromParts(executionId, slaId);
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/SLAMonitorStorage.java b/core/src/main/java/io/kestra/core/models/flows/sla/SLAMonitorStorage.java
new file mode 100644
index 0000000000..12adeb18ec
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/SLAMonitorStorage.java
@@ -0,0 +1,12 @@
+package io.kestra.core.models.flows.sla;
+
+import java.time.Instant;
+import java.util.function.Consumer;
+
+public interface SLAMonitorStorage {
+    void save(SLAMonitor slaMonitor);
+
+    void purge(String executionId);
+
+    void processExpired(Instant now, Consumer<SLAMonitor> consumer);
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/Violation.java b/core/src/main/java/io/kestra/core/models/flows/sla/Violation.java
new file mode 100644
index 0000000000..5330baebbc
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/Violation.java
@@ -0,0 +1,6 @@
+package io.kestra.core.models.flows.sla;
+
+import java.util.Map;
+
+public record Violation(String slaId, SLA.Behavior behavior, Map<String, Object> labels, String reason) {
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/types/ExecutionConditionSLA.java b/core/src/main/java/io/kestra/core/models/flows/sla/types/ExecutionConditionSLA.java
new file mode 100644
index 0000000000..19b2ca7817
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/types/ExecutionConditionSLA.java
@@ -0,0 +1,36 @@
+package io.kestra.core.models.flows.sla.types;
+
+import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.sla.ExecutionChangedSLA;
+import io.kestra.core.models.flows.sla.SLA;
+import io.kestra.core.models.flows.sla.Violation;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.utils.TruthUtils;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.util.Optional;
+
+@SuperBuilder
+@Getter
+@NoArgsConstructor
+public class ExecutionConditionSLA extends SLA implements ExecutionChangedSLA {
+    @NotNull
+    @NotEmpty
+    private String condition;
+
+    @Override
+    public Optional<Violation> evaluate(RunContext runContext, Execution execution) throws InternalException {
+        String result = runContext.render(this.getCondition());
+        if (TruthUtils.isTruthy(result)) {
+            String reason = "execution condition violation: " + this.getCondition() + ".";
+            return Optional.of(new Violation(this.getId(), this.getBehavior(), this.getLabels(), reason));
+        }
+
+        return Optional.empty();
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/models/flows/sla/types/MaxDurationSLA.java b/core/src/main/java/io/kestra/core/models/flows/sla/types/MaxDurationSLA.java
new file mode 100644
index 0000000000..ae1f00a6a5
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/flows/sla/types/MaxDurationSLA.java
@@ -0,0 +1,36 @@
+package io.kestra.core.models.flows.sla.types;
+
+import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.sla.ExecutionMonitoringSLA;
+import io.kestra.core.models.flows.sla.SLA;
+import io.kestra.core.models.flows.sla.Violation;
+import io.kestra.core.runners.RunContext;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Optional;
+
+@SuperBuilder
+@Getter
+@NoArgsConstructor
+public class MaxDurationSLA extends SLA implements ExecutionMonitoringSLA {
+    @NotNull
+    private Duration duration;
+
+    @Override
+    public Optional<Violation> evaluate(RunContext runContext, Execution execution) throws InternalException {
+        Duration executionDuration = Duration.between(execution.getState().getStartDate(), Instant.now());
+        if (executionDuration.compareTo(this.getDuration()) > 0) {
+            String reason = "execution duration of " + executionDuration.truncatedTo(ChronoUnit.MILLIS) + " exceed the maximum duration of " + this.getDuration() + ".";
+            return Optional.of(new Violation(this.getId(), this.getBehavior(), this.getLabels(), reason));
+        }
+
+        return Optional.empty();
+    }
+}
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 b096f98999..c1c7e3b5e4 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -3,9 +3,11 @@
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.metrics.MetricRegistry;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.flows.sla.Violation;
 import io.kestra.core.models.tasks.ExecutableTask;
 import io.kestra.core.models.tasks.ExecutionUpdatableTask;
 import io.kestra.core.models.tasks.FlowableTask;
@@ -15,17 +17,19 @@
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
-import io.kestra.core.services.ConditionService;
-import io.kestra.core.services.ExecutionService;
-import io.kestra.core.services.LogService;
-import io.kestra.core.services.WorkerGroupService;
+import io.kestra.core.queues.QueueException;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.services.*;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.core.utils.MapUtils;
 import io.kestra.plugin.core.flow.Pause;
 import io.kestra.plugin.core.flow.Subflow;
 import io.kestra.plugin.core.flow.WaitFor;
 import io.kestra.plugin.core.flow.WorkingDirectory;
 import io.micronaut.context.ApplicationContext;
 import jakarta.inject.Inject;
+import jakarta.inject.Named;
 import jakarta.inject.Singleton;
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
@@ -75,6 +79,13 @@ public class ExecutorService {
     @Inject
     private WorkerGroupService workerGroupService;
 
+    @Inject
+    private SLAService slaService;
+
+    @Inject
+    @Named(QueueFactoryInterface.KILL_NAMED)
+    protected QueueInterface<ExecutionKilled> killQueue;
+
     protected FlowExecutorInterface flowExecutorInterface() {
         // bean is injected late, so we need to wait
         if (this.flowExecutorInterface == null) {
@@ -1036,4 +1047,79 @@ public void log(Logger log, Boolean in, ExecutionKilledExecution value) {
             value
         );
     }
+
+    /**
+     * Handle flow ExecutionChangedSLA on an executor.
+     * If there are SLA violations, it will take care of updating the execution based on the SLA behavior.
+     * @see #processViolation(RunContext, Executor, Violation)
+     * <p>
+     * WARNING: ATM, only the first violation will update the execution.
+     */
+    public Executor handleExecutionChangedSLA(Executor executor) throws QueueException {
+        if (ListUtils.isEmpty(executor.getFlow().getSla()) || executor.getExecution().getState().isTerminated()) {
+            return executor;
+        }
+
+        RunContext runContext = runContextFactory.of(executor.getFlow(), executor.getExecution());
+        List<Violation> violations = slaService.evaluateExecutionChangedSLA(runContext, executor.getFlow(), executor.getExecution());
+        if (!violations.isEmpty()) {
+            // For now, we only consider the first violation to be capable of updating the execution.
+            // Other violations would only be logged.
+            Violation violation = violations.getFirst();
+            return processViolation(runContext, executor, violation);
+        }
+
+        return executor;
+    }
+
+    /**
+     * Process an SLA violation on an executor:
+     * - If behavior is FAIL or CANCEL: kill the execution, then return it with the new state.
+     * - If behavior is NONE: do nothing and return an unmodified executor.
+     * <p>
+     * Then, if there are labels, they are added to the SLA (modifying the executor)
+     */
+    public Executor processViolation(RunContext runContext, Executor executor, Violation violation) throws QueueException {
+        Execution newExecution = switch (violation.behavior()) {
+            case FAIL -> {
+                runContext.logger().error("Flow fail due to SLA '{}' violated: {}", violation.slaId(), violation.reason());
+                yield markAs(executor.getExecution(), State.Type.FAILED);
+            }
+            case CANCEL -> markAs(executor.getExecution(), State.Type.CANCELLED);
+            case NONE -> executor.getExecution();
+        };
+
+        if (!MapUtils.isEmpty(violation.labels())) {
+            List<Label> labels = new ArrayList<>(newExecution.getLabels());
+            violation.labels().forEach((key, value) -> labels.add(new Label(key, String.valueOf(value))));
+            newExecution = newExecution.withLabels(labels);
+        }
+
+        return executor.withExecution(newExecution, "SLAViolation");
+    }
+
+    private Execution markAs(Execution execution, State.Type state) throws QueueException {
+        Execution newExecution = execution.findLastNotTerminated()
+            .map(taskRun -> {
+                try {
+                    return execution.withTaskRun(taskRun.withState(state));
+                } catch (InternalException e) {
+                    // in case we cannot update the last not terminated task run, we ignore it
+                    return execution;
+                }
+            })
+            .orElse(execution)
+            .withState(state);
+
+        killQueue.emit(ExecutionKilledExecution
+            .builder()
+            .state(ExecutionKilled.State.REQUESTED)
+            .executionId(execution.getId())
+            .isOnKillCascade(false) // TODO we may offer the choice to the user here
+            .tenantId(execution.getTenantId())
+            .build()
+        );
+
+        return newExecution;
+    }
 }
diff --git a/core/src/main/java/io/kestra/core/services/SLAService.java b/core/src/main/java/io/kestra/core/services/SLAService.java
new file mode 100644
index 0000000000..ed54c7660f
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/services/SLAService.java
@@ -0,0 +1,54 @@
+package io.kestra.core.services;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.sla.ExecutionChangedSLA;
+import io.kestra.core.models.flows.sla.SLA;
+import io.kestra.core.models.flows.sla.Violation;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.utils.ListUtils;
+import jakarta.inject.Singleton;
+
+import java.util.List;
+import java.util.Optional;
+
+@Singleton
+public class SLAService {
+
+    /**
+     * Evaluate execution changed SLA of a flow for an execution.
+     * Each violated SLA will be logged.
+     */
+    public List<Violation> evaluateExecutionChangedSLA(RunContext runContext, Flow flow, Execution execution) {
+        return ListUtils.emptyOnNull(flow.getSla()).stream()
+            .filter(ExecutionChangedSLA.class::isInstance)
+            .map(
+                sla -> {
+                    try {
+                        return sla.evaluate(runContext, execution);
+                    } catch (Exception e) {
+                        runContext.logger().error("Ignoring SLA '{}' because of the error: {}", sla.getId(), e.getMessage(), e);
+                        return Optional.<Violation>empty();
+                    }
+                }
+            )
+            .flatMap(violation -> violation.stream())
+            .peek(violation -> runContext.logger().warn("SLA '{}' violated: {}", violation.slaId(), violation.reason()))
+            .toList();
+    }
+
+    /**
+     * Evaluate a single SLA for an execution.
+     * Each violated SLA will be logged.
+     */
+    public Optional<Violation> evaluateExecutionMonitoringSLA(RunContext runContext, Execution execution, SLA sla) {
+        try {
+            Optional<Violation> maybeViolation = sla.evaluate(runContext, execution);
+            maybeViolation.ifPresent(violation -> runContext.logger().warn("SLA '{}' violated: {}", violation.slaId(), violation.reason()));
+            return maybeViolation;
+        } catch (Exception e) {
+            runContext.logger().error("Ignoring SLA '{}' because of the error: {}", sla.getId(), e.getMessage(), e);
+            return Optional.empty();
+        }
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/utils/MapUtils.java b/core/src/main/java/io/kestra/core/utils/MapUtils.java
index 29e262acbe..3fd0bdd9f5 100644
--- a/core/src/main/java/io/kestra/core/utils/MapUtils.java
+++ b/core/src/main/java/io/kestra/core/utils/MapUtils.java
@@ -141,6 +141,13 @@ public static <K, V> Map<K, V> emptyOnNull(Map<K, V> map) {
         return map == null ? new HashMap<>() : map;
     }
 
+    /**
+     * Utility method that returns true if the map is null or empty.
+     */
+    public static boolean isEmpty(Map<?, ?> map) {
+        return map == null || map.isEmpty();
+    }
+
     /**
      * Creates a hash map that can hold <code>numMappings</code> entry.
      * This is a copy of the same methods available starting with Java 19.
diff --git a/core/src/test/java/io/kestra/core/utils/MapUtilsTest.java b/core/src/test/java/io/kestra/core/utils/MapUtilsTest.java
index c6565dcc99..d960471f3c 100644
--- a/core/src/test/java/io/kestra/core/utils/MapUtilsTest.java
+++ b/core/src/test/java/io/kestra/core/utils/MapUtilsTest.java
@@ -104,6 +104,14 @@ void emptyOnNull() {
         assertThat(map.size(), is(1));
     }
 
+    @Test
+    void isEmpty() {
+        assertThat(MapUtils.isEmpty(null), is(true));
+        assertThat(MapUtils.isEmpty(Collections.emptyMap()), is(true));
+        assertThat(MapUtils.isEmpty(Map.of("key", "value")), is(false));
+    }
+
+
     @Test
     void shouldReturnMapWhenNestingMapGivenFlattenMap() {
         Map<String, Object> results = MapUtils.flattenToNestedMap(Map.of(
diff --git a/jdbc-h2/src/main/java/io/kestra/runner/h2/H2SLAMonitorStorage.java b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2SLAMonitorStorage.java
new file mode 100644
index 0000000000..2b0140a892
--- /dev/null
+++ b/jdbc-h2/src/main/java/io/kestra/runner/h2/H2SLAMonitorStorage.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.h2;
+
+import io.kestra.core.models.flows.sla.SLAMonitor;
+import io.kestra.jdbc.runner.AbstractJdbcSLAMonitorStorage;
+import io.kestra.repository.h2.H2Repository;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+
+@Singleton
+@H2QueueEnabled
+public class H2SLAMonitorStorage extends AbstractJdbcSLAMonitorStorage {
+    public H2SLAMonitorStorage(@Named("slamonitor") H2Repository<SLAMonitor> repository) {
+        super(repository);
+    }
+}
diff --git a/jdbc-h2/src/main/resources/migrations/h2/V1_23__sla_monitor.sql b/jdbc-h2/src/main/resources/migrations/h2/V1_23__sla_monitor.sql
new file mode 100644
index 0000000000..90e4cf3692
--- /dev/null
+++ b/jdbc-h2/src/main/resources/migrations/h2/V1_23__sla_monitor.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS sla_monitor (
+    "key" VARCHAR(250) NOT NULL PRIMARY KEY,
+    "value" TEXT NOT NULL,
+    "execution_id" VARCHAR(150) NOT NULL GENERATED ALWAYS AS (JQ_STRING("value", '.executionId')) STORED,
+    "sla_id" VARCHAR(150) NOT NULL GENERATED ALWAYS AS (JQ_STRING("value", '.slaId')) STORED,
+    "deadline" TIMESTAMP NOT NULL GENERATED ALWAYS AS (PARSEDATETIME(JQ_STRING("value", '.deadline'), 'yyyy-MM-dd''T''HH:mm:ss.SSSXXX')) STORED
+);
+
+CREATE INDEX IF NOT EXISTS sla_monitor__deadline ON sla_monitor ("deadline");
+CREATE INDEX IF NOT EXISTS sla_monitor__execution_id ON sla_monitor ("execution_id");
\ No newline at end of file
diff --git a/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlSLAMonitorStorage.java b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlSLAMonitorStorage.java
new file mode 100644
index 0000000000..fbe7d0ceed
--- /dev/null
+++ b/jdbc-mysql/src/main/java/io/kestra/runner/mysql/MysqlSLAMonitorStorage.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.mysql;
+
+import io.kestra.core.models.flows.sla.SLAMonitor;
+import io.kestra.jdbc.runner.AbstractJdbcSLAMonitorStorage;
+import io.kestra.repository.mysql.MysqlRepository;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+
+@Singleton
+@MysqlQueueEnabled
+public class MysqlSLAMonitorStorage  extends AbstractJdbcSLAMonitorStorage {
+    public MysqlSLAMonitorStorage(@Named("slamonitor") MysqlRepository<SLAMonitor> repository) {
+        super(repository);
+    }
+}
diff --git a/jdbc-mysql/src/main/resources/migrations/mysql/V1_23__sla_monitor.sql b/jdbc-mysql/src/main/resources/migrations/mysql/V1_23__sla_monitor.sql
new file mode 100644
index 0000000000..6a197ec7bf
--- /dev/null
+++ b/jdbc-mysql/src/main/resources/migrations/mysql/V1_23__sla_monitor.sql
@@ -0,0 +1,9 @@
+CREATE TABLE IF NOT EXISTS sla_monitor (
+    `key` VARCHAR(250) NOT NULL PRIMARY KEY,
+    `value` JSONB NOT NULL,
+    `execution_id` VARCHAR(150) NOT NULL GENERATED ALWAYS AS (value ->> '$.executionId') STORED,
+    `sla_id` VARCHAR(150) NOT NULL GENERATED ALWAYS AS (value ->> '$.slaId') STORED,
+    `deadline` DATETIME(6) NOT NULL GENERATED ALWAYS AS (STR_TO_DATE(value ->> '$.deadline' , '%Y-%m-%dT%H:%i:%s.%fZ')) STORED,
+    INDEX ix_deadline (deadline),
+    INDEX ix_execution_id (execution_id)
+);
\ No newline at end of file
diff --git a/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresSLAMonitorStorage.java b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresSLAMonitorStorage.java
new file mode 100644
index 0000000000..150f92d912
--- /dev/null
+++ b/jdbc-postgres/src/main/java/io/kestra/runner/postgres/PostgresSLAMonitorStorage.java
@@ -0,0 +1,15 @@
+package io.kestra.runner.postgres;
+
+import io.kestra.core.models.flows.sla.SLAMonitor;
+import io.kestra.jdbc.runner.AbstractJdbcSLAMonitorStorage;
+import io.kestra.repository.postgres.PostgresRepository;
+import jakarta.inject.Named;
+import jakarta.inject.Singleton;
+
+@Singleton
+@PostgresQueueEnabled
+public class PostgresSLAMonitorStorage  extends AbstractJdbcSLAMonitorStorage {
+    public PostgresSLAMonitorStorage(@Named("slamonitor") PostgresRepository<SLAMonitor> repository) {
+        super(repository);
+    }
+}
diff --git a/jdbc-postgres/src/main/resources/migrations/postgres/V1_23__sla_monitor.sql b/jdbc-postgres/src/main/resources/migrations/postgres/V1_23__sla_monitor.sql
new file mode 100644
index 0000000000..6c22281496
--- /dev/null
+++ b/jdbc-postgres/src/main/resources/migrations/postgres/V1_23__sla_monitor.sql
@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS sla_monitor (
+    key VARCHAR(250) NOT NULL PRIMARY KEY,
+    value JSONB NOT NULL,
+    execution_id VARCHAR(150) NOT NULL GENERATED ALWAYS AS (value ->> 'executionId') STORED,
+    sla_id VARCHAR(150) NOT NULL GENERATED ALWAYS AS (value ->> 'slaId') STORED,
+    deadline TIMESTAMPTZ NOT NULL GENERATED ALWAYS AS (PARSE_ISO8601_DATETIME(value ->> 'deadline')) STORED
+);
+
+CREATE INDEX IF NOT EXISTS sla_monitor__deadline ON sla_monitor (deadline);
+CREATE INDEX IF NOT EXISTS sla_monitor__execution_id ON sla_monitor (execution_id);
\ No newline at end of file
diff --git a/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java b/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java
index 3d6ab6c0cf..636a8f34c8 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/JdbcTableConfigsFactory.java
@@ -5,6 +5,7 @@
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.executions.MetricEntry;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.sla.SLAMonitor;
 import io.kestra.core.models.templates.Template;
 import io.kestra.core.models.topologies.FlowTopology;
 import io.kestra.core.models.triggers.Trigger;
@@ -117,6 +118,12 @@ public InstantiableJdbcTableConfig executionQueued() {
         return new InstantiableJdbcTableConfig("executionqueued", ExecutionQueued.class, "execution_queued");
     }
 
+    @Bean
+    @Named("slamonitor")
+    public InstantiableJdbcTableConfig slaMonitor() {
+        return new InstantiableJdbcTableConfig("slamonitor", SLAMonitor.class, "sla_monitor");
+    }
+
     public static class InstantiableJdbcTableConfig extends JdbcTableConfig {
         public InstantiableJdbcTableConfig(String name, @Nullable Class<?> cls, String table) {
             super(name, cls, table);
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSLAMonitorStorage.java b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSLAMonitorStorage.java
new file mode 100644
index 0000000000..6a24fdc2c7
--- /dev/null
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSLAMonitorStorage.java
@@ -0,0 +1,61 @@
+package io.kestra.jdbc.runner;
+
+import io.kestra.core.models.flows.sla.SLAMonitor;
+import io.kestra.core.models.flows.sla.SLAMonitorStorage;
+import io.kestra.jdbc.repository.AbstractJdbcRepository;
+import org.jooq.DSLContext;
+import org.jooq.Field;
+import org.jooq.impl.DSL;
+
+import java.time.Instant;
+import java.util.Map;
+import java.util.function.Consumer;
+
+public abstract class AbstractJdbcSLAMonitorStorage extends AbstractJdbcRepository implements SLAMonitorStorage {
+    protected io.kestra.jdbc.AbstractJdbcRepository<SLAMonitor> jdbcRepository;
+
+    protected AbstractJdbcSLAMonitorStorage(io.kestra.jdbc.AbstractJdbcRepository<SLAMonitor> jdbcRepository) {
+        this.jdbcRepository = jdbcRepository;
+    }
+
+    @Override
+    public void save(SLAMonitor slaMonitor) {
+        this.jdbcRepository
+            .getDslContextWrapper()
+            .transaction(configuration -> {
+                DSLContext context = DSL.using(configuration);
+                Map<Field<Object>, Object> fields = this.jdbcRepository.persistFields(slaMonitor);
+                this.jdbcRepository.persist(slaMonitor, context, fields);
+            });
+    }
+
+    @Override
+    public void purge(String executionId) {
+        this.jdbcRepository
+            .getDslContextWrapper()
+            .transaction(configuration -> {
+                DSLContext context = DSL.using(configuration);
+                context.delete(this.jdbcRepository.getTable())
+                    .where(field("execution_id").eq(executionId))
+                    .execute();
+            });
+    }
+
+    @Override
+    public void processExpired(Instant date, Consumer<SLAMonitor> consumer) {
+        this.jdbcRepository
+            .getDslContextWrapper()
+            .transaction(configuration -> {
+                DSLContext context = DSL.using(configuration);
+                var select = context.select()
+                    .from(this.jdbcRepository.getTable())
+                    .where(field("deadline").lt(date));
+
+                this.jdbcRepository.fetch(select)
+                    .forEach(slaMonitor -> {
+                        consumer.accept(slaMonitor);
+                        jdbcRepository.delete(slaMonitor);
+                    });
+            });
+    }
+}
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 59de96dccf..6944c97033 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -10,6 +10,7 @@
 import io.kestra.core.models.executions.statistics.ExecutionCount;
 import io.kestra.core.models.flows.*;
 import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.sla.*;
 import io.kestra.core.models.tasks.ExecutableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.topologies.FlowTopology;
@@ -25,10 +26,7 @@
 import io.kestra.core.server.ServiceStateChangeEvent;
 import io.kestra.core.services.*;
 import io.kestra.core.topologies.FlowTopologyService;
-import io.kestra.core.utils.Await;
-import io.kestra.core.utils.Either;
-import io.kestra.core.utils.IdUtils;
-import io.kestra.core.utils.TruthUtils;
+import io.kestra.core.utils.*;
 import io.kestra.jdbc.JdbcMapper;
 import io.kestra.jdbc.repository.AbstractJdbcExecutionRepository;
 import io.kestra.jdbc.repository.AbstractJdbcFlowTopologyRepository;
@@ -161,6 +159,12 @@ public class JdbcExecutor implements ExecutorInterface, Service {
     @Inject
     private LogService logService;
 
+    @Inject
+    private SLAMonitorStorage slaMonitorStorage;
+
+    @Inject
+    private SLAService slaService;
+
     private final FlowRepositoryInterface flowRepository;
 
     private final JdbcServiceLivenessCoordinator serviceLivenessCoordinator;
@@ -223,6 +227,13 @@ public void run() {
             TimeUnit.SECONDS
         );
 
+        ScheduledFuture<?> scheduledSLAMonitorFuture = scheduledDelay.scheduleAtFixedRate(
+            this::executionSLAMonitor,
+            0,
+            1,
+            TimeUnit.SECONDS
+        );
+
         // look at exceptions on the scheduledDelay thread
         Thread.ofVirtual().name("jdbc-delay-exception-watcher").start(
             () -> {
@@ -240,6 +251,23 @@ public void run() {
             }
         );
 
+        // look at exceptions on the scheduledSLAMonitorFuture thread
+        Thread.ofVirtual().name("jdbc-sla-monitor-exception-watcher").start(
+            () -> {
+                Await.until(scheduledSLAMonitorFuture::isDone);
+
+                try {
+                    scheduledSLAMonitorFuture.get();
+                } catch (ExecutionException | InterruptedException | CancellationException e) {
+                    if (e.getCause() != null && e.getCause().getClass() != CannotCreateTransactionException.class) {
+                        log.error("Executor fatal exception in the scheduledSLAMonitor thread", e);
+                        close();
+                        KestraContext.getContext().shutdown();
+                    }
+                }
+            }
+        );
+
         this.receiveCancellations.addFirst(flowQueue.receive(
             FlowTopology.class,
             either -> {
@@ -379,6 +407,21 @@ private void executionQueue(Either<Execution, DeserializationException> either)
                     );
                 }
 
+                // create an SLA monitor if needed
+                if (execution.getState().getCurrent() == State.Type.CREATED && !ListUtils.isEmpty(flow.getSla())) {
+                    List<SLAMonitor> monitors = flow.getSla().stream()
+                        .filter(ExecutionMonitoringSLA.class::isInstance)
+                        .map( ExecutionMonitoringSLA.class::cast)
+                        .map(sla -> SLAMonitor.builder()
+                            .executionId(execution.getId())
+                            .slaId(((SLA) sla).getId())
+                            .deadline(execution.getState().getStartDate().plus(sla.getDuration()))
+                            .build()
+                        )
+                        .toList();
+                    monitors.forEach(monitor -> slaMonitorStorage.save(monitor));
+                }
+
                 // queue execution if needed (limit concurrency)
                 if (execution.getState().getCurrent() == State.Type.CREATED && flow.getConcurrency() != null) {
                     ExecutionCount count = executionRepository.executionCounts(
@@ -410,6 +453,9 @@ private void executionQueue(Either<Execution, DeserializationException> either)
                     }
                 }
 
+                // handle SLA
+                executor = executorService.handleExecutionChangedSLA(executor);
+
                 // process the execution
                 if (log.isDebugEnabled()) {
                     executorService.log(log, true, executor);
@@ -872,6 +918,11 @@ private void toExecution(Executor executor, boolean ignoreFailure) {
                         subflowExecutionStorage.delete(subflowExecution);
                     });
 
+                // purge SLA monitors
+                if (!ListUtils.isEmpty(executor.getFlow().getSla()) && executor.getFlow().getSla().stream().anyMatch(ExecutionMonitoringSLA.class::isInstance)) {
+                    slaMonitorStorage.purge(executor.getExecution().getId());
+                }
+
                 // check if there exist a queued execution and submit it to the execution queue
                 if (executor.getFlow().getConcurrency() != null && executor.getFlow().getConcurrency().getBehavior() == Concurrency.Behavior.QUEUE) {
                     executionQueuedStorage.pop(executor.getFlow().getTenantId(),
@@ -982,6 +1033,45 @@ else if (executionDelay.getDelayType().equals(ExecutionDelay.DelayType.CONTINUE_
         });
     }
 
+    private void executionSLAMonitor() {
+        if (shutdown.get()) {
+            return;
+        }
+
+        slaMonitorStorage.processExpired(Instant.now(), slaMonitor -> {
+            Executor result = executionRepository.lock(slaMonitor.getExecutionId(), pair -> {
+                Executor executor = new Executor(pair.getLeft(), null);
+                Flow flow = flowRepository.findByExecution(pair.getLeft());
+                Optional<SLA> sla = flow.getSla().stream().filter(s -> s.getId().equals(slaMonitor.getSlaId())).findFirst();
+                if (sla.isEmpty()) {
+                    // this can happen in case the flow has been updated and the SLA removed
+                    log.debug("Cannot find the SLA '{}' if the flow for execution '{}', ignoring it.", slaMonitor.getSlaId(), slaMonitor.getExecutionId());
+                    return null;
+                }
+
+                try {
+                    log.info("Processing expired SLA monitor '{}' for execution '{}'.", slaMonitor.getSlaId(), slaMonitor.getExecutionId());
+                    RunContext runContext = runContextFactory.of(executor.getFlow(), executor.getExecution());
+                    Optional<Violation> violation = slaService.evaluateExecutionMonitoringSLA(runContext, executor.getExecution(), sla.get());
+                    if (violation.isPresent()) { // should always be true
+                        executor = executorService.processViolation(runContext, executor, violation.get());
+                    }
+                } catch (Exception e) {
+                    executor = handleFailedExecutionFromExecutor(executor, e);
+                }
+
+                return Pair.of(
+                    executor,
+                    pair.getRight()
+                );
+            });
+
+            if (result != null) {
+                this.toExecution(result);
+            }
+        });
+    }
+
     private boolean deduplicateNexts(Execution execution, ExecutorState executorState, List<TaskRun> taskRuns) {
         return taskRuns
             .stream()