ferruzzi commented on code in PR #76410:
URL: https://github.com/apache/airflow/pull/76410#discussion_r4519975722
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -425,6 +471,136 @@ def __init__(
def heartbeat_callback(self, *, session: Session = NEW_SESSION) -> None:
stats.incr("scheduler_heartbeat", 1, 1)
+ def _check_executor_health(self) -> None:
+ """
+ Kick off a health check for each reporting executor, in the background.
+
+ Called on its own timer (``[scheduler]
executor_health_check_interval``), independent of the
+ scheduling loop and of each executor's own
:meth:`~airflow.executors.base_executor.BaseExecutor.heartbeat`.
+ A new check for a given executor is skipped while its previous check
is still running, so a
+ slow or hung dependency can only ever delay that executor's own health
being updated, never
+ pile up work or block this (or any other) scheduler tick.
+
+ ``get_health()`` is passed a timeout, but an implementation that
ignores it can still leave its
+ future pending forever. That future's own worker slot is never
reclaimed -- the pool is sized
+ one worker per executor, so a stuck probe can only ever cost that
executor's own slot -- but
+ without this check it would silently stop being checked again, with no
sign anything was
+ wrong. Once a pending future is a few intervals old, it's reported as
stalled instead.
+ """
+ for executor in self.executors:
+ key = _executor_health_key(executor)
+ if key not in self._executor_health_keys:
+ # Doesn't override get_health(); there's nothing to check or
record.
+ continue
+ future, started = self._executor_health_check_futures.get(key,
(None, 0.0))
+ if future is not None and not future.done():
+ stall_after = (
+ self._executor_health_check_interval *
self._executor_health_check_stall_after_intervals
+ )
+ if time.monotonic() - started > stall_after and key not in
self._stalled_health_checks:
+ self._stalled_health_checks.add(key)
+ self.log.warning("Health check for executor %s has stopped
responding", key)
+ stats.incr("scheduler.executor_health_check.stalled",
tags={"executor": key})
+ else:
+ self.log.debug(
+ "Skipping health check for executor %s, the previous
check hasn't finished", key
+ )
+ continue
+ self._stalled_health_checks.discard(key)
+ future =
self._executor_health_check_pool.submit(self._run_executor_health_check,
executor, key)
+ self._executor_health_check_futures[key] = (future,
time.monotonic())
+
+ def _run_executor_health_check(self, executor: BaseExecutor, key: str) ->
None:
+ """Run one executor's get_health() off the scheduler thread and record
an explicit result."""
+ try:
+ healthy =
executor.get_health(timeout=self._executor_health_check_timeout)
+ except Exception:
+ self.log.exception("Health check for executor %s raised an
exception", key)
+ self._record_executor_health_result(key, executor.team_name,
HealthCheckResult.UNKNOWN)
+ return
+ if healthy is None:
+ result = HealthCheckResult.UNKNOWN
+ elif healthy:
+ result = HealthCheckResult.HEALTHY
+ else:
+ self.log.warning("Executor %s reported itself unhealthy", key)
+ result = HealthCheckResult.UNHEALTHY
+ self._record_executor_health_result(key, executor.team_name, result)
Review Comment:
I think this should be inside another `try`, no? What would happen if there
is a database error, for example? This runs on the thread pool, so trace back
where an exception there gets handled.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -425,6 +471,136 @@ def __init__(
def heartbeat_callback(self, *, session: Session = NEW_SESSION) -> None:
stats.incr("scheduler_heartbeat", 1, 1)
+ def _check_executor_health(self) -> None:
Review Comment:
It looks like we can use better test coverage for this one,
`_run_executor_health_check`, `_touch_executor_heartbeats`, and
`_get_or_create_executor_job`.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -425,6 +471,136 @@ def __init__(
def heartbeat_callback(self, *, session: Session = NEW_SESSION) -> None:
stats.incr("scheduler_heartbeat", 1, 1)
+ def _check_executor_health(self) -> None:
+ """
+ Kick off a health check for each reporting executor, in the background.
+
+ Called on its own timer (``[scheduler]
executor_health_check_interval``), independent of the
+ scheduling loop and of each executor's own
:meth:`~airflow.executors.base_executor.BaseExecutor.heartbeat`.
+ A new check for a given executor is skipped while its previous check
is still running, so a
+ slow or hung dependency can only ever delay that executor's own health
being updated, never
+ pile up work or block this (or any other) scheduler tick.
+
+ ``get_health()`` is passed a timeout, but an implementation that
ignores it can still leave its
+ future pending forever. That future's own worker slot is never
reclaimed -- the pool is sized
+ one worker per executor, so a stuck probe can only ever cost that
executor's own slot -- but
+ without this check it would silently stop being checked again, with no
sign anything was
+ wrong. Once a pending future is a few intervals old, it's reported as
stalled instead.
+ """
+ for executor in self.executors:
+ key = _executor_health_key(executor)
+ if key not in self._executor_health_keys:
+ # Doesn't override get_health(); there's nothing to check or
record.
+ continue
+ future, started = self._executor_health_check_futures.get(key,
(None, 0.0))
+ if future is not None and not future.done():
+ stall_after = (
+ self._executor_health_check_interval *
self._executor_health_check_stall_after_intervals
+ )
+ if time.monotonic() - started > stall_after and key not in
self._stalled_health_checks:
+ self._stalled_health_checks.add(key)
+ self.log.warning("Health check for executor %s has stopped
responding", key)
+ stats.incr("scheduler.executor_health_check.stalled",
tags={"executor": key})
+ else:
+ self.log.debug(
+ "Skipping health check for executor %s, the previous
check hasn't finished", key
+ )
+ continue
+ self._stalled_health_checks.discard(key)
+ future =
self._executor_health_check_pool.submit(self._run_executor_health_check,
executor, key)
+ self._executor_health_check_futures[key] = (future,
time.monotonic())
+
+ def _run_executor_health_check(self, executor: BaseExecutor, key: str) ->
None:
+ """Run one executor's get_health() off the scheduler thread and record
an explicit result."""
+ try:
+ healthy =
executor.get_health(timeout=self._executor_health_check_timeout)
+ except Exception:
+ self.log.exception("Health check for executor %s raised an
exception", key)
+ self._record_executor_health_result(key, executor.team_name,
HealthCheckResult.UNKNOWN)
+ return
+ if healthy is None:
+ result = HealthCheckResult.UNKNOWN
+ elif healthy:
+ result = HealthCheckResult.HEALTHY
+ else:
+ self.log.warning("Executor %s reported itself unhealthy", key)
+ result = HealthCheckResult.UNHEALTHY
+ self._record_executor_health_result(key, executor.team_name, result)
+
+ @provide_session
+ def _touch_executor_heartbeats(self, *, session: Session = NEW_SESSION) ->
None:
+ """
+ Keep every reporting executor's Job row heartbeat moving on the
scheduler's own cadence.
+
+ ``get_health()`` itself only runs every
``executor_health_check_interval`` (default 30s), far
+ slower than a Job row can go quiet for before
``health_check_threshold`` calls it dead.
+ Updating the heartbeat only when a check finishes would make a
perfectly healthy executor
+ flicker to UNHEALTHY between checks, every cycle. Keeping the
heartbeat on the scheduler's own
+ short cadence instead, the same way it already does for its own Job
row, means the row's
+ liveness reflects "the scheduler is still running and still intends to
check this executor",
+ while ``health_check_result`` -- not this heartbeat -- carries
whatever the last actual check
+ found.
+ """
+ if not self._executor_health_keys:
+ return
+ now = timezone.utcnow()
+ for key, team_name in self._executor_health_keys.items():
+ job = self._get_or_create_executor_job(key, team_name,
session=session)
+ job.latest_heartbeat = now
+ session.commit()
+
+ def _get_or_create_executor_job(
+ self, executor_key: str, team_name: str | None, *, session: Session
+ ) -> Job:
+ """
+ Return this scheduler's own Job row for ``executor_key``, creating it
if needed.
+
+ Keyed by ``owner_job_id`` (this scheduler's own Job id), not just
``hostname``: two schedulers
+ running the same executor against shared infrastructure (e.g. the same
Celery broker) each get
+ their own row, so one scheduler's healthy report can never paper over
another's failing one.
+ A restarted scheduler gets a fresh row for the same reason its own Job
row is fresh -- a new
+ id -- instead of quietly taking over a row left behind by its previous
instance.
+ """
+ job = session.scalar(
+ select(Job).where(
+ Job.job_type == EXECUTOR_JOB_TYPE,
+ Job.hostname == executor_key,
+ Job.owner_job_id == self.job.id,
+ Job.end_date.is_(None),
+ )
+ )
+ if job is None:
+ job = Job(job_type=EXECUTOR_JOB_TYPE)
+ job.hostname = executor_key
Review Comment:
The key isn't strictly the hostname, is it? You set it to
`ExecutorName.__repr__()` so the jobs list and health endpoint are going to
show something like `"hostname": ":CeleryExecutor:"` and not the actual
hostname. This should stay as the actual hostname that `Job.__init__` already
sets, and maybe see if `Job.executor_class` would be a better fit for the key?
That's a leftover column from an old feature @potiuk removed a few years ago in
#32219
##########
airflow-core/src/airflow/api/common/airflow_health.py:
##########
@@ -156,6 +189,63 @@ def _dag_processor_detailed_status(jobs: list[Job]) ->
DetailedHealthStatus:
return _coverage_status(expected, covered)
+def _configured_executor_keys() -> set[str]:
+ """
+ Return the key each configured executor that implements a health check
would be recorded under.
+
+ Most executors don't override
:meth:`~airflow.executors.base_executor.BaseExecutor.get_health`,
+ so this deliberately excludes them rather than counting every configured
executor: an executor
+ that was never going to report anything isn't a gap in coverage, and a
deployment where nothing
+ implements a health check should see no ``executor`` block at all in
``/monitor/health``, not an
+ ``unhealthy`` one.
+ """
+ from airflow.executors.base_executor import BaseExecutor
+
+ try:
+ names = ExecutorLoader.get_executor_names()
+ except Exception:
+ # A health probe must not fail on malformed executor config; callers
fall back to liveness only.
+ log.warning("Could not read the executor configuration", exc_info=True)
+ return set()
+
+ keys = set()
+ for name in names:
+ try:
+ executor_cls, _ = ExecutorLoader.import_executor_cls(name)
+ except Exception:
+ log.warning("Could not import configured executor %s",
name.module_path, exc_info=True)
+ continue
+ if executor_cls.get_health is not BaseExecutor.get_health:
+ keys.add(repr(name))
+ return keys
+
+
+def _executor_detailed_status(jobs: list[Job]) -> DetailedHealthStatus:
+ """
+ Status from configured-executor coverage: each configured executor is its
own partition.
+
+ A shared executor backend (e.g. one Celery broker) can be reachable from
one scheduler and not
+ another, so coverage is decided per live, currently-owned row rather than
by any single row for
+ that executor key: one scheduler finding it down is real signal, not noise
an unrelated healthy
+ row should paper over.
+ """
+ expected = _configured_executor_keys()
+ if not expected:
+ return _liveness_status(jobs)
+
+ covered: set[str] = set()
+ unhealthy: set[str] = set()
+ for job in _live_executor_jobs(jobs):
+ if not job.hostname or job.hostname not in expected:
+ continue
+ if job.health_check_result == HealthCheckResult.HEALTHY:
+ covered.add(job.hostname)
+ else:
+ unhealthy.add(job.hostname)
Review Comment:
Every row starts as UNKNOWN, which is right, but here an UNKNOWN gets lumped
in with UNHEALTHY. So every executor reads down until its first check
finishes, and an executor whose check can't tell (returns None) reads down
forever. I'd treat "can't tell yet" the same way `_configured_executor_keys()`
already treats "doesn't report": it isn't a gap in coverage.
- Skip UNKNOWN rows in this loop, so they don't land in either list
(consider it neither `covered` nor `unhealthy`).
- Drop an executor from `expected` when it has live rows but every one of
them is UNKNOWN. An executor with no live row at all should stay in `expected`
and still count as uncovered, the way it does now.
- If that leaves `expected` empty, return `_liveness_status(jobs)`, the same
fallback the function already uses at the top.
Something like this (untested code, and swap `job.hostname` for wherever the
key ends up after my other comment):
```python
covered: set[str] = set()
unhealthy: set[str] = set()
seen: set[str] = set()
for job in _live_executor_jobs(jobs):
if not job.hostname or job.hostname not in expected:
continue
seen.add(job.hostname)
if job.health_check_result == HealthCheckResult.HEALTHY:
covered.add(job.hostname)
elif job.health_check_result == HealthCheckResult.UNHEALTHY:
unhealthy.add(job.hostname)
# Every live row for these is still UNKNOWN: no answer yet, so not a gap
in coverage.
expected -= seen - covered - unhealthy
if not expected:
return _liveness_status(jobs)
covered -= unhealthy
return _coverage_status(expected, covered)
```
`_executor_legacy_status` has the same problem, since it only reads healthy
once some row is HEALTHY. Setting status to UNHEALTHY only when
`detailed_status` is DOWN keeps the two from disagreeing. It's worth adding a
few UNKNOWN cases tot he tests as well: a first check still pending, a restart
while another scheduler reports, and UNKNOWN alongside UNHEALTHY.
--
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]
