diff --git a/CHANGELOG.md b/CHANGELOG.md index 600b5620..43606476 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/modules/airflow/pages/getting_started/first_steps.adoc b/docs/modules/airflow/pages/getting_started/first_steps.adoc index f1eb3290..1cb87c27 100644 --- a/docs/modules/airflow/pages/getting_started/first_steps.adoc +++ b/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 diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 816a4f1b..6b759330 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/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 { + 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/ca-cert/50-assert.yaml.j2 b/tests/templates/kuttl/ca-cert/50-assert.yaml.j2 index b85052aa..ed51a3f1 100644 --- a/tests/templates/kuttl/ca-cert/50-assert.yaml.j2 +++ b/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 diff --git a/tests/templates/kuttl/cluster-operation/11-assert.yaml.j2 b/tests/templates/kuttl/cluster-operation/11-assert.yaml.j2 new file mode 100644 index 00000000..89698d6a --- /dev/null +++ b/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 diff --git a/tests/templates/kuttl/cluster-operation/11-kill-scheduler.yaml.j2 b/tests/templates/kuttl/cluster-operation/11-kill-scheduler.yaml.j2 new file mode 100644 index 00000000..52d77b14 --- /dev/null +++ b/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 diff --git a/tests/templates/kuttl/commons/health.py b/tests/templates/kuttl/commons/health.py index de745052..4f7d56d7 100755 --- a/tests/templates/kuttl/commons/health.py +++ b/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) diff --git a/tests/templates/kuttl/ldap/80-assert.yaml.j2 b/tests/templates/kuttl/ldap/80-assert.yaml.j2 index b85052aa..ed51a3f1 100644 --- a/tests/templates/kuttl/ldap/80-assert.yaml.j2 +++ b/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 diff --git a/tests/templates/kuttl/logging/51-assert.yaml.j2 b/tests/templates/kuttl/logging/51-assert.yaml.j2 index fa26b11c..6c1092bb 100644 --- a/tests/templates/kuttl/logging/51-assert.yaml.j2 +++ b/tests/templates/kuttl/logging/51-assert.yaml.j2 @@ -4,12 +4,21 @@ kind: TestAssert metadata: name: test-airflow-webserver-health-check timeout: 480 -commands: {% if test_scenario['values']['airflow'].find(",") > 0 %} {% set airflow_version = test_scenario['values']['airflow'].split(',')[0] %} {% else %} {% set airflow_version = test_scenario['values']['airflow'] %} {% endif %} +{% set components = "--component scheduler" %} +{% if airflow_version.startswith("3") %} +{# With 3.x the scheduler container also runs the dag-processor. #} +{% set components = components ~ " --component dag_processor" %} +{% endif %} +commands: + # Called twice so that the webserver logs more than one request, which the + # log aggregation is asserted on further down. - script: | - kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ airflow_version }}" - kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ airflow_version }}" + for _ in 1 2; do + kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py \ + --airflow-version "{{ airflow_version }}" {{ components }} + done diff --git a/tests/templates/kuttl/mount-dags-configmap/50-assert.yaml.j2 b/tests/templates/kuttl/mount-dags-configmap/50-assert.yaml.j2 index b85052aa..ed51a3f1 100644 --- a/tests/templates/kuttl/mount-dags-configmap/50-assert.yaml.j2 +++ b/tests/templates/kuttl/mount-dags-configmap/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 diff --git a/tests/templates/kuttl/mount-dags-gitsync/50-assert.yaml.j2 b/tests/templates/kuttl/mount-dags-gitsync/50-assert.yaml.j2 index b85052aa..817bbd9f 100644 --- a/tests/templates/kuttl/mount-dags-gitsync/50-assert.yaml.j2 +++ b/tests/templates/kuttl/mount-dags-gitsync/50-assert.yaml.j2 @@ -6,7 +6,13 @@ 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 \ + --component triggerer diff --git a/tests/templates/kuttl/remote-logging/60-assert.yaml.j2 b/tests/templates/kuttl/remote-logging/60-assert.yaml.j2 index b85052aa..ed51a3f1 100644 --- a/tests/templates/kuttl/remote-logging/60-assert.yaml.j2 +++ b/tests/templates/kuttl/remote-logging/60-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 diff --git a/tests/templates/kuttl/smoke/60-assert.yaml.j2 b/tests/templates/kuttl/smoke/60-assert.yaml.j2 index 8b1f71bf..2a39317d 100644 --- a/tests/templates/kuttl/smoke/60-assert.yaml.j2 +++ b/tests/templates/kuttl/smoke/60-assert.yaml.j2 @@ -4,9 +4,17 @@ kind: TestAssert metadata: name: test-airflow-webserver-health-check timeout: 480 -commands: {% if test_scenario['values']['airflow'].find(",") > 0 %} - - script: kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ test_scenario['values']['airflow'].split(',')[0] }}" +{% set airflow_version = test_scenario['values']['airflow'].split(',')[0] %} {% else %} - - script: kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py --airflow-version "{{ test_scenario['values']['airflow'] }}" +{% set airflow_version = test_scenario['values']['airflow'] %} +{% endif %} +{% set components = "--component scheduler --component triggerer" %} +{% if airflow_version.startswith("3") %} +{# Only 3.x runs the dag-processor as a process of its own. #} +{% set components = components ~ " --component dag_processor" %} {% endif %} +commands: + - script: | + kubectl exec -n $NAMESPACE test-airflow-python-0 -- python /tmp/health.py \ + --airflow-version "{{ airflow_version }}" {{ components }} diff --git a/tests/templates/kuttl/triggerer/50-assert.yaml.j2 b/tests/templates/kuttl/triggerer/50-assert.yaml.j2 index b85052aa..817bbd9f 100644 --- a/tests/templates/kuttl/triggerer/50-assert.yaml.j2 +++ b/tests/templates/kuttl/triggerer/50-assert.yaml.j2 @@ -6,7 +6,13 @@ 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 \ + --component triggerer diff --git a/tests/templates/kuttl/versioning/50-assert.yaml.j2 b/tests/templates/kuttl/versioning/50-assert.yaml.j2 index b85052aa..ed51a3f1 100644 --- a/tests/templates/kuttl/versioning/50-assert.yaml.j2 +++ b/tests/templates/kuttl/versioning/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