kaxil commented on code in PR #76751:
URL: https://github.com/apache/airflow/pull/76751#discussion_r4518144874
##########
airflow-core/src/airflow/api/common/airflow_health.py:
##########
@@ -53,16 +55,23 @@ class DetailedHealthStatus(str, Enum):
@provide_session
def get_jobs_health(job_runner_class, *, session: Session = NEW_SESSION) ->
list[Job]:
- """Return unfinished jobs for the runner class, ordered by latest
heartbeat."""
+ """
+ Return live unfinished jobs and the latest unfinished job, ordered by
heartbeat.
+
+ Keep the latest job even when dead so health responses retain its
heartbeat.
+ """
+ job_type = job_runner_class.job_type
+ unfinished = (Job.job_type == job_type, Job.end_date.is_(None))
+ cutoff = timezone.utcnow() -
timedelta(seconds=health_check_threshold(job_type, 0))
+ latest_job =
select(Job.id).where(*unfinished).order_by(Job.latest_heartbeat.desc()).limit(1).subquery()
+ live_job_ids = select(Job.id).where(
+ *unfinished,
+ Job.state == JobState.RUNNING,
+ Job.latest_heartbeat > cutoff,
+ )
+ selected_job_ids = live_job_ids.union_all(select(latest_job.c.id))
return list(
- session.scalars(
- select(Job)
- .where(
- Job.job_type == job_runner_class.job_type,
- Job.end_date.is_(None),
- )
- .order_by(Job.latest_heartbeat.desc())
- )
+
session.scalars(select(Job).where(Job.id.in_(selected_job_ids)).order_by(Job.latest_heartbeat.desc()))
Review Comment:
MySQL 8 can't semijoin an `IN` over a `UNION ALL`, so it runs this as a
dependent subquery per row. On a 20k-row `job` table `EXPLAIN ANALYZE` showed a
full table scan with `loops=20002`, finished jobs and other job types included,
so on MySQL the health check still scales with job history. Could this be two
separate queries (live jobs, plus the latest one with `LIMIT 1`) merged in
Python? Those use the indexes on MySQL too.
##########
airflow-core/src/airflow/api/common/airflow_health.py:
##########
@@ -53,16 +55,23 @@ class DetailedHealthStatus(str, Enum):
@provide_session
def get_jobs_health(job_runner_class, *, session: Session = NEW_SESSION) ->
list[Job]:
- """Return unfinished jobs for the runner class, ordered by latest
heartbeat."""
+ """
+ Return live unfinished jobs and the latest unfinished job, ordered by
heartbeat.
+
+ Keep the latest job even when dead so health responses retain its
heartbeat.
+ """
+ job_type = job_runner_class.job_type
+ unfinished = (Job.job_type == job_type, Job.end_date.is_(None))
+ cutoff = timezone.utcnow() -
timedelta(seconds=health_check_threshold(job_type, 0))
+ latest_job =
select(Job.id).where(*unfinished).order_by(Job.latest_heartbeat.desc()).limit(1).subquery()
+ live_job_ids = select(Job.id).where(
Review Comment:
This is a third copy of the alive rule, next to `Job._is_alive` and
`_IsAliveFilter.to_orm` in `api_fastapi/common/parameters/misc.py` (which
already notes it's kept in sync by hand). If the rule changes and this copy is
missed, health reports a live component as down. Could both SQL callers share a
small helper in `jobs/job.py`? Fine as a follow-up, since the copies agree
today.
--
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]
