Stop reporting components as degraded after an instance restart

airflow-core/docs/administration-and-deployment/logging-monitoring/check-health.rst

@@ -51,7 +51,6 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc
       "detailed_status": "healthy",
       "instances": [
         {
-          "status": "healthy",
           "hostname": "scheduler-1.example.com",
           "latest_scheduler_heartbeat": "2018-12-26T17:15:11+00:00"
         }
@@ -63,16 +62,9 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc
       "detailed_status": "degraded",
       "instances": [
         {
-          "status": "healthy",
           "hostname": "triggerer-1.example.com",
           "latest_triggerer_heartbeat": "2018-12-26T17:16:12+00:00",
           "team_name": "team-a"
-        },
-        {
-          "status": "unhealthy",
-          "hostname": "triggerer-2.example.com",
-          "latest_triggerer_heartbeat": "2018-12-26T17:10:00+00:00",
-          "team_name": null
         }
       ]
     },
@@ -82,7 +74,6 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc
       "detailed_status": "healthy",
       "instances": [
         {
-          "status": "healthy",
           "hostname": "dag-processor-1.example.com",
           "latest_dag_processor_heartbeat": "2018-12-26T17:16:12+00:00",
           "bundle_names": ["dags-team-a"]
@@ -100,11 +91,28 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc
   * ``status`` (legacy aggregate): ``"healthy"`` if **any** running instance is alive, otherwise ``"unhealthy"``
     (including when no running jobs exist for that component).
 
-  * ``detailed_status``: reflects the full set of running instances:
-
-    * ``"healthy"`` — every running instance is alive
-    * ``"degraded"`` — some instances are alive and some are not
-    * ``"down"`` — no running instance is alive (including when no jobs exist)
+  * ``detailed_status``: whether every part of that component's work is being covered by a live instance.
+    What counts as "every part" differs per component, because only some of them divide their work up:
+
+    * **Dag processor** — the parts are the Dag bundles in ``[dag_processor] dag_bundle_config_list``.
+      A processor started without ``--bundle-name`` covers every configured bundle;
+      one started with it covers only the bundles it was given.
+      ``"healthy"`` when every configured bundle has a live processor, ``"degraded"`` when only some do,
+      ``"down"`` when none do.
+    * **Triggerer** — with ``[core] multi_team`` enabled, the parts are the teams those bundles are scoped to
+      (plus the unscoped bundles), because a triggerer only picks up triggers for its own team.
+      ``"healthy"`` when every team scope has a live triggerer, ``"degraded"`` when only some do,
+      ``"down"`` when none do. With multi-team disabled, no team filtering applies, so any live triggerer
+      covers everything: ``"healthy"`` if one is alive, ``"down"`` if none is.
+    * **Scheduler** — schedulers are symmetric and share no partitioned work, so there is nothing partial
+      to report: ``"healthy"`` if at least one is alive, ``"down"`` if none is. ``"degraded"`` is never
+      returned for the scheduler. Use ``instances`` to see how many replicas are up, and your orchestrator
+      or the ``scheduler_heartbeat`` metric to alert on reduced scheduling throughput.
+
+    Because the expected set of work comes from configuration rather than from the job table,
+    ``detailed_status`` is unaffected by how instances come and go. Restarting an instance — including after
+    a ``SIGKILL``, an out-of-memory kill, or a node eviction, none of which let Airflow mark the old job row
+    as finished — does not report the component as ``"degraded"``.
 
   * ``latest_*_heartbeat``: the most recent heartbeat among running jobs of that type (ordered by heartbeat descending),
     or ``null`` when there are none.
@@ -113,9 +121,13 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc
     ``[triggerer] triggerer_health_check_threshold``,
     ``[dag_processor] health_check_threshold``).
 
-  * ``instances``: one entry per **running** job of that type (``null`` when there are none). Each entry includes:
+  * ``instances``: one entry per **live** instance of that type — ``null`` when none is live. Airflow cannot
+    mark a job as finished when its process is killed abruptly (``SIGKILL``, an out-of-memory kill, a node
+    eviction), so the job row of such an instance is never closed; listing it would show a host that is gone
+    and, under an orchestrator that assigns a fresh hostname on each restart, will never come back.
+    ``status`` and ``latest_*_heartbeat`` are still derived from every unfinished job, so how long a dead
+    component has been silent stays visible after it drops out of ``instances``. Each entry includes:
 
-    * ``status``: ``"healthy"`` or ``"unhealthy"`` for that instance
     * ``hostname``: host where the component is running
     * the corresponding ``latest_*_heartbeat`` for that instance
     * ``team_name`` (triggerer only): team the triggerer is scoped to, or ``null`` when unscoped

airflow-core/src/airflow/api/common/airflow_health.py

@@ -16,10 +16,14 @@
 # under the License.
 from __future__ import annotations
 
+import logging
+from enum import Enum
 from typing import TYPE_CHECKING, Any
 
 from sqlalchemy import select
 
+from airflow.configuration import conf
+from airflow.dag_processing.bundles.manager import _get_configured_bundle_team_names
 from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner
 from airflow.jobs.job import Job
 from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
@@ -29,10 +33,22 @@
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
 
-HEALTHY = "healthy"
-UNHEALTHY = "unhealthy"
-DEGRADED = "degraded"
-DOWN = "down"
+log = logging.getLogger(__name__)
+
+
+class HealthStatus(str, Enum):
+    """Aggregate health of a component: whether it has at least one live instance."""
+
+    HEALTHY = "healthy"
+    UNHEALTHY = "unhealthy"
+
+
+class DetailedHealthStatus(str, Enum):
+    """How much of a component's work has a live instance covering it."""
+
+    HEALTHY = "healthy"
+    DEGRADED = "degraded"
+    DOWN = "down"
 
 
 @provide_session
@@ -54,14 +70,26 @@ def _job_instance_health(job: Job, heartbeat_field_name: str) -> dict[str, Any]:
     heartbeat = job.latest_heartbeat.isoformat() if job.latest_heartbeat else None
     return {
         "hostname": job.hostname,
-        "status": HEALTHY if job.is_alive() else UNHEALTHY,
         heartbeat_field_name: heartbeat,
     }
 
 
-def _legacy_status(jobs: list[Job]) -> str:
+def _live_jobs(jobs: list[Job]) -> list[Job]:
+    """
+    Narrow unfinished job rows down to the replicas that are actually running.
+
+    ``instances`` describes the deployment as it is now, so a row left behind by a replica that
+    never got to write its ``end_date`` must not be listed: the host it names is gone, and under an
+    orchestrator that assigns a fresh hostname per restart it will never come back. The unfinished
+    rows are still what ``status`` and ``latest_*_heartbeat`` are derived from, so how long a dead
+    component has been silent stays visible even once it drops out of ``instances``.
+    """
+    return [job for job in jobs if job.is_alive()]
+
+
+def _legacy_status(jobs: list[Job]) -> HealthStatus:
     """Top-level status: healthy if any instance is alive."""
-    return HEALTHY if any(job.is_alive() for job in jobs) else UNHEALTHY
+    return HealthStatus.HEALTHY if any(job.is_alive() for job in jobs) else HealthStatus.UNHEALTHY
 
 
 def _triggerer_instance_health(job: Job) -> dict[str, Any]:
@@ -78,19 +106,74 @@ def _dag_processor_instance_health(job: Job) -> dict[str, Any]:
     }
 
 
