feat: Add DagProcessor and Triggerer roles

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Added
 
 - Add a flag to determine if database initialization steps should be executed ([#669]).
+- Add new roles for dag-processor and triggerer processes ([#679]).
 
 ### Fixed
 
@@ -19,6 +20,7 @@
 [#668]: https://github.com/stackabletech/airflow-operator/pull/668
 [#669]: https://github.com/stackabletech/airflow-operator/pull/669
 [#678]: https://github.com/stackabletech/airflow-operator/pull/678
+[#679]: https://github.com/stackabletech/airflow-operator/pull/679
 [#683]: https://github.com/stackabletech/airflow-operator/pull/683
 
 ## [25.7.0] - 2025-07-23

docs/modules/airflow/examples/getting_started/code/airflow.yaml

@@ -19,8 +19,16 @@ spec:
   celeryExecutors:
     roleGroups:
       default:
-        replicas: 2
+        replicas: 1
   schedulers:
     roleGroups:
       default:
         replicas: 1
+  dagProcessors:
+    roleGroups:
+      default:
+        replicas: 1
+  triggerers:
+    roleGroups:
+      default:
+        replicas: 1

docs/modules/airflow/examples/getting_started/code/getting_started.sh

@@ -88,6 +88,8 @@ echo "Awaiting Airflow rollout finish ..."
 kubectl rollout status --watch --timeout=5m statefulset/airflow-webserver-default
 kubectl rollout status --watch --timeout=5m statefulset/airflow-worker-default
 kubectl rollout status --watch --timeout=5m statefulset/airflow-scheduler-default
+kubectl rollout status --watch --timeout=5m statefulset/airflow-dagprocessor-default
+kubectl rollout status --watch --timeout=5m statefulset/airflow-triggerer-default
 # end::watch-airflow-rollout[]
 
 echo "Starting port-forwarding of port 8080"

docs/modules/airflow/examples/getting_started/code/getting_started.sh.j2

@@ -88,6 +88,8 @@ echo "Awaiting Airflow rollout finish ..."
 kubectl rollout status --watch --timeout=5m statefulset/airflow-webserver-default
 kubectl rollout status --watch --timeout=5m statefulset/airflow-worker-default
 kubectl rollout status --watch --timeout=5m statefulset/airflow-scheduler-default
+kubectl rollout status --watch --timeout=5m statefulset/airflow-dagprocessor-default
+kubectl rollout status --watch --timeout=5m statefulset/airflow-triggerer-default
 # end::watch-airflow-rollout[]
 
 echo "Starting port-forwarding of port 8080"

docs/modules/airflow/pages/getting_started/first_steps.adoc

@@ -38,11 +38,15 @@ NOTE: The admin user is disabled if you use a non-default authentication mechani
 
 == Airflow
 
-An Airflow cluster is made of up three components:
+An Airflow cluster is made of up five components, two of which are optional:
 
 * `webserver`: this provides the main UI for user-interaction
 * `executors`: the CeleryExecutor or KubernetesExecutor nodes over which the job workload is distributed by the scheduler
 * `scheduler`: responsible for triggering jobs and persisting their metadata to the backend database
+* `dagProcessors`: (Optional) responsible for monitoring, parsing and preparing DAGs for processing.
+If this role is not specified then the process will be started as a scheduler subprocess (Airflow 2.x), or as a standalone process in the same container as the scheduler (Airflow 3.x+)
+* `triggerers`: (Optional) DAGs making use of deferrable operators can be used together with one or more triggerer processes to free up worker slots.
+This deferral process is also useful for providing a measure of high availability
 
 Create a file named `airflow.yaml` with the following contents:
 
@@ -92,7 +96,9 @@ airflow-redis-master            1/1     16m
 airflow-redis-replicas          1/1     16m
 airflow-scheduler-default       1/1     11m
 airflow-webserver-default       1/1     11m
-airflow-celery-executor-default 2/2     11m
+airflow-celery-executor-default 1/1     11m
+airflow-dagprocessor-default    1/1     11m
+airflow-triggerer-default       1/1     11m
 ----
 
 When the Airflow cluster has been created and the database is initialized, Airflow can be opened in the

docs/modules/airflow/pages/index.adoc

@@ -3,11 +3,11 @@
 :keywords: Stackable Operator, Apache Airflow, Kubernetes, k8s, operator, job pipeline, scheduler, ETL
 :airflow: https://airflow.apache.org/
 :dags: https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html
-:k8s-crs: https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/
 :github: https://github.com/stackabletech/airflow-operator/
 :crd: {crd-docs-base-url}/airflow-operator/{crd-docs-version}/
 :crd-airflowcluster: {crd-docs}/airflow.stackable.tech/airflowcluster/v1alpha1/
 :feature-tracker: https://features.stackable.tech/unified
+:deferrable-operators: https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html#deferrable-operators-triggers
 
 [.link-bar]
 * {github}[GitHub {external-link-icon}^]
@@ -27,7 +27,8 @@ It guides you through installing the operator alongside a PostgreSQL database an
 === Custom resources
 
 The AirflowCluster is the resource for the configuration of the Airflow instance.
-The resource defines three xref:concepts:roles-and-role-groups.adoc[roles]: `webserver`, `worker` and `scheduler` (the `worker` role is embedded within `spec.celeryExecutors`: this is described in the next section).
+The resource defines five xref:concepts:roles-and-role-groups.adoc[roles]: `webserver`, `worker`, `scheduler`, `dagProcessor` and `triggerer` (the `worker` role is embedded within `spec.celeryExecutors`: this is described in the next section).
+The `dagProcessor` and `triggerer` roles are optional.
 The various configuration options are explained in the xref:usage-guide/index.adoc[].
 It helps you tune your cluster to your needs by configuring xref:usage-guide/storage-resources.adoc[resource usage], xref:usage-guide/security.adoc[security], xref:usage-guide/logging.adoc[logging] and more.
 
@@ -70,6 +71,17 @@ kubernetesExecutors:
       ...
 ----
 
+=== DAG-Processors
+
+In Airflow 2.x, a DAG-Processor can be started either as a standalone process or a subprocess within the scheduler component.
+For Airflow 3.x+ it _must_ be started as a standalone process, either in a separate container or in the scheduler container.
+In each case the default will be applied (subprocess or combined in the scheduler container) if the role is not specified.
+
+=== Triggerers
+
+DAGs using deferrable operators can be combined with the triggerer component to free up worker slots and/or provide high availability.
+For more information, please refer to the {deferrable-operators}[documentation {external-link-icon}^].
+
 === Kubernetes resources
 
 Based on the custom resources you define, the operator creates ConfigMaps, StatefulSets and Services.

docs/modules/airflow/pages/usage-guide/storage-resources.adoc

@@ -1,13 +1,13 @@
 = Resource Requests
-:description: Find out about minimal HA Airflow requirements for CPU and memory, with defaults for schedulers, Celery executors, webservers using Kubernetes resource limits.
+:description: Find out about minimal HA Airflow requirements for CPU and memory, with defaults for schedulers, Celery executors, webservers, dagProcessors and triggerers using Kubernetes resource limits.
 
 include::home:concepts:stackable_resource_requests.adoc[]
 
-A minimal HA setup consisting of 2 schedulers, 2 workers and 2 webservers has the following https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/[resource requirements]:
+A minimal HA setup consisting of 2 schedulers, 2 workers, 2 webservers, 2 dag-processors and 1 triggerer has the following https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/[resource requirements]:
 
-* `8700m` CPU request
-* `17400m` CPU limit
-* `15872Mi` memory request and limit
+* `11600` CPU request
+* `23200` CPU limit
+* `18432Mi` memory request and limit
 
 Corresponding to the values above, the operator uses the following resource defaults:
 
@@ -22,6 +22,9 @@ spec:
           max: "2"
         memory:
           limit: 1Gi
+    roleGroups:
+      default:
+        replicas: 2
   celeryExecutors:
     config:
       resources:
@@ -30,6 +33,9 @@ spec:
           max: "2"
         memory:
           limit: 3Gi
+    roleGroups:
+      default:
+        replicas: 2
   webservers:
     config:
       resources:
@@ -38,4 +44,29 @@ spec:
           max: "2"
         memory:
           limit: 3Gi
+    roleGroups:
+      default:
+        replicas: 2
+  dagProcessors:
+    config:
+      resources:
+        cpu:
+          min: "1"
+          max: "2"
+        memory:
+          limit: 1Gi
+    roleGroups:
+      default:
+        replicas: 2
+  triggerers:
+    config:
+      resources:
+        cpu:
+          min: "1"
+          max: "2"
+        memory:
+          limit: 1Gi
+    roleGroups:
+      default:
+        replicas: 1
 ----

rust/operator-binary/src/airflow_controller.rs

@@ -1156,7 +1156,10 @@ fn build_server_rolegroup_statefulset(
                 AirflowRole::Scheduler => {
                     "OrderedReady" // Scheduler pods should start after another, since part of their startup phase is initializing the database, see crd/src/lib.rs
                 }
-                AirflowRole::Webserver | AirflowRole::Worker => "Parallel",
+                AirflowRole::Webserver
+                | AirflowRole::Worker
+                | AirflowRole::DagProcessor
+                | AirflowRole::Triggerer => "Parallel",
             }
             .to_string(),
         ),

rust/operator-binary/src/crd/mod.rs

@@ -205,17 +205,25 @@ pub mod versioned {
         #[serde(default)]
         pub cluster_operation: ClusterOperation,
 
-        /// The `webserver` role provides the main UI for user interaction.
+        /// The `webservers` role provides the main UI for user interaction.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         pub webservers: Option<Role<AirflowConfigFragment, v1alpha1::WebserverRoleConfig>>,
 
-        /// The `scheduler` is responsible for triggering jobs and persisting their metadata to the backend database.
+        /// The `schedulers` is responsible for triggering jobs and persisting their metadata to the backend database.
         /// Jobs are scheduled on the workers/executors.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         pub schedulers: Option<Role<AirflowConfigFragment>>,
 
         #[serde(flatten)]
         pub executor: AirflowExecutor,
+
+        /// The `dagProcessors` role runs the DAG processor routine for DAG preparation.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub dag_processors: Option<Role<AirflowConfigFragment>>,
+
+        /// The `triggerers` role runs the triggerer process for use with deferrable DAG operators.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub triggerers: Option<Role<AirflowConfigFragment>>,
     }
 
     #[derive(Clone, Deserialize, Debug, JsonSchema, PartialEq, Serialize)]
@@ -342,7 +350,10 @@ impl v1alpha1::AirflowCluster {
     pub fn group_listener_name(&self, role: &AirflowRole) -> Option<String> {
         match role {
             AirflowRole::Webserver => Some(role_service_name(&self.name_any(), &role.to_string())),
-            AirflowRole::Scheduler | AirflowRole::Worker => None,
+            AirflowRole::Scheduler
+            | AirflowRole::Worker
+            | AirflowRole::DagProcessor
+            | AirflowRole::Triggerer => None,
         }
     }
 
@@ -356,6 +367,8 @@ impl v1alpha1::AirflowCluster {
                 .to_owned()
                 .map(extract_role_from_webserver_config),
             AirflowRole::Scheduler => self.spec.schedulers.to_owned(),
+            AirflowRole::DagProcessor => self.spec.dag_processors.to_owned(),
+            AirflowRole::Triggerer => self.spec.triggerers.to_owned(),
             AirflowRole::Worker => {
                 if let AirflowExecutor::CeleryExecutor { config } = &self.spec.executor {
                     Some(config.clone())
@@ -416,6 +429,24 @@ impl v1alpha1::AirflowCluster {
                         roles: AirflowRole::roles(),
                     })?
             }
+            AirflowRole::DagProcessor => {
+                self.spec
+                    .dag_processors
+                    .as_ref()
+                    .context(UnknownAirflowRoleSnafu {
+                        role: role.to_string(),
+                        roles: AirflowRole::roles(),
+                    })?
+            }
+            AirflowRole::Triggerer => {
+                self.spec
+                    .triggerers
+                    .as_ref()
+                    .context(UnknownAirflowRoleSnafu {
+                        role: role.to_string(),
+                        roles: AirflowRole::roles(),
+                    })?
+            }
         };
 
         // Retrieve role resource config
@@ -564,6 +595,12 @@ pub enum AirflowRole {
 
     #[strum(serialize = "worker")]
     Worker,
+
+    #[strum(serialize = "dagprocessor")]
+    DagProcessor,
+
+    #[strum(serialize = "triggerer")]
+    Triggerer,
 }
 
 impl AirflowRole {
@@ -625,10 +662,27 @@ impl AirflowRole {
                     command.extend(vec![
                         "prepare_signal_handlers".to_string(),
                         container_debug_command(),
-                        "airflow dag-processor &".to_string(),
                         "airflow scheduler &".to_string(),
                     ]);
+                    if airflow.spec.dag_processors.is_none() {
+                        // If no dag_processors role has been specified, the
+                        // process needs to be included with the scheduler
+                        // (with 3.x there is no longer the possibility of
+                        // starting it as a subprocess, so it has to be
+                        // explicitly started *somewhere*)
+                        command.extend(vec!["airflow dag-processor &".to_string()]);
+                    }
                 }
+                AirflowRole::DagProcessor => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow dag-processor &".to_string(),
+                ]),
+                AirflowRole::Triggerer => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow triggerer &".to_string(),
+                ]),
                 AirflowRole::Worker => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
@@ -671,6 +725,16 @@ impl AirflowRole {
                         "airflow scheduler &".to_string(),
                     ]);
                 }
+                AirflowRole::DagProcessor => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow dag-processor &".to_string(),
+                ]),
+                AirflowRole::Triggerer => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow triggerer &".to_string(),
+                ]),
                 AirflowRole::Worker => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
