Skip to content

[Airflow 2.x] check if stand alone dag processor is active in get_health endpoint #48612

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pierrejeambrun merged 5 commits into from
May 19, 2025
Merged

[Airflow 2.x] check if stand alone dag processor is active in get_health endpoint #48612

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
121b4e1
check if stand alone dag processor is used
rawwar Apr 1, 2025
756bb7a
Update airflow/api/common/airflow_health.py
rawwar Apr 2, 2025
0d56ddb
Merge branch 'v2-11-test' of github.com:apache/airflow into kalyan/da…
rawwar May 18, 2025
39d8a3e
fix tests
rawwar May 18, 2025
9735ad3
update tests
rawwar May 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion airflow/api/common/airflow_health.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

from typing import Any

from airflow.configuration import conf
from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
from airflow.jobs.triggerer_job_runner import TriggererJobRunner
Expand Down Expand Up @@ -61,7 +62,7 @@ def get_airflow_health() -> dict[str, Any]:
try:
latest_dag_processor_job = DagProcessorJobRunner.most_recent_job()

if latest_dag_processor_job:
if conf.getboolean("scheduler", "standalone_dag_processor") and latest_dag_processor_job:
latest_dag_processor_heartbeat = latest_dag_processor_job.latest_heartbeat.isoformat()
if latest_dag_processor_job.is_alive():
dag_processor_status = HEALTHY
Expand Down
11 changes: 9 additions & 2 deletions tests/api/common/test_airflow_health.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,11 +47,12 @@ def test_get_airflow_health_only_metadatabase_healthy(
assert health_status == expected_status


@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=True)
@patch("airflow.api.common.airflow_health.SchedulerJobRunner.most_recent_job", return_value=Exception)
@patch("airflow.api.common.airflow_health.TriggererJobRunner.most_recent_job", return_value=Exception)
@patch("airflow.api.common.airflow_health.DagProcessorJobRunner.most_recent_job", return_value=Exception)
def test_get_airflow_health_metadatabase_unhealthy(
latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
get_boolean_mock, latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
):
health_status = get_airflow_health()

Expand All @@ -65,6 +66,11 @@ def test_get_airflow_health_metadatabase_unhealthy(
assert health_status == expected_status


def test_get_airflow_health_no_dag_processor():
health_status = get_airflow_health()
assert health_status["dag_processor"] == {"status": None, "latest_dag_processor_heartbeat": None}


LATEST_SCHEDULER_JOB_MOCK = MagicMock()
LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime.now()
LATEST_SCHEDULER_JOB_MOCK.is_alive = MagicMock(return_value=True)
Expand Down Expand Up @@ -103,6 +109,7 @@ def test_get_airflow_health_scheduler_healthy_no_triggerer(
LATEST_DAG_PROCESSOR_JOB_MOCK.is_alive = MagicMock(return_value=True)


@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=True)
@patch("airflow.api.common.airflow_health.SchedulerJobRunner.most_recent_job", return_value=None)
@patch(
"airflow.api.common.airflow_health.TriggererJobRunner.most_recent_job",
Expand All @@ -113,7 +120,7 @@ def test_get_airflow_health_scheduler_healthy_no_triggerer(
return_value=LATEST_DAG_PROCESSOR_JOB_MOCK,
)
def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record(
latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
get_boolean_mock, latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
):
health_status = get_airflow_health()

Expand Down
Loading