-def _aggregate_detailed_status(jobs: list[Job]) -> str:
-    """detailed_status: healthy (all alive), degraded (some alive), down (none alive)."""
-    alive_count = sum(1 for job in jobs if job.is_alive())
-    if alive_count == 0:
-        return DOWN
-    if alive_count == len(jobs):
-        return HEALTHY
-    return DEGRADED
+# ``detailed_status`` answers "is every part of this component's work being done", which needs a
+# denominator. Counting job rows cannot supply one: ``end_date`` is only written by a cooperative
+# shutdown, so a replica lost to SIGKILL, an OOM kill, or a node eviction leaves an unfinished row
+# behind forever and a restarted replica adds a second one. The denominator is therefore taken from
+# the declared work partition instead, which is unaffected by how replicas come and go:
+#
+# * Dag processor -- the bundles in ``[dag_processor] dag_bundle_config_list``.
+# * Triggerer -- the team scopes those bundles declare, since a triggerer only picks up triggers for
+#   its own team (see ``Trigger.ids_for_triggerer``).
+# * Scheduler -- schedulers are symmetric, so there is no partition and no partial state to report.
+
+
+def _configured_bundle_teams() -> dict[str, str | None]:
+    """Map every configured Dag bundle to the team owning it, empty when the config is unreadable."""
+    try:
+        return _get_configured_bundle_team_names()
+    except Exception:
+        # A health probe must not fail on malformed bundle config; callers fall back to liveness only.
+        log.warning("Could not read the Dag bundle configuration", exc_info=True)
+        return {}
+
+
+def _liveness_status(jobs: list[Job]) -> DetailedHealthStatus:
+    """Status for a component with no declared work partition: one live replica covers everything."""
+    return DetailedHealthStatus.HEALTHY if any(job.is_alive() for job in jobs) else DetailedHealthStatus.DOWN
+
+
+def _coverage_status(expected: set[Any], covered: set[Any]) -> DetailedHealthStatus:
+    """Status from how much of a component's declared work partition its live replicas cover."""
+    if not expected - covered:
+        return DetailedHealthStatus.HEALTHY
+    if expected & covered:
+        return DetailedHealthStatus.DEGRADED
+    return DetailedHealthStatus.DOWN
+
+
+def _dag_processor_detailed_status(jobs: list[Job]) -> DetailedHealthStatus:
+    """Status from bundle coverage, which partitions processor work with or without multi-team mode."""
+    expected = set(_configured_bundle_teams())
+    if not expected:
+        return _liveness_status(jobs)
+
+    covered: set[str] = set()
+    for job in jobs:
+        if job.is_alive():
+            # A processor started without ``--bundle-name`` parses every configured bundle.
+            covered |= expected if not job.bundle_names else expected & set(job.bundle_names)
+    return _coverage_status(expected, covered)
+
+
+def _triggerer_detailed_status(jobs: list[Job]) -> DetailedHealthStatus:
+    """Status from team coverage, since a triggerer only picks up triggers for its own team."""
+    if not conf.getboolean("core", "multi_team"):
+        # Outside multi-team mode no team filter is applied, so any live triggerer serves every trigger.
+        return _liveness_status(jobs)
+
+    # ``None`` is a scope in its own right: triggers from bundles that declare no team are only picked
+    # up by a triggerer started without ``--team-name``, so it must stay in the expected set.
+    expected = set(_configured_bundle_teams().values())
+    if not expected:
+        return _liveness_status(jobs)
+
+    return _coverage_status(expected, {job.team_name for job in jobs if job.is_alive()})
 
 
 def get_airflow_health() -> dict[str, Any]:
     """Get the health for Airflow metadatabase, scheduler, triggerer, and dag processor."""
