diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d4ab5e4..14f1ea4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,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 @@ -42,6 +43,7 @@ [#835]: https://github.com/stackabletech/airflow-operator/pull/835 [#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 diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 6ce8ca0c..397f665a 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -618,16 +618,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(), @@ -1035,8 +1039,13 @@ mod tests { commons::product_image_selection::ResolvedProductImage, versioned::test_utils::RoundtripTestData, }; + use strum::IntoEnumIterator; use crate::{ + controller::{ + ValidatedCluster, + build::test_support::{validated_cluster, validated_cluster_with}, + }, crd::{AirflowRole, trusted_proxies::TrustedProxy}, v1alpha1, v1alpha2, }; @@ -1234,6 +1243,79 @@ mod tests { ); } + /// The commands that background a process, i.e. the candidates for `$!`. + fn backgrounded_commands(role: &AirflowRole, cluster: &ValidatedCluster) -> Vec { + 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 { let git_sync_section = r#" diff --git a/tests/templates/kuttl/cluster-operation/11-assert.yaml b/tests/templates/kuttl/cluster-operation/11-assert.yaml new file mode 100644 index 00000000..f6bbf649 --- /dev/null +++ b/tests/templates/kuttl/cluster-operation/11-assert.yaml @@ -0,0 +1,83 @@ +--- +# 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 +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-scheduler-default-0 -c airflow -- python - <<'EOF' + import requests + + health = requests.get("http://airflow-webserver:8080/api/v2/monitor/health", timeout=10).json() + + print(f" scheduler: {health['scheduler']}") + print(f" dag_processor: {health['dag_processor']}") + EOF + + exit 1 + + - script: | + kubectl -n $NAMESPACE exec -i airflow-webserver-default-0 -c airflow -- python - <<'EOF' + import sys + import time + + import requests + + HEALTH_URL = "http://airflow-webserver:8080/api/v2/monitor/health" + HEARTBEATS = { + "scheduler": "latest_scheduler_heartbeat", + "dag_processor": "latest_dag_processor_heartbeat", + } + + baseline = None + + for attempt in range(1, 21): + try: + response = requests.get(HEALTH_URL, timeout=10) + except requests.RequestException as error: + print(f"attempt {attempt}: {error}") + else: + if response.ok: + health = response.json() + state = { + name: (health[name]["status"], health[name][key]) + for name, key in HEARTBEATS.items() + } + + if baseline is None: + baseline = {name: heartbeat for name, (_, heartbeat) in state.items()} + print(f"heartbeats before the restart took effect: {baseline}") + elif all( + status == "healthy" and heartbeat != baseline[name] + for name, (status, heartbeat) in state.items() + ): + print(f"recovered after the restart on attempt {attempt}: {state}") + break + else: + print(f"attempt {attempt}: {state}") + else: + print(f"attempt {attempt}: HTTP {response.status_code} {response.text[:200]}") + + time.sleep(3) + else: + sys.exit("the restarted container did not bring the scheduler and dag-processor back") + EOF diff --git a/tests/templates/kuttl/cluster-operation/11-kill-scheduler.yaml b/tests/templates/kuttl/cluster-operation/11-kill-scheduler.yaml new file mode 100644 index 00000000..86d75c36 --- /dev/null +++ b/tests/templates/kuttl/cluster-operation/11-kill-scheduler.yaml @@ -0,0 +1,72 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +metadata: + name: kill-scheduler +timeout: 120 +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 + + - script: | + kubectl -n $NAMESPACE exec -i airflow-scheduler-default-0 -c airflow -- python - <<'EOF' + import sys + import time + + import requests + + HEALTH_URL = "http://airflow-webserver:8080/api/v2/monitor/health" + + for attempt in range(1, 11): + try: + response = requests.get(HEALTH_URL, timeout=10) + except requests.RequestException as error: + print(f"attempt {attempt}: {error}") + else: + if response.ok: + scheduler = response.json()["scheduler"] + + if scheduler["status"] == "healthy": + print(f"scheduler before the kill: {scheduler}") + break + + print(f"attempt {attempt}: scheduler is {scheduler}") + else: + print(f"attempt {attempt}: HTTP {response.status_code} {response.text[:200]}") + + time.sleep(3) + else: + sys.exit("expected a healthy scheduler, see the attempts above") + EOF + + - 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