ferruzzi commented on code in PR #76410:
URL: https://github.com/apache/airflow/pull/76410#discussion_r4472988337
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -425,6 +435,74 @@ 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 configured 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.
+ """
+ for executor in self.executors:
+ key = executor.name.alias if executor.name and executor.name.alias
else type(executor).__name__
+ future = self._executor_health_check_futures.get(key)
+ if future is not None and not future.done():
+ self.log.debug(
+ "Skipping health check for executor %s, the previous check
hasn't finished", key
+ )
+ continue
+ self._executor_health_check_futures[key] =
self._executor_health_check_pool.submit(
+ self._run_executor_health_check, executor, key
+ )
+
+ def _run_executor_health_check(self, executor: BaseExecutor, key: str) ->
None:
+ """Run one executor's get_health() off the scheduler thread and record
the result, if any."""
+ try:
+ healthy = executor.get_health()
Review Comment:
The in-flight guard protects the loop, but `get_health()` is called with no
bound. A probe that never returns leaves its `Future` pending forever, this
guard skips every later cycle, and the row goes stale for good. The executor
would read unhealthy until the scheduler restarts, which looks the same as a
real backend outage.
I'd make it part of the contract now:
```
def get_health(self, timeout: float) -> bool | None:
```
and require implementations to pass `timeout` through to their own I/O. The
client libraries already take one: Celery has `inspect(timeout=...)`, botocore
has
`Config(connect_timeout=..., read_timeout=...)`, etc. This matters even for
healthy backends. botocore defaults to 60s connect and 60s read, so an
ECS-style probe left on defaults can hold its slot for up to two minutes with
nothing wrong.
The scheduler would have to pass something below
`executor_health_check_interval`, let's arbitrarily say half of it to start
with. A probe that honours the timeout then always finishes before the next
tick, so this guard only trips for one that doesn't. Nothing overrides
`get_health()` yet, so this the best time to add the
parameter. After the Celery follow-up it becomes a breaking change.
For an implementation that ignores its timeout, keep a backstop so the hang
gets reported instead of silently skipped. Store the submit time with the
future, and once a pending future is a few intervals old, warn once and emit a
metric. Something like:
```
future, started = self._executor_health_check_futures.get(key, (None,
0.0))
if future is not None and not future.done():
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})
continue
```
Don't resubmit after that. The worker can't be reclaimed, so each
resubmission would leak another one. The current pool sizing already contains
the damage: with one worker per executor and one check in flight per executor,
a hung probe can only use up its own slot.
##########
airflow-core/newsfragments/00000.feature.rst:
##########
@@ -0,0 +1,9 @@
+Added scaffolding for per-executor health reporting. ``BaseExecutor`` gains a
``get_health()`` hook
Review Comment:
New policy is that we only add newsfragments for `significant` changes. You
could definitely argue that this qualifies as significant, but if you ant to
keep it then you'll need to rename the file to `76410.significant.rst`
##########
airflow-core/src/airflow/api/common/airflow_health.py:
##########
@@ -156,6 +164,47 @@ 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(name.alias or name.module_path)
+ return keys
+
+
+def _executor_detailed_status(jobs: list[Job]) -> DetailedHealthStatus:
+ """Status from configured-executor coverage: each configured executor is
its own partition."""
+ expected = _configured_executor_keys()
+ if not expected:
+ return _liveness_status(jobs)
+
+ covered = {job.hostname for job in jobs if job.is_alive() and job.hostname}
+ return _coverage_status(expected, covered)
Review Comment:
I don't think the coverage comparison will match for an executor configured
by module path.
`_configured_executor_keys()` builds `expected` from `name.alias or
name.module_path`, but `_record_executor_heartbeat` stores `executor.name.alias
if ... else type(executor).__name__`, and this function compares the two. For
example: `ExecutorLoader._get_executor_names` creates
`ExecutorName(alias=None, module_path=name)` for a bare module path, so for
`[core] executor = my.package.MyExecutor` you get
`expected={"my.package.MyExecutor"}` against `covered={"MyExecutor"}` and
`_coverage_status` ends up returning `DOWN`.
Core executors configured by name do get an alias, so the aliased path
matches and the current tests pass on it, but I suspect if you add a
parametrized case covering a module-path-configured executor to test this,
you'll see a failure.
Using one identity on both sides would fix it. `ExecutorName.__repr__()` is
already the stable key (it's what `__hash__` uses), so writing that and
comparing against it avoids the two expressions drifting apart again.
--
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]