-    metadatabase_status = HEALTHY
+    metadatabase_status = HealthStatus.HEALTHY
 
     latest_scheduler_heartbeat = None
     latest_triggerer_heartbeat = None
@@ -100,55 +183,52 @@ def get_airflow_health() -> dict[str, Any]:
     triggerer_instances: list[dict[str, Any]] | None = None
     dag_processor_instances: list[dict[str, Any]] | None = None
 
-    scheduler_status = UNHEALTHY
-    triggerer_status = UNHEALTHY
-    dag_processor_status = UNHEALTHY
+    scheduler_status = HealthStatus.UNHEALTHY
+    triggerer_status = HealthStatus.UNHEALTHY
+    dag_processor_status = HealthStatus.UNHEALTHY
 
-    scheduler_detailed_status = DOWN
-    triggerer_detailed_status = DOWN
-    dag_processor_detailed_status = DOWN
+    scheduler_detailed_status = DetailedHealthStatus.DOWN
+    triggerer_detailed_status = DetailedHealthStatus.DOWN
+    dag_processor_detailed_status = DetailedHealthStatus.DOWN
 
     try:
         scheduler_jobs = get_jobs_health(SchedulerJobRunner)
         scheduler_status = _legacy_status(scheduler_jobs)
-        scheduler_detailed_status = _aggregate_detailed_status(scheduler_jobs)
-        if scheduler_jobs:
+        scheduler_detailed_status = _liveness_status(scheduler_jobs)
+        if scheduler_jobs and scheduler_jobs[0].latest_heartbeat:
+            latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat()
+        if live_scheduler_jobs := _live_jobs(scheduler_jobs):
             scheduler_instances = [
-                _job_instance_health(job, "latest_scheduler_heartbeat") for job in scheduler_jobs
+                _job_instance_health(job, "latest_scheduler_heartbeat") for job in live_scheduler_jobs
             ]
-
-            if scheduler_jobs[0].latest_heartbeat:
-                latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat()
     except Exception:
-        metadatabase_status = UNHEALTHY
+        metadatabase_status = HealthStatus.UNHEALTHY
 
     try:
         triggerer_jobs = get_jobs_health(TriggererJobRunner)
         triggerer_status = _legacy_status(triggerer_jobs)
