From f20fb797aa0072095766f904b9da813701c7b880 Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Tue, 30 Jun 2026 16:02:51 -0700 Subject: [PATCH 01/12] add further health endpoint information to instances of each component --- .../src/airflow/api/common/airflow_health.py | 120 ++++++++++++++---- .../core_api/datamodels/monitor.py | 38 ++++++ 2 files changed, 131 insertions(+), 27 deletions(-) diff --git a/airflow-core/src/airflow/api/common/airflow_health.py b/airflow-core/src/airflow/api/common/airflow_health.py index 1086e22df576a..818c5620874c2 100644 --- a/airflow-core/src/airflow/api/common/airflow_health.py +++ b/airflow-core/src/airflow/api/common/airflow_health.py @@ -16,14 +16,54 @@ # 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 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 jobs for the runner class ordered by latest heartbeat descending.""" + return list( + session.scalars( + select(Job).where(Job.job_type == job_runner_class.job_type).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 _aggregate_status(instances: list[dict[str, Any]]) -> str: + statuses = {instance["status"] for instance in instances} + if not statuses or statuses == {HEALTHY}: + return HEALTHY + if statuses == {UNHEALTHY}: + return UNHEALTHY + return DEGRADED + + +def _alive_jobs(jobs: list[Job]) -> list[Job]: + return [job for job in jobs if job.is_alive()] def get_airflow_health() -> dict[str, Any]: @@ -32,47 +72,61 @@ def get_airflow_health() -> dict[str, Any]: 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 + triggerer_status: str | None = None + dag_processor_status: str | None = None try: - latest_scheduler_job = SchedulerJobRunner.most_recent_job() + scheduler_jobs = get_jobs_health(SchedulerJobRunner) + alive_scheduler_jobs = _alive_jobs(scheduler_jobs) + scheduler_instances = [ + _job_instance_health(job, "latest_scheduler_heartbeat") for job in alive_scheduler_jobs + ] + scheduler_status = _aggregate_status(scheduler_instances) if scheduler_instances else UNHEALTHY + schedulers_status = scheduler_status - 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 and 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() - - 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 - else: - triggerer_status = None + triggerer_jobs = get_jobs_health(TriggererJobRunner) + alive_triggerer_jobs = _alive_jobs(triggerer_jobs) + triggerer_instances = [ + { + **_job_instance_health(job, "latest_triggerer_heartbeat"), + "team_name": None, + } + for job in alive_triggerer_jobs + ] + triggerer_status = _aggregate_status(triggerer_instances) if triggerer_instances else (UNHEALTHY if triggerer_jobs else None) + triggerers_status = triggerer_status + + if triggerer_jobs and triggerer_jobs[0].latest_heartbeat: + latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat() except Exception: metadatabase_status = UNHEALTHY + triggerer_status = UNHEALTHY try: - latest_dag_processor_job = DagProcessorJobRunner.most_recent_job() - - 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 - else: - dag_processor_status = None + dag_processor_jobs = get_jobs_health(DagProcessorJobRunner) + alive_dag_processor_jobs = _alive_jobs(dag_processor_jobs) + dag_processor_instances = [ + _job_instance_health(job, "latest_dag_processor_heartbeat") for job in alive_dag_processor_jobs + ] + dag_processor_status = _aggregate_status(dag_processor_instances) if dag_processor_instances else (UNHEALTHY if dag_processor_jobs else None) + dag_processors_status = dag_processor_status + + if dag_processor_jobs and dag_processor_jobs[0].latest_heartbeat: + latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat() except Exception: metadatabase_status = UNHEALTHY + dag_processor_status = UNHEALTHY airflow_health_status = { "metadatabase": {"status": metadatabase_status}, @@ -88,6 +142,18 @@ def get_airflow_health() -> dict[str, Any]: "status": dag_processor_status, "latest_dag_processor_heartbeat": latest_dag_processor_heartbeat, }, + "schedulers": { + "status": schedulers_status, + "instances": scheduler_instances, + }, + "triggerers": { + "status": triggerers_status, + "instances": triggerer_instances, + }, + "dag_processors": { + "status": dag_processors_status, + "instances": dag_processor_instances, + }, } return airflow_health_status diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py index 3bfc5425fd7e6..6060b2e856673 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py @@ -42,6 +42,41 @@ class DagProcessorInfoResponse(BaseInfoResponse): latest_dag_processor_heartbeat: 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 + +class SchedulersInfoResponse(BaseModel): + """Schedulers info serializer for responses.""" + status: str | None + instances: list[SchedulerInstanceInfoResponse] | None = None + +class TriggerersInfoResponse(BaseModel): + """Triggerers info serializer for responses.""" + status: str | None + instances: list[TriggererInstanceInfoResponse] | None = None + +class DagProcessorsInfoResponse(BaseModel): + """Dag processors info serializer for responses.""" + status: str | None + instances: list[DagProcessorInstanceInfoResponse] | None = None class HealthInfoResponse(BaseModel): """Health serializer for responses.""" @@ -50,3 +85,6 @@ class HealthInfoResponse(BaseModel): scheduler: SchedulerInfoResponse triggerer: TriggererInfoResponse dag_processor: DagProcessorInfoResponse | None = None + schedulers: SchedulersInfoResponse | None = None + triggerers: TriggerersInfoResponse | None = None + dag_processors: DagProcessorsInfoResponse | None = None From 69561256b121f91af876306db6f0806b8d6384bc Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Wed, 1 Jul 2026 14:46:38 -0700 Subject: [PATCH 02/12] adjust status reporting to follow the detailed status specificiation --- .../src/airflow/api/common/airflow_health.py | 15 +++-------- .../core_api/datamodels/monitor.py | 25 +++++-------------- 2 files changed, 9 insertions(+), 31 deletions(-) diff --git a/airflow-core/src/airflow/api/common/airflow_health.py b/airflow-core/src/airflow/api/common/airflow_health.py index 818c5620874c2..03e3d29822793 100644 --- a/airflow-core/src/airflow/api/common/airflow_health.py +++ b/airflow-core/src/airflow/api/common/airflow_health.py @@ -133,26 +133,17 @@ def get_airflow_health() -> dict[str, Any]: "scheduler": { "status": scheduler_status, "latest_scheduler_heartbeat": latest_scheduler_heartbeat, + "detailed_status": scheduler_status, }, "triggerer": { "status": triggerer_status, "latest_triggerer_heartbeat": latest_triggerer_heartbeat, + "detailed_status": triggerer_status, }, "dag_processor": { "status": dag_processor_status, "latest_dag_processor_heartbeat": latest_dag_processor_heartbeat, - }, - "schedulers": { - "status": schedulers_status, - "instances": scheduler_instances, - }, - "triggerers": { - "status": triggerers_status, - "instances": triggerer_instances, - }, - "dag_processors": { - "status": dag_processors_status, - "instances": dag_processor_instances, + "detailed_status": dag_processor_status, }, } diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py index 6060b2e856673..abd346ed938ea 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py @@ -29,18 +29,24 @@ 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 SchedulerInstanceInfoResponse(BaseInfoResponse): """Scheduler instance info serializer for responses.""" @@ -56,28 +62,12 @@ class TriggererInstanceInfoResponse(BaseInfoResponse): 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 -class SchedulersInfoResponse(BaseModel): - """Schedulers info serializer for responses.""" - status: str | None - instances: list[SchedulerInstanceInfoResponse] | None = None - -class TriggerersInfoResponse(BaseModel): - """Triggerers info serializer for responses.""" - status: str | None - instances: list[TriggererInstanceInfoResponse] | None = None - -class DagProcessorsInfoResponse(BaseModel): - """Dag processors info serializer for responses.""" - status: str | None - instances: list[DagProcessorInstanceInfoResponse] | None = None - class HealthInfoResponse(BaseModel): """Health serializer for responses.""" @@ -85,6 +75,3 @@ class HealthInfoResponse(BaseModel): scheduler: SchedulerInfoResponse triggerer: TriggererInfoResponse dag_processor: DagProcessorInfoResponse | None = None - schedulers: SchedulersInfoResponse | None = None - triggerers: TriggerersInfoResponse | None = None - dag_processors: DagProcessorsInfoResponse | None = None From 63b87e61e597c53b5a857dafe1e4c8b783265e70 Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Wed, 1 Jul 2026 16:17:02 -0700 Subject: [PATCH 03/12] add detailed instance reporting --- .../src/airflow/api/common/airflow_health.py | 131 +++++++++++------- .../core_api/datamodels/monitor.py | 1 + 2 files changed, 82 insertions(+), 50 deletions(-) diff --git a/airflow-core/src/airflow/api/common/airflow_health.py b/airflow-core/src/airflow/api/common/airflow_health.py index 03e3d29822793..8776a8bee4764 100644 --- a/airflow-core/src/airflow/api/common/airflow_health.py +++ b/airflow-core/src/airflow/api/common/airflow_health.py @@ -36,10 +36,15 @@ @provide_session def get_jobs_health(job_runner_class, *, session: Session = NEW_SESSION) -> list[Job]: - """Return all jobs for the runner class ordered by latest heartbeat descending.""" + """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).order_by(Job.latest_heartbeat.desc()) + select(Job) + .where( + Job.job_type == job_runner_class.job_type, + Job.state == "running", + ) + .order_by(Job.latest_heartbeat.desc()) ) ) @@ -53,98 +58,124 @@ def _job_instance_health(job: Job, heartbeat_field_name: str) -> dict[str, Any]: } -def _aggregate_status(instances: list[dict[str, Any]]) -> str: - statuses = {instance["status"] for instance in instances} - if not statuses or statuses == {HEALTHY}: - return HEALTHY - if statuses == {UNHEALTHY}: - return UNHEALTHY - return DEGRADED +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 _alive_jobs(jobs: list[Job]) -> list[Job]: - return [job for job in jobs if job.is_alive()] - +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 = None - dag_processor_status: str | None = None + 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 + # --- Scheduler --- try: scheduler_jobs = get_jobs_health(SchedulerJobRunner) - alive_scheduler_jobs = _alive_jobs(scheduler_jobs) - scheduler_instances = [ - _job_instance_health(job, "latest_scheduler_heartbeat") for job in alive_scheduler_jobs - ] - scheduler_status = _aggregate_status(scheduler_instances) if scheduler_instances else UNHEALTHY - schedulers_status = scheduler_status - - if scheduler_jobs and scheduler_jobs[0].latest_heartbeat: - latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat() + 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 scheduler_jobs[0].latest_heartbeat: + latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat() except Exception: metadatabase_status = UNHEALTHY + # --- Triggerer --- try: triggerer_jobs = get_jobs_health(TriggererJobRunner) - alive_triggerer_jobs = _alive_jobs(triggerer_jobs) - triggerer_instances = [ - { - **_job_instance_health(job, "latest_triggerer_heartbeat"), - "team_name": None, - } - for job in alive_triggerer_jobs - ] - triggerer_status = _aggregate_status(triggerer_instances) if triggerer_instances else (UNHEALTHY if triggerer_jobs else None) - triggerers_status = triggerer_status - - if triggerer_jobs and triggerer_jobs[0].latest_heartbeat: - latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat() + if triggerer_jobs: + triggerer_status = _legacy_status(triggerer_jobs) + triggerer_detailed_status = _aggregate_detailed_status(triggerer_jobs) + triggerer_instances = [ + { + **_job_instance_health(job, "latest_triggerer_heartbeat"), + "team_name": None, + } + for job in triggerer_jobs + ] + + if triggerer_jobs[0].latest_heartbeat: + latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat() + else: + # If no active/alive running triggerers exist, report old fallback defaults + triggerer_status = None + triggerer_detailed_status = None except Exception: metadatabase_status = UNHEALTHY triggerer_status = UNHEALTHY + triggerer_detailed_status = UNHEALTHY + # --- DAG Processor --- try: dag_processor_jobs = get_jobs_health(DagProcessorJobRunner) - alive_dag_processor_jobs = _alive_jobs(dag_processor_jobs) - dag_processor_instances = [ - _job_instance_health(job, "latest_dag_processor_heartbeat") for job in alive_dag_processor_jobs - ] - dag_processor_status = _aggregate_status(dag_processor_instances) if dag_processor_instances else (UNHEALTHY if dag_processor_jobs else None) - dag_processors_status = dag_processor_status - - if dag_processor_jobs and dag_processor_jobs[0].latest_heartbeat: - latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat() + 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 = [ + { + **_job_instance_health(job, "latest_dag_processor_heartbeat"), + "team_name": None, + } + for job in dag_processor_jobs + ] + + 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_status, + "detailed_status": scheduler_detailed_status, + "instances": scheduler_instances, }, "triggerer": { "status": triggerer_status, "latest_triggerer_heartbeat": latest_triggerer_heartbeat, - "detailed_status": triggerer_status, + "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_status, + "detailed_status": dag_processor_detailed_status, + "instances": dag_processor_instances, }, } - return airflow_health_status + return airflow_health_status \ No newline at end of file diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py index abd346ed938ea..e64268112dc6c 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py @@ -67,6 +67,7 @@ class DagProcessorInstanceInfoResponse(BaseInfoResponse): hostname: str | None latest_dag_processor_heartbeat: str | None + team_name: str | None class HealthInfoResponse(BaseModel): """Health serializer for responses.""" From 6deb15fe22df4bfb1e924c2db7d0b15bb022482c Mon Sep 17 00:00:00 2001 From: Teghveer Singh Ateliey <48258080+Tegh25@users.noreply.github.com> Date: Wed, 8 Jul 2026 18:24:00 -0500 Subject: [PATCH 04/12] Add triggerer team_name, Dag processor bundle_name to Job schema When a triggerer or Dag processor job is created, the team/bundle name field is added into its Job instance. These fields have been added to the Job class as strings and list of strings. --- .../src/airflow/api/common/airflow_health.py | 35 ++++++----- .../api_fastapi/core_api/datamodels/job.py | 2 + .../core_api/datamodels/monitor.py | 43 +++++++------- .../cli/commands/dag_processor_command.py | 2 +- .../airflow/cli/commands/triggerer_command.py | 5 +- airflow-core/src/airflow/jobs/job.py | 8 ++- .../src/airflow/jobs/triggerer_job_runner.py | 2 + .../0125_3_4_0_add_scope_columns_to_job.py | 58 +++++++++++++++++++ 8 files changed, 115 insertions(+), 40 deletions(-) create mode 100644 airflow-core/src/airflow/migrations/versions/0125_3_4_0_add_scope_columns_to_job.py diff --git a/airflow-core/src/airflow/api/common/airflow_health.py b/airflow-core/src/airflow/api/common/airflow_health.py index 8776a8bee4764..3cc485499b38e 100644 --- a/airflow-core/src/airflow/api/common/airflow_health.py +++ b/airflow-core/src/airflow/api/common/airflow_health.py @@ -21,7 +21,7 @@ from sqlalchemy import select from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner -from airflow.jobs.job import Job +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 @@ -42,7 +42,7 @@ def get_jobs_health(job_runner_class, *, session: Session = NEW_SESSION) -> list select(Job) .where( Job.job_type == job_runner_class.job_type, - Job.state == "running", + Job.state == JobState.RUNNING, ) .order_by(Job.latest_heartbeat.desc()) ) @@ -63,6 +63,20 @@ def _legacy_status(jobs: list[Job]) -> str: 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()) @@ -72,6 +86,7 @@ def _aggregate_detailed_status(jobs: list[Job]) -> str: return HEALTHY return DEGRADED + def get_airflow_health() -> dict[str, Any]: """Get the health for Airflow metadatabase, scheduler, triggerer, and dag processor.""" metadatabase_status = HEALTHY @@ -113,13 +128,7 @@ def get_airflow_health() -> dict[str, Any]: if triggerer_jobs: triggerer_status = _legacy_status(triggerer_jobs) triggerer_detailed_status = _aggregate_detailed_status(triggerer_jobs) - triggerer_instances = [ - { - **_job_instance_health(job, "latest_triggerer_heartbeat"), - "team_name": None, - } - for job in 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() @@ -138,13 +147,7 @@ def get_airflow_health() -> dict[str, Any]: 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 = [ - { - **_job_instance_health(job, "latest_dag_processor_heartbeat"), - "team_name": None, - } - for job in 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() diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/job.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/job.py index f97be99191abe..8e316d7b2e381 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/job.py +++ b/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 ) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py index e64268112dc6c..5bf60bd83bd05 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/monitor.py @@ -25,6 +25,29 @@ 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.""" @@ -48,26 +71,6 @@ class DagProcessorInfoResponse(BaseInfoResponse): detailed_status: str | None instances: list[DagProcessorInstanceInfoResponse] | None = 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 - team_name: str | None class HealthInfoResponse(BaseModel): """Health serializer for responses.""" diff --git a/airflow-core/src/airflow/cli/commands/dag_processor_command.py b/airflow-core/src/airflow/cli/commands/dag_processor_command.py index 04e410a5d664b..7e7df45a8d670 100644 --- a/airflow-core/src/airflow/cli/commands/dag_processor_command.py +++ b/airflow-core/src/airflow/cli/commands/dag_processor_command.py @@ -38,7 +38,7 @@ def _create_dag_processor_job_runner(args: Any) -> DagProcessorJobRunner: if args.bundle_name: cli_utils.validate_dag_bundle_arg(args.bundle_name) return DagProcessorJobRunner( - job=Job(), + job=Job(bundle_names=args.bundle_name), processor=DagFileProcessorManager( max_runs=args.num_runs, bundle_names_to_parse=args.bundle_name, diff --git a/airflow-core/src/airflow/cli/commands/triggerer_command.py b/airflow-core/src/airflow/cli/commands/triggerer_command.py index 7e1d98f44dba5..e01b43f70b638 100644 --- a/airflow-core/src/airflow/cli/commands/triggerer_command.py +++ b/airflow-core/src/airflow/cli/commands/triggerer_command.py @@ -62,7 +62,10 @@ def triggerer_run( set_component_mp_start_method("triggerer") with _serve_logs(skip_serve_logs): triggerer_job_runner = TriggererJobRunner( - job=Job(heartrate=triggerer_heartrate), capacity=capacity, queues=queues, team_name=team_name + job=Job(heartrate=triggerer_heartrate, team_name=team_name), + capacity=capacity, + queues=queues, + team_name=team_name, ) run_job(job=triggerer_job_runner.job, execute_callable=triggerer_job_runner._execute) diff --git a/airflow-core/src/airflow/jobs/job.py b/airflow-core/src/airflow/jobs/job.py index b19cd103de908..fa59a440abc5c 100644 --- a/airflow-core/src/airflow/jobs/job.py +++ b/airflow-core/src/airflow/jobs/job.py @@ -24,7 +24,7 @@ from time import sleep from typing import TYPE_CHECKING, NoReturn -from sqlalchemy import Index, Integer, String, case, select +from sqlalchemy import ForeignKey, Index, Integer, String, case, select from sqlalchemy.exc import OperationalError from sqlalchemy.orm import Mapped, backref, foreign, mapped_column, relationship from sqlalchemy.orm.session import make_transient @@ -40,7 +40,7 @@ from airflow.utils.net import get_hostname from airflow.utils.platform import getuser from airflow.utils.session import NEW_SESSION, create_session, provide_session -from airflow.utils.sqlalchemy import UtcDateTime +from airflow.utils.sqlalchemy import ExtendedJSON, UtcDateTime class JobState(str, Enum): @@ -101,6 +101,10 @@ class Job(Base, LoggingMixin): executor_class: Mapped[str | None] = mapped_column(String(500)) hostname: Mapped[str | None] = mapped_column(String(500)) unixname: Mapped[str | None] = mapped_column(String(1000)) + team_name: Mapped[str | None] = mapped_column( + String(50), ForeignKey("team.name", ondelete="SET NULL"), nullable=True + ) + bundle_names: Mapped[list[str] | None] = mapped_column(ExtendedJSON, nullable=True) __table_args__ = ( Index("job_type_heart", job_type, latest_heartbeat), diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index f851e491a69dd..786cd36a22366 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py @@ -213,6 +213,8 @@ def __init__( raise ValueError(f"Capacity number {capacity!r} is invalid") self.queues = queues self.team_name = team_name + if job.team_name is None: + job.team_name = team_name # Set up only when _execute() starts the subprocess; keep it defined so that # signal handlers (or other code) firing before startup don't hit AttributeError. self.trigger_runner: TriggerRunnerSupervisor | None = None diff --git a/airflow-core/src/airflow/migrations/versions/0125_3_4_0_add_scope_columns_to_job.py b/airflow-core/src/airflow/migrations/versions/0125_3_4_0_add_scope_columns_to_job.py new file mode 100644 index 0000000000000..a95cd23358de2 --- /dev/null +++ b/airflow-core/src/airflow/migrations/versions/0125_3_4_0_add_scope_columns_to_job.py @@ -0,0 +1,58 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Add team_name and bundle_names scope columns to job table. + +Revision ID: f8c2a1d94e03 +Revises: 5a5d3253e946 +Create Date: 2026-07-08 02:00:00.000000 + +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +from airflow.utils.sqlalchemy import ExtendedJSON + +# revision identifiers, used by Alembic. +revision = "f8c2a1d94e03" +down_revision = "5a5d3253e946" +branch_labels = None +depends_on = None +airflow_version = "3.4.0" + + +def upgrade(): + """Add triggerer team and dag-processor bundle scope columns to job table.""" + with op.batch_alter_table("job", schema=None) as batch_op: + batch_op.add_column(sa.Column("team_name", sa.String(length=50), nullable=True)) + batch_op.add_column(sa.Column("bundle_names", ExtendedJSON(), nullable=True)) + batch_op.create_foreign_key( + batch_op.f("job_team_name_fkey"), "team", ["team_name"], ["name"], ondelete="SET NULL" + ) + + +def downgrade(): + """Remove scope columns from job table.""" + with op.batch_alter_table("job", schema=None) as batch_op: + batch_op.drop_constraint(batch_op.f("job_team_name_fkey"), type_="foreignkey") + batch_op.drop_column("bundle_names") + batch_op.drop_column("team_name") From 6c316daa7a142ac8ca864775214964100cdb94fc Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Fri, 24 Jul 2026 13:19:39 -0700 Subject: [PATCH 05/12] removed redundant comments and run prek --- .../src/airflow/api/common/airflow_health.py | 14 +++++--------- 1 file changed, 5 insertions(+), 9 deletions(-) diff --git a/airflow-core/src/airflow/api/common/airflow_health.py b/airflow-core/src/airflow/api/common/airflow_health.py index 3cc485499b38e..ed0c8d5b907f0 100644 --- a/airflow-core/src/airflow/api/common/airflow_health.py +++ b/airflow-core/src/airflow/api/common/airflow_health.py @@ -90,7 +90,7 @@ def _aggregate_detailed_status(jobs: list[Job]) -> str: def get_airflow_health() -> dict[str, Any]: """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 @@ -107,7 +107,6 @@ def get_airflow_health() -> dict[str, Any]: triggerer_detailed_status: str | None = UNHEALTHY dag_processor_detailed_status: str | None = UNHEALTHY - # --- Scheduler --- try: scheduler_jobs = get_jobs_health(SchedulerJobRunner) if scheduler_jobs: @@ -116,24 +115,22 @@ def get_airflow_health() -> dict[str, Any]: scheduler_instances = [ _job_instance_health(job, "latest_scheduler_heartbeat") for job in scheduler_jobs ] - + if scheduler_jobs[0].latest_heartbeat: latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat() except Exception: metadatabase_status = UNHEALTHY - # --- Triggerer --- try: 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 triggerer_jobs[0].latest_heartbeat: latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat() else: - # If no active/alive running triggerers exist, report old fallback defaults triggerer_status = None triggerer_detailed_status = None except Exception: @@ -141,14 +138,13 @@ def get_airflow_health() -> dict[str, Any]: triggerer_status = UNHEALTHY triggerer_detailed_status = UNHEALTHY - # --- DAG Processor --- try: 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 dag_processor_jobs[0].latest_heartbeat: latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat() else: @@ -181,4 +177,4 @@ def get_airflow_health() -> dict[str, Any]: }, } - return airflow_health_status \ No newline at end of file + return airflow_health_status From a5f461abfa418108e57a6d794aa9a5d13cd8d4d3 Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Fri, 24 Jul 2026 16:23:15 -0700 Subject: [PATCH 06/12] resolve alembic_version dual head by pointing to new head --- ...columns_to_job.py => 0128_3_4_0_add_scope_columns_to_job.py} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename airflow-core/src/airflow/migrations/versions/{0125_3_4_0_add_scope_columns_to_job.py => 0128_3_4_0_add_scope_columns_to_job.py} (98%) diff --git a/airflow-core/src/airflow/migrations/versions/0125_3_4_0_add_scope_columns_to_job.py b/airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_scope_columns_to_job.py similarity index 98% rename from airflow-core/src/airflow/migrations/versions/0125_3_4_0_add_scope_columns_to_job.py rename to airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_scope_columns_to_job.py index a95cd23358de2..e4c0c19f10876 100644 --- a/airflow-core/src/airflow/migrations/versions/0125_3_4_0_add_scope_columns_to_job.py +++ b/airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_scope_columns_to_job.py @@ -34,7 +34,7 @@ # revision identifiers, used by Alembic. revision = "f8c2a1d94e03" -down_revision = "5a5d3253e946" +down_revision = "7a98f1b7dbd3" branch_labels = None depends_on = None airflow_version = "3.4.0" From 68496d58bb1fbe7abae542495f7d3fa0cac75b58 Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Sat, 25 Jul 2026 10:15:00 -0700 Subject: [PATCH 07/12] update old endpoint contracts --- .../unit/api/common/test_airflow_health.py | 153 +++++++++++------- .../core_api/routes/public/test_monitor.py | 11 +- 2 files changed, 107 insertions(+), 57 deletions(-) diff --git a/airflow-core/tests/unit/api/common/test_airflow_health.py b/airflow-core/tests/unit/api/common/test_airflow_health.py index 43c17e4108805..ddbdfa1e36d39 100644 --- a/airflow-core/tests/unit/api/common/test_airflow_health.py +++ b/airflow-core/tests/unit/api/common/test_airflow_health.py @@ -31,61 +31,72 @@ pytestmark = pytest.mark.db_test -@patch("airflow.api.common.airflow_health.SchedulerJobRunner.most_recent_job", return_value=None) -@patch("airflow.api.common.airflow_health.TriggererJobRunner.most_recent_job", return_value=None) -@patch("airflow.api.common.airflow_health.DagProcessorJobRunner.most_recent_job", return_value=None) -def test_get_airflow_health_only_metadatabase_healthy( - latest_scheduler_job_mock, - latest_triggerer_job_mock, - latest_dag_processor_job_mock, -): +@patch("airflow.api.common.airflow_health.get_jobs_health") +def test_get_airflow_health_only_metadatabase_healthy(mock_get_jobs_health): + mock_get_jobs_health.side_effect = [[], [], []] health_status = get_airflow_health() expected_status = { "metadatabase": {"status": HEALTHY}, - "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None}, - "triggerer": {"status": None, "latest_triggerer_heartbeat": None}, - "dag_processor": {"status": None, "latest_dag_processor_heartbeat": None}, + "scheduler": { + "status": UNHEALTHY, + "latest_scheduler_heartbeat": None, + "detailed_status": UNHEALTHY, + "instances": None, + }, + "triggerer": { + "status": None, + "latest_triggerer_heartbeat": None, + "detailed_status": None, + "instances": None, + }, + "dag_processor": { + "status": None, + "latest_dag_processor_heartbeat": None, + "detailed_status": None, + "instances": None, + }, } assert health_status == expected_status -@patch("airflow.api.common.airflow_health.SchedulerJobRunner.most_recent_job", return_value=Exception) -@patch("airflow.api.common.airflow_health.TriggererJobRunner.most_recent_job", return_value=Exception) -@patch("airflow.api.common.airflow_health.DagProcessorJobRunner.most_recent_job", return_value=Exception) -def test_get_airflow_health_metadatabase_unhealthy( - latest_scheduler_job_mock, - latest_triggerer_job_mock, - latest_dag_processor_job_mock, -): +@patch("airflow.api.common.airflow_health.get_jobs_health", side_effect=Exception) +def test_get_airflow_health_metadatabase_unhealthy(mock_get_jobs_health): health_status = get_airflow_health() expected_status = { "metadatabase": {"status": UNHEALTHY}, - "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None}, - "triggerer": {"status": UNHEALTHY, "latest_triggerer_heartbeat": None}, - "dag_processor": {"status": UNHEALTHY, "latest_dag_processor_heartbeat": None}, + "scheduler": { + "status": UNHEALTHY, + "latest_scheduler_heartbeat": None, + "detailed_status": UNHEALTHY, + "instances": None, + }, + "triggerer": { + "status": UNHEALTHY, + "latest_triggerer_heartbeat": None, + "detailed_status": UNHEALTHY, + "instances": None, + }, + "dag_processor": { + "status": UNHEALTHY, + "latest_dag_processor_heartbeat": None, + "detailed_status": UNHEALTHY, + "instances": None, + }, } assert health_status == expected_status - LATEST_SCHEDULER_JOB_MOCK = MagicMock(spec=Job) -LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime.now() +LATEST_SCHEDULER_JOB_MOCK.hostname = "scheduler-host" +LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime(2024, 1, 1) LATEST_SCHEDULER_JOB_MOCK.is_alive = MagicMock(return_value=True) -@patch( - "airflow.api.common.airflow_health.SchedulerJobRunner.most_recent_job", - return_value=LATEST_SCHEDULER_JOB_MOCK, -) -@patch("airflow.api.common.airflow_health.TriggererJobRunner.most_recent_job", return_value=None) -@patch("airflow.api.common.airflow_health.DagProcessorJobRunner.most_recent_job", return_value=None) -def test_get_airflow_health_scheduler_healthy_no_triggerer( - latest_scheduler_job_mock, - latest_triggerer_job_mock, - latest_dag_processor_job_mock, -): +@patch("airflow.api.common.airflow_health.get_jobs_health") +def test_get_airflow_health_scheduler_healthy_no_triggerer(mock_get_jobs_health): + mock_get_jobs_health.side_effect = [[LATEST_SCHEDULER_JOB_MOCK], [], []] health_status = get_airflow_health() expected_status = { @@ -93,49 +104,83 @@ def test_get_airflow_health_scheduler_healthy_no_triggerer( "scheduler": { "status": HEALTHY, "latest_scheduler_heartbeat": LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(), + "detailed_status": HEALTHY, + "instances": [ + { + "hostname": LATEST_SCHEDULER_JOB_MOCK.hostname, + "status": HEALTHY, + "latest_scheduler_heartbeat": LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(), + } + ], + }, + "triggerer": { + "status": None, + "latest_triggerer_heartbeat": None, + "detailed_status": None, + "instances": None, + }, + "dag_processor": { + "status": None, + "latest_dag_processor_heartbeat": None, + "detailed_status": None, + "instances": None, }, - "triggerer": {"status": None, "latest_triggerer_heartbeat": None}, - "dag_processor": {"status": None, "latest_dag_processor_heartbeat": None}, } assert health_status == expected_status LATEST_TRIGGERER_JOB_MOCK = MagicMock(spec=Job) -LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat = datetime.now() +LATEST_TRIGGERER_JOB_MOCK.hostname = "triggerer-host" +LATEST_TRIGGERER_JOB_MOCK.team_name = "team-a" +LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat = datetime(2024, 1, 2) LATEST_TRIGGERER_JOB_MOCK.is_alive = MagicMock(return_value=True) LATEST_DAG_PROCESSOR_JOB_MOCK = MagicMock(spec=Job) -LATEST_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat = datetime.now() +LATEST_DAG_PROCESSOR_JOB_MOCK.hostname = "dag-processor-host" +LATEST_DAG_PROCESSOR_JOB_MOCK.bundle_names = ["bundle-a"] +LATEST_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat = datetime(2024, 1, 3) LATEST_DAG_PROCESSOR_JOB_MOCK.is_alive = MagicMock(return_value=True) -@patch("airflow.api.common.airflow_health.SchedulerJobRunner.most_recent_job", return_value=None) -@patch( - "airflow.api.common.airflow_health.TriggererJobRunner.most_recent_job", - return_value=LATEST_TRIGGERER_JOB_MOCK, -) -@patch( - "airflow.api.common.airflow_health.DagProcessorJobRunner.most_recent_job", - return_value=LATEST_DAG_PROCESSOR_JOB_MOCK, -) -def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record( - latest_scheduler_job_mock, - latest_triggerer_job_mock, - latest_dag_processor_job_mock, -): +@patch("airflow.api.common.airflow_health.get_jobs_health") +def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record(mock_get_jobs_health): + mock_get_jobs_health.side_effect = [[], [LATEST_TRIGGERER_JOB_MOCK], [LATEST_DAG_PROCESSOR_JOB_MOCK]] health_status = get_airflow_health() expected_status = { "metadatabase": {"status": HEALTHY}, - "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None}, + "scheduler": { + "status": UNHEALTHY, + "latest_scheduler_heartbeat": None, + "detailed_status": UNHEALTHY, + "instances": None, + }, "triggerer": { "status": HEALTHY, "latest_triggerer_heartbeat": LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(), + "detailed_status": HEALTHY, + "instances": [ + { + "hostname": LATEST_TRIGGERER_JOB_MOCK.hostname, + "status": HEALTHY, + "latest_triggerer_heartbeat": LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(), + "team_name": LATEST_TRIGGERER_JOB_MOCK.team_name, + } + ], }, "dag_processor": { "status": HEALTHY, "latest_dag_processor_heartbeat": LATEST_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat.isoformat(), + "detailed_status": HEALTHY, + "instances": [ + { + "hostname": LATEST_DAG_PROCESSOR_JOB_MOCK.hostname, + "status": HEALTHY, + "latest_dag_processor_heartbeat": LATEST_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat.isoformat(), + "bundle_names": LATEST_DAG_PROCESSOR_JOB_MOCK.bundle_names, + } + ], }, } diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_monitor.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_monitor.py index d4b3d97f93252..db7ba5481dfdf 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_monitor.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_monitor.py @@ -94,9 +94,9 @@ def test_unhealthy_scheduler_no_job(self, test_client): assert body["scheduler"]["status"] == "unhealthy" assert body["scheduler"]["latest_scheduler_heartbeat"] is None - @mock.patch.object(SchedulerJobRunner, "most_recent_job") - def test_unhealthy_metadatabase_status(self, most_recent_job_mock, test_client): - most_recent_job_mock.side_effect = Exception + @mock.patch("airflow.api.common.airflow_health.get_jobs_health") + def test_unhealthy_metadatabase_status(self, mock_get_jobs_health, test_client): + mock_get_jobs_health.side_effect = Exception response = test_client.get("/monitor/health") assert response.status_code == 200 @@ -112,14 +112,17 @@ def test_health_with_dag_processor(self, mock_get_airflow_health, test_client): "scheduler": { "status": HEALTHY, "latest_scheduler_heartbeat": "2024-11-23T11:09:16.663124+00:00", + "detailed_status": HEALTHY, }, "triggerer": { "status": HEALTHY, "latest_triggerer_heartbeat": "2024-11-23T11:09:15.815483+00:00", + "detailed_status": HEALTHY, }, "dag_processor": { "status": HEALTHY, "latest_dag_processor_heartbeat": "2024-11-23T11:09:15.815483+00:00", + "detailed_status": HEALTHY, }, } @@ -140,10 +143,12 @@ def test_health_without_dag_processor(self, mock_get_airflow_health, test_client "scheduler": { "status": HEALTHY, "latest_scheduler_heartbeat": "2024-11-23T11:09:16.663124+00:00", + "detailed_status": HEALTHY, }, "triggerer": { "status": HEALTHY, "latest_triggerer_heartbeat": "2024-11-23T11:09:15.815483+00:00", + "detailed_status": HEALTHY, }, } From f84a094e59470f422a311cbe938bed5607a04788 Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Sat, 25 Jul 2026 10:25:48 -0700 Subject: [PATCH 08/12] update expected output contract --- .../tests/unit/api_fastapi/core_api/routes/public/test_job.py | 2 ++ .../api_fastapi/core_api/routes/public/test_task_instances.py | 2 ++ 2 files changed, 4 insertions(+) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py index 8e9538d3f634d..ef8285620a904 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py @@ -157,6 +157,7 @@ def test_get_jobs( assert len(matched) == 1 expected_job = { "id": matched[0].id, + "bundle_names": None, "dag_display_name": None, "dag_id": None, "state": matched[0].state, @@ -166,6 +167,7 @@ def test_get_jobs( "latest_heartbeat": from_datetime_to_zulu(matched[0].latest_heartbeat), "executor_class": None, "hostname": matched[0].hostname, + "team_name": None, "unixname": matched[0].unixname, } assert resp_job == expected_job diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index f68acf75d60e0..39d96f345b5fc 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -435,11 +435,13 @@ def test_should_respond_200_with_task_state_in_deferred(self, test_client, sessi "queue": None, }, "triggerer_job": { + "bundle_names": None, "dag_display_name": None, "dag_id": None, "end_date": None, "job_type": "TriggererJob", "state": "running", + "team_name": None, "unixname": getuser(), }, "team_name": None, From 89d2c69286cf551a62ec554d17bc642c9b33b965 Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Sat, 25 Jul 2026 10:26:04 -0700 Subject: [PATCH 09/12] prek run --- airflow-core/tests/unit/api/common/test_airflow_health.py | 1 + 1 file changed, 1 insertion(+) diff --git a/airflow-core/tests/unit/api/common/test_airflow_health.py b/airflow-core/tests/unit/api/common/test_airflow_health.py index ddbdfa1e36d39..bd925426be0d7 100644 --- a/airflow-core/tests/unit/api/common/test_airflow_health.py +++ b/airflow-core/tests/unit/api/common/test_airflow_health.py @@ -88,6 +88,7 @@ def test_get_airflow_health_metadatabase_unhealthy(mock_get_jobs_health): assert health_status == expected_status + LATEST_SCHEDULER_JOB_MOCK = MagicMock(spec=Job) LATEST_SCHEDULER_JOB_MOCK.hostname = "scheduler-host" LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime(2024, 1, 1) From 40b99d3847eebbf8cb296fe23354e5b80d4e9cad Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Sat, 25 Jul 2026 11:20:25 -0700 Subject: [PATCH 10/12] update revision head map --- airflow-core/src/airflow/utils/db.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow-core/src/airflow/utils/db.py b/airflow-core/src/airflow/utils/db.py index 1076715ee7180..c8341df7525fc 100644 --- a/airflow-core/src/airflow/utils/db.py +++ b/airflow-core/src/airflow/utils/db.py @@ -117,7 +117,7 @@ class MappedClassProtocol(Protocol): "3.1.8": "509b94a1042d", "3.2.0": "1d6611b6ab7c", "3.3.0": "d2f4e1b3c5a7", - "3.4.0": "7a98f1b7dbd3", + "3.4.0": "f8c2a1d94e03", } # Prefix used to identify tables holding data moved during migration. From 9778c63ca184b2641d6a19dceba537756d51882c Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Sat, 25 Jul 2026 11:44:27 -0700 Subject: [PATCH 11/12] add db migration prek runs --- airflow-core/docs/migrations-ref.rst | 4 +- .../core_api/openapi/_private_ui.yaml | 12 ++ .../openapi/v2-rest-api-generated.yaml | 137 ++++++++++++++++++ .../0128_3_4_0_add_scope_columns_to_job.py | 2 +- .../airflowctl/api/datamodels/generated.py | 76 +++++++--- 5 files changed, 212 insertions(+), 19 deletions(-) diff --git a/airflow-core/docs/migrations-ref.rst b/airflow-core/docs/migrations-ref.rst index dfc96be79019e..db6934c34f812 100644 --- a/airflow-core/docs/migrations-ref.rst +++ b/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. | +-------------------------+------------------+-------------------+--------------------------------------------------------------+ diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml index ca8f729998250..7d09313e802bf 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml +++ b/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 diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml index 1d5c1b6abf705..bae7a39da7849 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml @@ -14547,12 +14547,57 @@ components: - type: string - type: 'null' title: Latest Dag Processor Heartbeat + detailed_status: + anyOf: + - type: string + - type: 'null' + title: Detailed Status + instances: + anyOf: + - items: + $ref: '#/components/schemas/DagProcessorInstanceInfoResponse' + type: array + - type: 'null' + title: Instances type: object required: - status - latest_dag_processor_heartbeat + - detailed_status title: DagProcessorInfoResponse description: DagProcessor info serializer for responses. + DagProcessorInstanceInfoResponse: + properties: + status: + anyOf: + - type: string + - type: 'null' + title: Status + hostname: + anyOf: + - type: string + - type: 'null' + title: Hostname + latest_dag_processor_heartbeat: + anyOf: + - type: string + - type: 'null' + title: Latest Dag Processor Heartbeat + bundle_names: + anyOf: + - items: + type: string + type: array + - type: 'null' + title: Bundle Names + type: object + required: + - status + - hostname + - latest_dag_processor_heartbeat + - bundle_names + title: DagProcessorInstanceInfoResponse + description: Dag processor instance info serializer for responses. DagRunAssetReference: properties: run_id: @@ -15418,6 +15463,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 @@ -15994,12 +16051,49 @@ components: - type: string - type: 'null' title: Latest Scheduler Heartbeat + detailed_status: + anyOf: + - type: string + - type: 'null' + title: Detailed Status + instances: + anyOf: + - items: + $ref: '#/components/schemas/SchedulerInstanceInfoResponse' + type: array + - type: 'null' + title: Instances type: object required: - status - latest_scheduler_heartbeat + - detailed_status title: SchedulerInfoResponse description: Scheduler info serializer for responses. + SchedulerInstanceInfoResponse: + properties: + status: + anyOf: + - type: string + - type: 'null' + title: Status + hostname: + anyOf: + - type: string + - type: 'null' + title: Hostname + latest_scheduler_heartbeat: + anyOf: + - type: string + - type: 'null' + title: Latest Scheduler Heartbeat + type: object + required: + - status + - hostname + - latest_scheduler_heartbeat + title: SchedulerInstanceInfoResponse + description: Scheduler instance info serializer for responses. StructuredLogMessage: properties: timestamp: @@ -17112,12 +17206,55 @@ components: - type: string - type: 'null' title: Latest Triggerer Heartbeat + detailed_status: + anyOf: + - type: string + - type: 'null' + title: Detailed Status + instances: + anyOf: + - items: + $ref: '#/components/schemas/TriggererInstanceInfoResponse' + type: array + - type: 'null' + title: Instances type: object required: - status - latest_triggerer_heartbeat + - detailed_status title: TriggererInfoResponse description: Triggerer info serializer for responses. + TriggererInstanceInfoResponse: + properties: + status: + anyOf: + - type: string + - type: 'null' + title: Status + hostname: + anyOf: + - type: string + - type: 'null' + title: Hostname + latest_triggerer_heartbeat: + anyOf: + - type: string + - type: 'null' + title: Latest Triggerer Heartbeat + team_name: + anyOf: + - type: string + - type: 'null' + title: Team Name + type: object + required: + - status + - hostname + - latest_triggerer_heartbeat + - team_name + title: TriggererInstanceInfoResponse + description: Triggerer instance info serializer for responses. UpdateHITLDetailPayload: properties: chosen_options: diff --git a/airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_scope_columns_to_job.py b/airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_scope_columns_to_job.py index e4c0c19f10876..b9968d8c2146c 100644 --- a/airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_scope_columns_to_job.py +++ b/airflow-core/src/airflow/migrations/versions/0128_3_4_0_add_scope_columns_to_job.py @@ -20,7 +20,7 @@ Add team_name and bundle_names scope columns to job table. Revision ID: f8c2a1d94e03 -Revises: 5a5d3253e946 +Revises: 7a98f1b7dbd3 Create Date: 2026-07-08 02:00:00.000000 """ diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index d7b63613a8608..99ee88f825353 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py @@ -507,15 +507,17 @@ class DAGTagCollectionResponse(BaseModel): total_entries: Annotated[int, Field(title="Total Entries")] -class DagProcessorInfoResponse(BaseModel): +class DagProcessorInstanceInfoResponse(BaseModel): """ - DagProcessor info serializer for responses. + Dag processor instance info serializer for responses. """ status: Annotated[str | None, Field(title="Status")] = None + hostname: Annotated[str | None, Field(title="Hostname")] = None latest_dag_processor_heartbeat: Annotated[str | None, Field(title="Latest Dag Processor Heartbeat")] = ( None ) + bundle_names: Annotated[list[str] | None, Field(title="Bundle Names")] = None class DagRunAssetReference(BaseModel): @@ -795,6 +797,8 @@ class JobResponse(BaseModel): executor_class: Annotated[str | None, Field(title="Executor Class")] = None hostname: Annotated[str | None, Field(title="Hostname")] = None unixname: Annotated[str | None, Field(title="Unixname")] = None + team_name: Annotated[str | None, Field(title="Team Name")] = None + bundle_names: Annotated[list[str] | None, Field(title="Bundle Names")] = None dag_display_name: Annotated[str | None, Field(title="Dag Display Name")] = None @@ -969,12 +973,13 @@ class ReprocessBehavior(str, Enum): NONE = "none" -class SchedulerInfoResponse(BaseModel): +class SchedulerInstanceInfoResponse(BaseModel): """ - Scheduler info serializer for responses. + Scheduler instance info serializer for responses. """ status: Annotated[str | None, Field(title="Status")] = None + hostname: Annotated[str | None, Field(title="Hostname")] = None latest_scheduler_heartbeat: Annotated[str | None, Field(title="Latest Scheduler Heartbeat")] = None @@ -1181,13 +1186,15 @@ class TriggerResponse(BaseModel): triggerer_id: Annotated[int | None, Field(title="Triggerer Id")] = None -class TriggererInfoResponse(BaseModel): +class TriggererInstanceInfoResponse(BaseModel): """ - Triggerer info serializer for responses. + Triggerer instance info serializer for responses. """ status: Annotated[str | None, Field(title="Status")] = None + hostname: Annotated[str | None, Field(title="Hostname")] = None latest_triggerer_heartbeat: Annotated[str | None, Field(title="Latest Triggerer Heartbeat")] = None + team_name: Annotated[str | None, Field(title="Team Name")] = None class UpdateHITLDetailPayload(BaseModel): @@ -1918,6 +1925,19 @@ class DAGWarningResponse(BaseModel): dag_display_name: Annotated[str, Field(title="Dag Display Name")] +class DagProcessorInfoResponse(BaseModel): + """ + DagProcessor info serializer for responses. + """ + + status: Annotated[str | None, Field(title="Status")] = None + latest_dag_processor_heartbeat: Annotated[str | None, Field(title="Latest Dag Processor Heartbeat")] = ( + None + ) + detailed_status: Annotated[str | None, Field(title="Detailed Status")] = None + instances: Annotated[list[DagProcessorInstanceInfoResponse] | None, Field(title="Instances")] = None + + class DagStatsResponse(BaseModel): """ Dag Stats serializer for responses. @@ -1961,17 +1981,6 @@ class HTTPValidationError(BaseModel): detail: Annotated[list[ValidationError] | None, Field(title="Detail")] = None -class HealthInfoResponse(BaseModel): - """ - Health serializer for responses. - """ - - metadatabase: BaseInfoResponse - scheduler: SchedulerInfoResponse - triggerer: TriggererInfoResponse - dag_processor: DagProcessorInfoResponse | None = None - - class ImportErrorCollectionResponse(BaseModel): """ Import Error Collection Response. @@ -2072,6 +2081,17 @@ class QueuedEventCollectionResponse(BaseModel): total_entries: Annotated[int, Field(title="Total Entries")] +class SchedulerInfoResponse(BaseModel): + """ + Scheduler info serializer for responses. + """ + + status: Annotated[str | None, Field(title="Status")] = None + latest_scheduler_heartbeat: Annotated[str | None, Field(title="Latest Scheduler Heartbeat")] = None + detailed_status: Annotated[str | None, Field(title="Detailed Status")] = None + instances: Annotated[list[SchedulerInstanceInfoResponse] | None, Field(title="Instances")] = None + + class TaskDependencyCollectionResponse(BaseModel): """ Task scheduling dependencies collection serializer for responses. @@ -2200,6 +2220,17 @@ class TaskStateStoreCollectionResponse(BaseModel): total_entries: Annotated[int, Field(title="Total Entries")] +class TriggererInfoResponse(BaseModel): + """ + Triggerer info serializer for responses. + """ + + status: Annotated[str | None, Field(title="Status")] = None + latest_triggerer_heartbeat: Annotated[str | None, Field(title="Latest Triggerer Heartbeat")] = None + detailed_status: Annotated[str | None, Field(title="Detailed Status")] = None + instances: Annotated[list[TriggererInstanceInfoResponse] | None, Field(title="Instances")] = None + + class VariableCollectionResponse(BaseModel): """ Variable Collection serializer for responses. @@ -2458,6 +2489,17 @@ class HITLDetailHistory(BaseModel): task_instance: TaskInstanceHistoryResponse +class HealthInfoResponse(BaseModel): + """ + Health serializer for responses. + """ + + metadatabase: BaseInfoResponse + scheduler: SchedulerInfoResponse + triggerer: TriggererInfoResponse + dag_processor: DagProcessorInfoResponse | None = None + + class PluginCollectionResponse(BaseModel): """ Plugin Collection serializer. From 2e5cbf9859ec8c969e76978f442ec7df24cf9636 Mon Sep 17 00:00:00 2001 From: Jung-Hyun Andrew Kim Date: Sat, 25 Jul 2026 15:18:15 -0700 Subject: [PATCH 12/12] prek schema checks --- .../ui/openapi-gen/requests/schemas.gen.ts | 257 +++++++++++++++++- .../ui/openapi-gen/requests/types.gen.ts | 37 +++ 2 files changed, 291 insertions(+), 3 deletions(-) diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index 2e4c24e117063..1e7553c3fc91b 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -4388,14 +4388,95 @@ export const $DagProcessorInfoResponse = { } ], title: 'Latest Dag Processor Heartbeat' + }, + detailed_status: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Detailed Status' + }, + instances: { + anyOf: [ + { + items: { + '$ref': '#/components/schemas/DagProcessorInstanceInfoResponse' + }, + type: 'array' + }, + { + type: 'null' + } + ], + title: 'Instances' } }, type: 'object', - required: ['status', 'latest_dag_processor_heartbeat'], + required: ['status', 'latest_dag_processor_heartbeat', 'detailed_status'], title: 'DagProcessorInfoResponse', description: 'DagProcessor info serializer for responses.' } as const; +export const $DagProcessorInstanceInfoResponse = { + properties: { + status: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Status' + }, + hostname: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Hostname' + }, + latest_dag_processor_heartbeat: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Latest Dag Processor Heartbeat' + }, + bundle_names: { + anyOf: [ + { + items: { + type: 'string' + }, + type: 'array' + }, + { + type: 'null' + } + ], + title: 'Bundle Names' + } + }, + type: 'object', + required: ['status', 'hostname', 'latest_dag_processor_heartbeat', 'bundle_names'], + title: 'DagProcessorInstanceInfoResponse', + description: 'Dag processor instance info serializer for responses.' +} as const; + export const $DagRunAssetReference = { properties: { run_id: { @@ -5619,6 +5700,31 @@ export const $JobResponse = { ], 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: [ { @@ -6438,14 +6544,81 @@ export const $SchedulerInfoResponse = { } ], title: 'Latest Scheduler Heartbeat' + }, + detailed_status: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Detailed Status' + }, + instances: { + anyOf: [ + { + items: { + '$ref': '#/components/schemas/SchedulerInstanceInfoResponse' + }, + type: 'array' + }, + { + type: 'null' + } + ], + title: 'Instances' } }, type: 'object', - required: ['status', 'latest_scheduler_heartbeat'], + required: ['status', 'latest_scheduler_heartbeat', 'detailed_status'], title: 'SchedulerInfoResponse', description: 'Scheduler info serializer for responses.' } as const; +export const $SchedulerInstanceInfoResponse = { + properties: { + status: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Status' + }, + hostname: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Hostname' + }, + latest_scheduler_heartbeat: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Latest Scheduler Heartbeat' + } + }, + type: 'object', + required: ['status', 'hostname', 'latest_scheduler_heartbeat'], + title: 'SchedulerInstanceInfoResponse', + description: 'Scheduler instance info serializer for responses.' +} as const; + export const $StructuredLogMessage = { properties: { timestamp: { @@ -8197,14 +8370,92 @@ export const $TriggererInfoResponse = { } ], title: 'Latest Triggerer Heartbeat' + }, + detailed_status: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Detailed Status' + }, + instances: { + anyOf: [ + { + items: { + '$ref': '#/components/schemas/TriggererInstanceInfoResponse' + }, + type: 'array' + }, + { + type: 'null' + } + ], + title: 'Instances' } }, type: 'object', - required: ['status', 'latest_triggerer_heartbeat'], + required: ['status', 'latest_triggerer_heartbeat', 'detailed_status'], title: 'TriggererInfoResponse', description: 'Triggerer info serializer for responses.' } as const; +export const $TriggererInstanceInfoResponse = { + properties: { + status: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Status' + }, + hostname: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Hostname' + }, + latest_triggerer_heartbeat: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Latest Triggerer Heartbeat' + }, + team_name: { + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], + title: 'Team Name' + } + }, + type: 'object', + required: ['status', 'hostname', 'latest_triggerer_heartbeat', 'team_name'], + title: 'TriggererInstanceInfoResponse', + description: 'Triggerer instance info serializer for responses.' +} as const; + export const $UpdateHITLDetailPayload = { properties: { chosen_options: { diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 0d36910280ac8..b0341f512959d 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -1136,6 +1136,18 @@ export type DAGWarningResponse = { export type DagProcessorInfoResponse = { status: string | null; latest_dag_processor_heartbeat: string | null; + detailed_status: string | null; + instances?: Array | null; +}; + +/** + * Dag processor instance info serializer for responses. + */ +export type DagProcessorInstanceInfoResponse = { + status: string | null; + hostname: string | null; + latest_dag_processor_heartbeat: string | null; + bundle_names: Array<(string)> | null; }; /** @@ -1479,6 +1491,8 @@ export type JobResponse = { executor_class: string | null; hostname: string | null; unixname: string | null; + team_name?: string | null; + bundle_names?: Array<(string)> | null; dag_display_name?: string | null; }; @@ -1699,6 +1713,17 @@ export type ReprocessBehavior = 'failed' | 'completed' | 'none'; export type SchedulerInfoResponse = { status: string | null; latest_scheduler_heartbeat: string | null; + detailed_status: string | null; + instances?: Array | null; +}; + +/** + * Scheduler instance info serializer for responses. + */ +export type SchedulerInstanceInfoResponse = { + status: string | null; + hostname: string | null; + latest_scheduler_heartbeat: string | null; }; /** @@ -2040,6 +2065,18 @@ export type TriggerResponse = { export type TriggererInfoResponse = { status: string | null; latest_triggerer_heartbeat: string | null; + detailed_status: string | null; + instances?: Array | null; +}; + +/** + * Triggerer instance info serializer for responses. + */ +export type TriggererInstanceInfoResponse = { + status: string | null; + hostname: string | null; + latest_triggerer_heartbeat: string | null; + team_name: string | null; }; /**