Feature expand health endpoint reporting

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``7a98f1b7dbd3`` (head) | ``c4e7a1f9b2d0`` | ``3.4.0``         | Add index on asset_event (asset_id, partition_key).          |
+| ``f8c2a1d94e03`` (head) | ``7a98f1b7dbd3`` | ``3.4.0``         | Add team_name and bundle_names scope columns to job table.   |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``7a98f1b7dbd3``        | ``c4e7a1f9b2d0`` | ``3.4.0``         | Add index on asset_event (asset_id, partition_key).          |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``c4e7a1f9b2d0``        | ``436dc127462c`` | ``3.4.0``         | Add index on asset.uri.                                      |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

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

@@ -16,77 +16,164 @@
 # 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"
+
+
+@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,
+            )
+            .order_by(Job.latest_heartbeat.desc())
+        )
+    )
+
+
+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:
+    """Top-level status: healthy if any instance is alive."""
+    return HEALTHY if any(job.is_alive() for job in jobs) else UNHEALTHY
+
+
+def _triggerer_instance_health(job: Job) -> dict[str, Any]:
+    return {
+        **_job_instance_health(job, "latest_triggerer_heartbeat"),
+        "team_name": job.team_name,
+    }
+
+
+def _dag_processor_instance_health(job: Job) -> dict[str, Any]:
+    return {
+        **_job_instance_health(job, "latest_dag_processor_heartbeat"),
+        "bundle_names": job.bundle_names,
+    }
+
+
+def _aggregate_detailed_status(jobs: list[Job]) -> str:
+    """detailed_status: healthy (all alive), degraded (some alive), unhealthy (none alive)."""
+    alive_count = sum(1 for job in jobs if job.is_alive())
+    if alive_count == 0:
+        return UNHEALTHY
+    if alive_count == len(jobs):
+        return HEALTHY
+    return DEGRADED
 
 
 def get_airflow_health() -> dict[str, Any]:
-    """Get the health for Airflow metadatabase, scheduler and triggerer."""
+    """Get the health for Airflow metadatabase, scheduler, triggerer, and dag processor."""
     metadatabase_status = HEALTHY
+
     latest_scheduler_heartbeat = None
     latest_triggerer_heartbeat = None
     latest_dag_processor_heartbeat = None
 
+    scheduler_instances: list[dict[str, Any]] | None = None
+    triggerer_instances: list[dict[str, Any]] | None = None
+    dag_processor_instances: list[dict[str, Any]] | None = None
+
     scheduler_status = UNHEALTHY
     triggerer_status: str | None = UNHEALTHY
     dag_processor_status: str | None = UNHEALTHY
 
+    scheduler_detailed_status = UNHEALTHY
+    triggerer_detailed_status: str | None = UNHEALTHY
+    dag_processor_detailed_status: str | None = UNHEALTHY
+
     try:
-        latest_scheduler_job = SchedulerJobRunner.most_recent_job()
+        scheduler_jobs = get_jobs_health(SchedulerJobRunner)
+        if scheduler_jobs:
+            scheduler_status = _legacy_status(scheduler_jobs)  # The top sorted job is confirmed alive
+            scheduler_detailed_status = _aggregate_detailed_status(scheduler_jobs)
+            scheduler_instances = [
+                _job_instance_health(job, "latest_scheduler_heartbeat") for job in scheduler_jobs
+            ]
 
-        if latest_scheduler_job:
-            if latest_scheduler_job.latest_heartbeat:
-                latest_scheduler_heartbeat = latest_scheduler_job.latest_heartbeat.isoformat()
-            if latest_scheduler_job.is_alive():
-                scheduler_status = HEALTHY
+            if scheduler_jobs[0].latest_heartbeat:
+                latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat()
     except Exception:
         metadatabase_status = UNHEALTHY
 
     try:
-        latest_triggerer_job = TriggererJobRunner.most_recent_job()
+        triggerer_jobs = get_jobs_health(TriggererJobRunner)
+        if triggerer_jobs:
+            triggerer_status = _legacy_status(triggerer_jobs)
+            triggerer_detailed_status = _aggregate_detailed_status(triggerer_jobs)
+            triggerer_instances = [_triggerer_instance_health(job) for job in triggerer_jobs]
 
-        if latest_triggerer_job:
-            if latest_triggerer_job.latest_heartbeat:
-                latest_triggerer_heartbeat = latest_triggerer_job.latest_heartbeat.isoformat()
-            if latest_triggerer_job.is_alive():
-                triggerer_status = HEALTHY
+            if triggerer_jobs[0].latest_heartbeat:
+                latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat()
         else:
             triggerer_status = None