-        triggerer_detailed_status = _aggregate_detailed_status(triggerer_jobs)
-        if triggerer_jobs:
-            triggerer_instances = [_triggerer_instance_health(job) for job in triggerer_jobs]
-
-            if triggerer_jobs[0].latest_heartbeat:
-                latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat()
+        triggerer_detailed_status = _triggerer_detailed_status(triggerer_jobs)
+        if triggerer_jobs and triggerer_jobs[0].latest_heartbeat:
+            latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat()
+        if live_triggerer_jobs := _live_jobs(triggerer_jobs):
+            triggerer_instances = [_triggerer_instance_health(job) for job in live_triggerer_jobs]
     except Exception:
-        metadatabase_status = UNHEALTHY
-        triggerer_status = UNHEALTHY
-        triggerer_detailed_status = DOWN
+        metadatabase_status = HealthStatus.UNHEALTHY
+        triggerer_status = HealthStatus.UNHEALTHY
+        triggerer_detailed_status = DetailedHealthStatus.DOWN
 
     try:
         dag_processor_jobs = get_jobs_health(DagProcessorJobRunner)
         dag_processor_status = _legacy_status(dag_processor_jobs)
-        dag_processor_detailed_status = _aggregate_detailed_status(dag_processor_jobs)
-        if dag_processor_jobs:
-            dag_processor_instances = [_dag_processor_instance_health(job) for job in dag_processor_jobs]
-
-            if dag_processor_jobs[0].latest_heartbeat:
-                latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat()
+        dag_processor_detailed_status = _dag_processor_detailed_status(dag_processor_jobs)
+        if dag_processor_jobs and dag_processor_jobs[0].latest_heartbeat:
+            latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat()
+        if live_dag_processor_jobs := _live_jobs(dag_processor_jobs):
+            dag_processor_instances = [_dag_processor_instance_health(job) for job in live_dag_processor_jobs]
     except Exception:
-        metadatabase_status = UNHEALTHY
-        dag_processor_status = UNHEALTHY
-        dag_processor_detailed_status = DOWN
+        metadatabase_status = HealthStatus.UNHEALTHY
+        dag_processor_status = HealthStatus.UNHEALTHY
+        dag_processor_detailed_status = DetailedHealthStatus.DOWN
 
     airflow_health_status = {
         "metadatabase": {"status": metadatabase_status},

airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py

@@ -16,31 +16,34 @@
 # under the License.
 from __future__ import annotations
 
+from airflow.api.common.airflow_health import DetailedHealthStatus, HealthStatus
 from airflow.api_fastapi.core_api.base import BaseModel
 
 
 class BaseInfoResponse(BaseModel):
     """Base info serializer for responses."""
 
-    status: str | None
+    status: HealthStatus | None
 
 
-class SchedulerInstanceInfoResponse(BaseInfoResponse):
+# Instances carry no status of their own: only running replicas are listed, and how healthy the set
+# of them is together is what the component's ``status`` and ``detailed_status`` report.
+class SchedulerInstanceInfoResponse(BaseModel):
     """Scheduler instance info serializer for responses."""
 
     hostname: str | None
     latest_scheduler_heartbeat: str | None
 
 
-class TriggererInstanceInfoResponse(BaseInfoResponse):
+class TriggererInstanceInfoResponse(BaseModel):
     """Triggerer instance info serializer for responses."""
 
     hostname: str | None
     latest_triggerer_heartbeat: str | None
     team_name: str | None
 
 
-class DagProcessorInstanceInfoResponse(BaseInfoResponse):
+class DagProcessorInstanceInfoResponse(BaseModel):
     """Dag processor instance info serializer for responses."""
 
     hostname: str | None
@@ -52,23 +55,23 @@ class SchedulerInfoResponse(BaseInfoResponse):
     """Scheduler info serializer for responses."""
 
     latest_scheduler_heartbeat: str | None
-    detailed_status: str | None
+    detailed_status: DetailedHealthStatus | None
     instances: list[SchedulerInstanceInfoResponse] | None = None
 
 
 class TriggererInfoResponse(BaseInfoResponse):
     """Triggerer info serializer for responses."""
 
     latest_triggerer_heartbeat: str | None
-    detailed_status: str | None
+    detailed_status: DetailedHealthStatus | None
     instances: list[TriggererInstanceInfoResponse] | None = None
 
 
 class DagProcessorInfoResponse(BaseInfoResponse):
     """DagProcessor info serializer for responses."""
 
     latest_dag_processor_heartbeat: str | None
-    detailed_status: str | None
+    detailed_status: DetailedHealthStatus | None
     instances: list[DagProcessorInstanceInfoResponse] | None = None