Tegh25 commented on code in PR #70416:
URL: https://github.com/apache/airflow/pull/70416#discussion_r3754925543
##########
airflow-core/src/airflow/api/common/airflow_health.py:
##########
@@ -16,77 +16,165 @@
# under the License.
from __future__ import annotations
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import select
from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
+from airflow.jobs.job import Job, JobState
from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
from airflow.jobs.triggerer_job_runner import TriggererJobRunner
+from airflow.utils.session import NEW_SESSION, provide_session
+
+if TYPE_CHECKING:
+ from sqlalchemy.orm import Session
HEALTHY = "healthy"
UNHEALTHY = "unhealthy"
+DEGRADED = "degraded"
+DOWN = "down"
+
+
+@provide_session
+def get_jobs_health(job_runner_class, *, session: Session = NEW_SESSION) ->
list[Job]:
+ """Return all running jobs for the runner class, ordered by latest
heartbeat."""
+ return list(
+ session.scalars(
+ select(Job)
+ .where(
+ Job.job_type == job_runner_class.job_type,
+ Job.state == JobState.RUNNING,
Review Comment:
I agree filtering `state == RUNNING` can cause issues for the reasons you
described. We also should not reuse `_is_alive()`'s heartbeat threshold in
`get_jobs_health()`, otherwise the query would only return alive jobs and stale
instances would never appear.
*Proposed filter:*
`job_type`, `end_date IS NULL`, and `latest_heartbeat > now - retention`
where `retention` is a multiple of `health_check_threshold`. Inside that window
`is_alive()` still marks fresh vs stale, and it adds some kind of upper time
range so degraded can recover without a reaper.
What about the empty-list path at L135/L152? Clean exit of a job sets
`end_date`, so the filtered list becomes empty and we currently report None
instead of main's `unhealthy`.
##########
airflow-core/tests/unit/api/common/test_airflow_health.py:
##########
@@ -16,127 +16,246 @@
# under the License.
from __future__ import annotations
-from datetime import datetime
+from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
import pytest
+from airflow._shared.timezones import timezone
from airflow.api.common.airflow_health import (
+ DEGRADED,
+ DOWN,
HEALTHY,
UNHEALTHY,
get_airflow_health,
+ get_jobs_health,
)
-from airflow.jobs.job import Job
+from airflow.jobs.job import Job, JobState
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.jobs.triggerer_job_runner import TriggererJobRunner
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import clear_db_jobs
pytestmark = pytest.mark.db_test
-@patch("airflow.api.common.airflow_health.SchedulerJobRunner.most_recent_job",
return_value=None)
-@patch("airflow.api.common.airflow_health.TriggererJobRunner.most_recent_job",
return_value=None)
-@patch("airflow.api.common.airflow_health.DagProcessorJobRunner.most_recent_job",
return_value=None)
-def test_get_airflow_health_only_metadatabase_healthy(
- latest_scheduler_job_mock,
- latest_triggerer_job_mock,
- latest_dag_processor_job_mock,
-):
+@patch("airflow.api.common.airflow_health.get_jobs_health")
+def test_get_airflow_health_only_metadatabase_healthy(mock_get_jobs_health):
+ mock_get_jobs_health.side_effect = [[], [], []]
Review Comment:
Yep, I can definitely work on adding a DB-backed test that does not mock
`get_jobs_health()` once we decide on an approach for job filtering.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