+            triggerer_detailed_status = None
     except Exception:
         metadatabase_status = UNHEALTHY
+        triggerer_status = UNHEALTHY
+        triggerer_detailed_status = UNHEALTHY
 
     try:
-        latest_dag_processor_job = DagProcessorJobRunner.most_recent_job()
+        dag_processor_jobs = get_jobs_health(DagProcessorJobRunner)
+        if dag_processor_jobs:
+            dag_processor_status = _legacy_status(dag_processor_jobs)
+            dag_processor_detailed_status = _aggregate_detailed_status(dag_processor_jobs)
+            dag_processor_instances = [_dag_processor_instance_health(job) for job in dag_processor_jobs]
 
-        if latest_dag_processor_job:
-            if latest_dag_processor_job.latest_heartbeat:
-                latest_dag_processor_heartbeat = latest_dag_processor_job.latest_heartbeat.isoformat()
-            if latest_dag_processor_job.is_alive():
-                dag_processor_status = HEALTHY
+            if dag_processor_jobs[0].latest_heartbeat:
+                latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat()
         else:
             dag_processor_status = None
+            dag_processor_detailed_status = None
     except Exception:
         metadatabase_status = UNHEALTHY
+        dag_processor_status = UNHEALTHY
+        dag_processor_detailed_status = UNHEALTHY
 
     airflow_health_status = {
         "metadatabase": {"status": metadatabase_status},
         "scheduler": {
             "status": scheduler_status,
             "latest_scheduler_heartbeat": latest_scheduler_heartbeat,
+            "detailed_status": scheduler_detailed_status,
+            "instances": scheduler_instances,
         },
         "triggerer": {
             "status": triggerer_status,
             "latest_triggerer_heartbeat": latest_triggerer_heartbeat,
+            "detailed_status": triggerer_detailed_status,
+            "instances": triggerer_instances,
         },
         "dag_processor": {
             "status": dag_processor_status,
             "latest_dag_processor_heartbeat": latest_dag_processor_heartbeat,
+            "detailed_status": dag_processor_detailed_status,
+            "instances": dag_processor_instances,
         },
     }
 

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

@@ -37,6 +37,8 @@ class JobResponse(BaseModel):
     executor_class: str | None
     hostname: str | None
     unixname: str | None
+    team_name: str | None = None
+    bundle_names: list[str] | None = None
     dag_display_name: str | None = Field(
         validation_alias=AliasPath("dag_model", "dag_display_name"), default=None
     )

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

@@ -25,22 +25,51 @@ class BaseInfoResponse(BaseModel):
     status: str | None
 
 
+class SchedulerInstanceInfoResponse(BaseInfoResponse):
+    """Scheduler instance info serializer for responses."""
+
+    hostname: str | None
+    latest_scheduler_heartbeat: str | None
+
+
+class TriggererInstanceInfoResponse(BaseInfoResponse):
+    """Triggerer instance info serializer for responses."""
+
+    hostname: str | None
+    latest_triggerer_heartbeat: str | None
+    team_name: str | None
+
+
+class DagProcessorInstanceInfoResponse(BaseInfoResponse):
+    """Dag processor instance info serializer for responses."""
+
+    hostname: str | None
+    latest_dag_processor_heartbeat: str | None
+    bundle_names: list[str] | None
+
+
 class SchedulerInfoResponse(BaseInfoResponse):
     """Scheduler info serializer for responses."""
 
     latest_scheduler_heartbeat: str | None
+    detailed_status: str | None
+    instances: list[SchedulerInstanceInfoResponse] | None = None
 
 
 class TriggererInfoResponse(BaseInfoResponse):
     """Triggerer info serializer for responses."""
 
     latest_triggerer_heartbeat: str | None
+    detailed_status: str | None
+    instances: list[TriggererInstanceInfoResponse] | None = None
 
 
 class DagProcessorInfoResponse(BaseInfoResponse):
     """DagProcessor info serializer for responses."""
 
     latest_dag_processor_heartbeat: str | None
+    detailed_status: str | None
+    instances: list[DagProcessorInstanceInfoResponse] | None = None
 
 
 class HealthInfoResponse(BaseModel):

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -3421,6 +3421,18 @@ components:
           - type: string
           - type: 'null'
           title: Unixname
+        team_name:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Team Name
+        bundle_names:
+          anyOf:
+          - items:
+              type: string
+            type: array
+          - type: 'null'
+          title: Bundle Names
         dag_display_name:
           anyOf:
           - type: string