@@ -727,6 +791,8 @@ impl AirflowRole {
             AirflowRole::Webserver => Some(HTTP_PORT),
             AirflowRole::Scheduler => None,
             AirflowRole::Worker => None,
+            AirflowRole::DagProcessor => None,
+            AirflowRole::Triggerer => None,
         }
     }
 
@@ -745,7 +811,7 @@ impl AirflowRole {
                 .webservers
                 .to_owned()
                 .map(|webserver| webserver.role_config.listener_class),
-            Self::Worker | Self::Scheduler => None,
+            Self::Worker | Self::Scheduler | Self::DagProcessor | Self::Triggerer => None,
         }
     }
 }
@@ -881,9 +947,10 @@ impl AirflowConfig {
             logging: product_logging::spec::default_logging(),
             affinity: get_affinity(cluster_name, role),
             graceful_shutdown_timeout: Some(match role {
-                AirflowRole::Webserver | AirflowRole::Scheduler => {
-                    DEFAULT_AIRFLOW_GRACEFUL_SHUTDOWN_TIMEOUT
-                }
+                AirflowRole::Webserver
+                | AirflowRole::Scheduler
+                | AirflowRole::DagProcessor
+                | AirflowRole::Triggerer => DEFAULT_AIRFLOW_GRACEFUL_SHUTDOWN_TIMEOUT,
                 AirflowRole::Worker => DEFAULT_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT,
             }),
         }
@@ -956,6 +1023,26 @@ fn default_resources(role: &AirflowRole) -> ResourcesFragment<AirflowStorageConf
                 runtime_limits: NoRuntimeLimitsFragment {},
             },
         ),
+        AirflowRole::DagProcessor => (
+            CpuLimitsFragment {
+                min: Some(Quantity("1".to_owned())),
+                max: Some(Quantity("2".to_owned())),
+            },
+            MemoryLimitsFragment {
+                limit: Some(Quantity("1Gi".to_owned())),
+                runtime_limits: NoRuntimeLimitsFragment {},
+            },
+        ),
+        AirflowRole::Triggerer => (
+            CpuLimitsFragment {
+                min: Some(Quantity("1".to_owned())),
+                max: Some(Quantity("2".to_owned())),
+            },
+            MemoryLimitsFragment {
+                limit: Some(Quantity("1Gi".to_owned())),
+                runtime_limits: NoRuntimeLimitsFragment {},
+            },
+        ),
     };
 
     ResourcesFragment {