ferruzzi commented on code in PR #70416: URL: https://github.com/apache/airflow/pull/70416#discussion_r3786885614
########## 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: @Tegh25 Good discussion, but I think section 5 of your SoW document already settles this and we don't need a new mechanism. The acceptance table has this row: > | 3 jobs, all stale | unhealthy | 3 unhealthy | `latest_heartbeat > now - retention` can't satisfy that. If all three are stale they're all outside the window, the query returns nothing, and we can't report "3 unhealthy". The same row rules out the current `state == RUNNING` filter from the other side, since a cleanly stopped job leaves RUNNING and vanishes from the list. So the filter should be `job_type` (plus `end_date IS NULL` if we want to drop cleanly-exited jobs) and nothing else. No state, no time window. Then `is_alive()` decides healthy vs unhealthy per instance, which is what the table describes, and the aggregate follows: all instances alive = healthy, some alive = degraded, none alive = down. On your L135/L152 question: empty list must not map to None. Today `most_recent_job()` has no filter at all beyond `job_type`, so `status: null` means "this component has never run" and nothing else. Any component that has ever run reports healthy or unhealthy. If we let an empty filtered list produce `null`, then a triggerer that's been dead for an hour reports `"status": null`, and anyone alerting on `status != "healthy"` silently stops getting paged exactly when it matters most. Empty means DOWN, and DOWN means legacy status = "unhealthy". Same for the scheduler, which is already correct since it has no `else: None` branch. That also removes the need for a retention knob: the leftover rows from killed containers that @JH-A-Kim hit are a reaper's job, not the read path's. That's tracked separately, and until it exists those rows (correctly) show as unhealthy instances rather than disappearing. One doc fix while we're here: Phase 1 says "Query all RUNNING jobs per type (or recent jobs within a window)". That parenthetical is what led us here and it contradicts section 5. Likely worth updating that. With that decided you should be unblocked on the DB-backed test you offered, and the whole section 5 table is a good source for cases. -- 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]
