fix: scheduler container supervises scheduler instead of dag-processor

CHANGELOG.md

@@ -42,6 +42,7 @@
   See [our internal issue](https://github.com/stackabletech/hdfs-operator/issues/626) and [the fix](https://github.com/kube-rs/kube/pull/2042) for details ([#840]).
 - Task logs are served from `BASE_LOG_FOLDER` instead of the `task` handler's `base_log_folder` at the Vector agent log directory ([#834]).
 - Avoid Python import race conditions by pre-cloning the git repo not only for Celery-based stacklets, but also for Kubernetes executor-based setups ([#844]).
+- The Airflow 3.x scheduler container now supervises the scheduler instead of the dag-processor. Before, a dead scheduler left the Pod `Running` and `Ready` with nothing scheduling DAGs ([#847]).
 
 [#814]: https://github.com/stackabletech/airflow-operator/pull/814
 [#821]: https://github.com/stackabletech/airflow-operator/pull/821
@@ -53,6 +54,7 @@
 [#838]: https://github.com/stackabletech/airflow-operator/pull/838
 [#840]: https://github.com/stackabletech/airflow-operator/pull/840
 [#844]: https://github.com/stackabletech/airflow-operator/pull/844
+[#847]: https://github.com/stackabletech/airflow-operator/pull/847
 
 ## [26.7.0] - 2026-07-21
 

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

@@ -41,7 +41,7 @@ An Airflow cluster is made up of several components, two of which are optional:
 * `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+)
+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+). We recommend specifying this role for production settings as running multiple processes in the same container complicates proper supervision.
 * `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
 

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

@@ -603,16 +603,20 @@ impl AirflowRole {
                     command.extend(vec![
                         "prepare_signal_handlers".to_string(),
                         container_debug_command(),
-                        "airflow scheduler &".to_string(),
                     ]);
                     if !has_dag_processors {
                         // 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*)
+                        // explicitly started *somewhere*).
+                        // It is started before the scheduler because the
+                        // trailing `wait_for_termination $!` binds to the
+                        // process that was backgrounded last, and that must be
+                        // the scheduler.
                         command.extend(vec!["airflow dag-processor &".to_string()]);
                     }
+                    command.extend(vec!["airflow scheduler &".to_string()]);
                 }
                 AirflowRole::DagProcessor => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
@@ -1034,9 +1038,13 @@ mod tests {
         commons::product_image_selection::ResolvedProductImage,
         versioned::test_utils::RoundtripTestData,
     };
+    use strum::IntoEnumIterator;
 
     use super::*;
-    use crate::{v1alpha1, v1alpha2};
+    use crate::{
+        controller::build::test_support::{validated_cluster, validated_cluster_with},
+        v1alpha1, v1alpha2,
+    };
 
     #[test]
     fn test_constants() {
@@ -1241,6 +1249,79 @@ mod tests {
         );
     }
 
+    /// The commands that background a process, i.e. the candidates for `$!`.
+    fn backgrounded_commands(role: &AirflowRole, cluster: &ValidatedCluster) -> Vec<String> {
+        role.get_commands(cluster)
+            .into_iter()
+            .filter(|command| command.ends_with(" &"))
+            .collect()
+    }
+
+    #[test]
+    fn every_role_backgrounds_its_own_process_last() {
+        let cluster = validated_cluster("kubernetesExecutors", "{config: {}}");
+
+        assert!(
+            cluster.image.product_version.starts_with("3."),
+            "the test cluster must run Airflow 3.x for this to test anything"
+        );
+        assert!(!cluster.has_role(&AirflowRole::DagProcessor));
+
+        for role in AirflowRole::iter() {
+            let own_process = match role {
+                AirflowRole::Webserver => "airflow api-server &",
+                AirflowRole::Scheduler => "airflow scheduler &",
+                AirflowRole::Worker => "airflow celery worker &",
+                AirflowRole::DagProcessor => "airflow dag-processor &",
+                AirflowRole::Triggerer => "airflow triggerer &",
+            };
+
+            assert_eq!(
+                backgrounded_commands(&role, &cluster).last(),
+                Some(&own_process.to_string()),
+                "{role:?} must background its own process last"
+            );
+        }
+    }
+
+    #[test]
+    fn a_scheduler_without_a_dag_processor_role_starts_the_dag_processor() {
+        let cluster = validated_cluster("kubernetesExecutors", "{config: {}}");
+        let backgrounded = backgrounded_commands(&AirflowRole::Scheduler, &cluster);
+
+        assert!(
+            backgrounded.contains(&"airflow dag-processor &".to_string()),
+            "the scheduler must start the dag-processor, but backgrounds only: {backgrounded:?}"
+        );
+    }
+
+    #[test]
+    fn a_scheduler_with_a_dag_processor_role_does_not_start_the_dag_processor() {
+        let cluster = validated_cluster_with("kubernetesExecutors", "{config: {}}", |cluster| {
+            cluster["spec"]
+                .as_mapping_mut()
+                .expect("the test CR has a spec mapping")
+                .insert(
+                    "dagProcessors".into(),
+                    serde_yaml::from_str("{config: {}, roleGroups: {default: {config: {}}}}")
+                        .expect("the dag-processor role is valid YAML"),
+                );
+        });
+
+        assert!(
+            cluster.has_role(&AirflowRole::DagProcessor),
+            "the fixture must declare a dag-processor role for this to test anything"
+        );
+
+        let backgrounded = backgrounded_commands(&AirflowRole::Scheduler, &cluster);
+
+        assert!(
+            !backgrounded.contains(&"airflow dag-processor &".to_string()),
+            "the dedicated dag-processor role runs the process, so the scheduler must not \
+             start a second one, but backgrounds: {backgrounded:?}"
+        );
+    }
+
     impl RoundtripTestData for v1alpha1::AirflowClusterSpec {
         fn roundtrip_test_data() -> Vec<Self> {
             let git_sync_section = r#"

tests/templates/kuttl/ca-cert/50-assert.yaml.j2

@@ -6,7 +6,12 @@ metadata:
 timeout: 480
 commands:
 {% if test_scenario['values']['airflow-latest'].find(",") > 0 %}
-  - script: kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ test_scenario['values']['airflow-latest'].split(',')[0] }}"
+{% set airflow_version = test_scenario['values']['airflow-latest'].split(',')[0] %}
 {% else %}
-  - script: kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ test_scenario['values']['airflow-latest'] }}"
+{% set airflow_version = test_scenario['values']['airflow-latest'] %}
 {% endif %}
+  - script: |
+      kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py \
+          --airflow-version "{{ airflow_version }}" \
+          --component scheduler \
+          --component dag_processor

tests/templates/kuttl/cluster-operation/11-assert.yaml.j2

@@ -0,0 +1,51 @@
+---
+# The scheduler process was killed in the previous step. The start script ends in
+# `wait_for_termination $!`, so the container must exit and be restarted by Kubernetes.
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: test-scheduler-is-supervised
+timeout: 120
+{% if test_scenario['values']['airflow-latest'].find(",") > 0 %}
+{% set airflow_version = test_scenario['values']['airflow-latest'].split(',')[0] %}
+{% else %}
+{% set airflow_version = test_scenario['values']['airflow-latest'] %}
+{% endif %}
+commands:
+  - script: |
+      for _ in $(seq 15); do
+          restart_count=$(kubectl -n $NAMESPACE get pod airflow-scheduler-default-0 \
+              -o jsonpath='{.status.containerStatuses[?(@.name=="airflow")].restartCount}')
+
+          if [ "${restart_count:-0}" -ge 1 ]; then
+              exit 0
+          fi
+
+          sleep 2
+      done
+
+      echo "The scheduler died but its container was not restarted."
+      echo "Kubernetes considers the Pod healthy:"
+      kubectl -n $NAMESPACE get pod airflow-scheduler-default-0
+
+      echo "Airflow does not:"
+      kubectl -n $NAMESPACE exec -i airflow-webserver-default-0 -c airflow -- python - \
+          --airflow-version "{{ airflow_version }}" \
+          --component scheduler \
+          --component dag_processor \
+          --timeout 1 \
+          < ../../../../templates/kuttl/commons/health.py || true
+
+      exit 1
+
+  # The restarted container must bring both processes back. Requiring a
+  # heartbeat that differs from the first one observed rules out a scheduler or
+  # dag-processor that reports healthy on a stale heartbeat.
+  - script: |
+      kubectl -n $NAMESPACE exec -i airflow-webserver-default-0 -c airflow -- python - \
+          --airflow-version "{{ airflow_version }}" \
+          --component scheduler \
+          --component dag_processor \
+          --heartbeat-changed \
+          --timeout 60 \
+          < ../../../../templates/kuttl/commons/health.py

tests/templates/kuttl/cluster-operation/11-kill-scheduler.yaml.j2

@@ -0,0 +1,56 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+metadata:
+  name: kill-scheduler
+timeout: 120
+{% if test_scenario['values']['airflow-latest'].find(",") > 0 %}
+{% set airflow_version = test_scenario['values']['airflow-latest'].split(',')[0] %}
+{% else %}
+{% set airflow_version = test_scenario['values']['airflow-latest'] %}
+{% endif %}
+commands:
+  - script: |
+      restart_count=$(kubectl -n $NAMESPACE get pod airflow-scheduler-default-0 \
+          -o jsonpath='{.status.containerStatuses[?(@.name=="airflow")].restartCount}')
+
+      if [ "${restart_count:-0}" -ne 0 ]; then
+          echo "The airflow container already restarted ${restart_count} times before the" \
+               "kill, so a restart afterwards would not prove that the scheduler is" \
+               "supervised." >&2
+          exit 1
+      fi
+
+  # Killing an already dead scheduler would not prove anything, so wait for it
+  # to be up. The script is piped in rather than copied with `kubectl cp`, which
+  # would need `tar` in the Airflow image.
+  - script: |
+      kubectl -n $NAMESPACE exec -i airflow-scheduler-default-0 -c airflow -- python - \
+          --airflow-version "{{ airflow_version }}" \
+          --component scheduler \
+          --timeout 60 \
+          < ../../../../templates/kuttl/commons/health.py
+
+  - script: |
+      kubectl -n $NAMESPACE exec -i airflow-scheduler-default-0 -c airflow -- bash <<'EOF'
+      set -euo pipefail
+
+      scheduler_killed=false
+
+      for pid in $(cat /proc/1/task/1/children); do
+          command_line=$(tr '\0' ' ' < "/proc/$pid/cmdline")
+
+          case "$command_line" in
+          *"airflow scheduler"*)
+              echo "Killing the scheduler (PID $pid)"
+              kill -9 "$pid"
+              scheduler_killed=true
+              ;;
+          esac
+      done
+
+      if [ "$scheduler_killed" = false ]; then
+          echo "No scheduler process found" >&2
+          exit 1
+      fi
+      EOF

tests/templates/kuttl/commons/health.py

@@ -5,6 +5,40 @@
 import time
 import argparse
 
+# Every component of the health endpoint reports its heartbeat under a key named
+# after itself, e.g. `latest_scheduler_heartbeat` for the `scheduler`. The
+# exception is `metadatabase`, which reports a status only.
+HEARTBEAT_KEY = "latest_{}_heartbeat".format
+
+
+def health_url(airflow_version):
+    """Return the health endpoint of the given Airflow version."""
+    if airflow_version and airflow_version.startswith("3"):
+        return "http://airflow-webserver:8080/api/v2/monitor/health"
+    else:
+        return "http://airflow-webserver:8080/api/v1/health"
+
+
+def component_states(health, components):
+    """Return the (status, heartbeat) pair of every requested component."""
+    return {
+        name: (health[name]["status"], health[name].get(HEARTBEAT_KEY(name)))
+        for name in components
+    }
+
+
+def all_healthy(states, baseline):
+    """Check that every component is healthy.
+
+    If a baseline was taken, every component must also have sent a heartbeat
+    since, i.e. one that differs from the baseline.
+    """
+    return all(
+        status == "healthy" and (baseline is None or heartbeat != baseline[name])
+        for name, (status, heartbeat) in states.items()
+    )
+
+
 if __name__ == "__main__":
     log_level = "DEBUG"
     logging.basicConfig(
@@ -15,11 +49,33 @@
 
     parser = argparse.ArgumentParser(description="Health check script")
     parser.add_argument("--airflow-version", type=str, help="Airflow version")
+    parser.add_argument(
+        "--component",
+        action="append",
+        default=[],
+        metavar="NAME",
+        help="Require this component to report healthy, e.g. `scheduler`. "
+        "Can be repeated. Without it, a reachable endpoint is enough.",
+    )
+    parser.add_argument(
+        "--heartbeat-changed",
+        action="store_true",
+        help="Additionally require every --component to send a heartbeat that "
+        "differs from the first one observed. Use this to assert that a "
+        "component recovered, rather than that it was healthy at some point.",
+    )
+    parser.add_argument(
+        "--timeout",
+        type=int,
+        default=0,
+        help="Give up after this many seconds. The default of 0 retries until "
+        "the caller times out.",
+    )
     opts = parser.parse_args()
 
-    url = "http://airflow-webserver:8080/api/v1/health"
-    if opts.airflow_version and opts.airflow_version.startswith("3"):
-        url = "http://airflow-webserver:8080/api/v2/monitor/health"
+    url = health_url(opts.airflow_version)
+    deadline = time.monotonic() + opts.timeout if opts.timeout else None
+    baseline = None
 
     count = 0
 
@@ -29,7 +85,17 @@
             res = requests.get(url, timeout=5)
             code = res.status_code
             if code == 200:
-                break
+                states = component_states(res.json(), opts.component)
+
+                if opts.heartbeat_changed and baseline is None:
+                    baseline = {name: beat for name, (_, beat) in states.items()}
+                    print(f"Heartbeats to be superseded {baseline} ....")
+                elif all_healthy(states, baseline):
+                    break
+                else:
+                    print(
+                        f"Components are not (yet) healthy {states}, retrying attempt no [{count}] ...."
+                    )
             else:
                 print(
                     f"Got non 200 status code [{code}], retrying attempt no [{count}] ...."
@@ -45,6 +111,12 @@
                 f"General error occurred {str(e)}, retrying attempt no [{count}] ...."
             )
 
+        if deadline and time.monotonic() > deadline:
+            sys.exit(
+                f"The health check did not succeed within {opts.timeout} seconds, "
+                "see the attempts above."
+            )
+
         # Wait a little bit before retrying
         time.sleep(1)
     sys.exit(0)

tests/templates/kuttl/ldap/80-assert.yaml.j2

@@ -6,7 +6,12 @@ metadata:
 timeout: 480
 commands:
 {% if test_scenario['values']['airflow-latest'].find(",") > 0 %}
-  - script: kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ test_scenario['values']['airflow-latest'].split(',')[0] }}"
+{% set airflow_version = test_scenario['values']['airflow-latest'].split(',')[0] %}
 {% else %}
-  - script: kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ test_scenario['values']['airflow-latest'] }}"
+{% set airflow_version = test_scenario['values']['airflow-latest'] %}
 {% endif %}
+  - script: |
+      kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py \
+          --airflow-version "{{ airflow_version }}" \
+          --component scheduler \
+          --component dag_processor