ferruzzi commented on code in PR #70416:
URL: https://github.com/apache/airflow/pull/70416#discussion_r3754119316
##########
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:
It looks like there may be a bug with this which we will need to fix
separately, but if you look at `Trigger.assign_unassigned` in
`models/trigger.py:402-406`, they use `end_date IS NULL` and the time since the
last heartbeat to determine if it's "alive" and they specifically don't use
`state`. We should do similar here, but it may not be so easy... Without some
kind of upper time range, once a triggerer goes "DEGRADED" there is no way for
that failed job to fall off the end of the list so the triggerer will never
recover unless that job row is removed from the db, which is also wrong....
You would then also have to modify the same concept in the dag-processor
around L152 and triggerer around L135 for the same reason, and some of your new
tests may need to be adjusted accordingly.
The issue is that if the job quits unexpectedly (process killed, for
example), we don't have a job reaper and that will stay RUNNING for the rest of
time. We should likely add that reaper, but that's definitely out of scope for
this project and the proposed solution is already in use elsewhere so it should
be fine here too I think. I need to think the reaper idea through a bit, but
I'll cut an Issue for it when I have a better idea of the actual issue there.
##########
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:
Every test in here patches `get_jobs_health`, so the filter inside it never
runs against real job rows in the same test as the legacy status derivation. I
think that's why the suite passes despite the change I flagged over in
`airflow_health.py`.
If you add a DB-backed case where a triggerer has one row which is
non-RUNNING and assert `status == "unhealthy"`, I bet it would fail because the
empty result from `get_jobs_health` falls into the `else: triggerer_status =
None` at L135. On main that same DB state reports UNHEALTHY, since
`most_recent_job()` ignores state and
`is_alive()` then returns False.
That's different from the killed-process case in my other comment. A clean
exit leaves the row non-RUNNING, so it disappears from the report; a `kill -9`
leaves it RUNNING so it sticks around as a zombie. But both of those failure
cases come from the same assumption of using the raw state as the deciding
factor, which is why I think they can both be fixed in one place.
--
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]
