This is an automated email from the ASF dual-hosted git repository.
vincbeck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 9da89e70ac0 Feature expand health endpoint reporting (#70416)
9da89e70ac0 is described below
commit 9da89e70ac036e8d8c23208777f9575cf6eee22d
Author: Jung-Hyun Andrew Kim <[email protected]>
AuthorDate: Fri Aug 28 05:35:05 2026 -0700
Feature expand health endpoint reporting (#70416)
---
.../logging-monitoring/check-health.rst | 100 +++--
airflow-core/docs/migrations-ref.rst | 4 +-
.../src/airflow/api/common/airflow_health.py | 138 ++++--
.../airflow/api_fastapi/core_api/datamodels/job.py | 2 +
.../api_fastapi/core_api/datamodels/monitor.py | 29 ++
.../api_fastapi/core_api/openapi/_private_ui.yaml | 12 +
.../core_api/openapi/v2-rest-api-generated.yaml | 137 ++++++
.../airflow/cli/commands/dag_processor_command.py | 2 +-
.../src/airflow/cli/commands/triggerer_command.py | 4 +-
airflow-core/src/airflow/jobs/job.py | 8 +-
.../src/airflow/jobs/triggerer_job_runner.py | 4 +-
.../0133_3_4_0_add_scope_columns_to_job.py | 58 +++
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 257 ++++++++++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 37 ++
airflow-core/src/airflow/utils/db.py | 2 +-
.../tests/unit/api/common/test_airflow_health.py | 494 +++++++++++++++++----
.../api_fastapi/core_api/routes/public/test_job.py | 21 +
.../core_api/routes/public/test_monitor.py | 63 ++-
.../core_api/routes/public/test_task_instances.py | 2 +
.../cli/commands/test_dag_processor_command.py | 1 +
.../unit/cli/commands/test_triggerer_command.py | 22 +-
airflow-core/tests/unit/jobs/test_triggerer_job.py | 21 +-
.../src/airflowctl/api/datamodels/generated.py | 74 ++-
23 files changed, 1313 insertions(+), 179 deletions(-)
diff --git
a/airflow-core/docs/administration-and-deployment/logging-monitoring/check-health.rst
b/airflow-core/docs/administration-and-deployment/logging-monitoring/check-health.rst
index c9b1966714a..260c0ada751 100644
---
a/airflow-core/docs/administration-and-deployment/logging-monitoring/check-health.rst
+++
b/airflow-core/docs/administration-and-deployment/logging-monitoring/check-health.rst
@@ -36,48 +36,94 @@ Webserver Health Check Endpoint
-------------------------------
To check the health status of your Airflow instance, you can simply access the
endpoint
-``/api/v2/monitor/health``. It will return a JSON object that provides a
high-level glance at the health status across multiple Airflow components.
+``/api/v2/monitor/health``. It will return a JSON object that provides a
high-level glance at the health status across multiple Airflow components,
+including per-instance details when multiple schedulers, triggerers, or Dag
processors are running.
.. code-block:: JSON
{
- "metadatabase":{
- "status":"healthy"
+ "metadatabase": {
+ "status": "healthy"
},
- "scheduler":{
- "status":"healthy",
- "latest_scheduler_heartbeat":"2018-12-26 17:15:11+00:00"
+ "scheduler": {
+ "status": "healthy",
+ "latest_scheduler_heartbeat": "2018-12-26T17:15:11+00:00",
+ "detailed_status": "healthy",
+ "instances": [
+ {
+ "status": "healthy",
+ "hostname": "scheduler-1.example.com",
+ "latest_scheduler_heartbeat": "2018-12-26T17:15:11+00:00"
+ }
+ ]
},
- "triggerer":{
- "status":"healthy",
- "latest_triggerer_heartbeat":"2018-12-26 17:16:12+00:00"
+ "triggerer": {
+ "status": "healthy",
+ "latest_triggerer_heartbeat": "2018-12-26T17:16:12+00:00",
+ "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
+ }
+ ]
},
- "dag_processor":{
- "status":"healthy",
- "latest_dag_processor_heartbeat":"2018-12-26 17:16:12+00:00"
+ "dag_processor": {
+ "status": "healthy",
+ "latest_dag_processor_heartbeat": "2018-12-26T17:16:12+00:00",
+ "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"]
+ }
+ ]
}
}
-* The ``status`` of each component can be either "healthy" or "unhealthy"
+* ``metadatabase``
- * The status of ``metadatabase`` depends on whether a valid connection can
be initiated with the database
+ * ``status`` is ``"healthy"`` when a valid connection can be initiated with
the database, otherwise ``"unhealthy"``.
- * The status of ``scheduler`` depends on when the latest scheduler heartbeat
was received
+* Component-level fields for ``scheduler``, ``triggerer``, and
``dag_processor``
- * If the last heartbeat was received more than 30 seconds (default value)
earlier than the current time, the scheduler is
- considered unhealthy
- * This threshold value can be specified using the option
``scheduler_health_check_threshold`` within the
- ``[scheduler]`` section in ``airflow.cfg``
- * If you run more than one scheduler, only the state of one scheduler will
be reported, i.e. only one working scheduler is enough
- for the scheduler state to be considered healthy
+ * ``status`` (legacy aggregate): ``"healthy"`` if **any** running instance
is alive, otherwise ``"unhealthy"``
+ (including when no running jobs exist for that component).
- * The status of the ``triggerer`` behaves exactly like that of the
``scheduler`` as described above.
- Note that the ``status`` and ``latest_triggerer_heartbeat`` fields in the
health check response will be null for
- deployments that do not include a ``triggerer`` component.
+ * ``detailed_status``: reflects the full set of running instances:
- * The status of the ``dag_processor`` behaves exactly like that of the
``scheduler`` as described above.
- Note that the ``status`` and ``latest_dag_processor_heartbeat`` fields in
the health check response will be null for
- deployments that do not include a ``dag_processor`` component.
+ * ``"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)
+
+ * ``latest_*_heartbeat``: the most recent heartbeat among running jobs of
that type (ordered by heartbeat descending),
+ or ``null`` when there are none.
+ An instance is considered alive when its latest heartbeat is within the
component health-check threshold
+ (defaults and config options: ``[scheduler]
scheduler_health_check_threshold``,
+ ``[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:
+
+ * ``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
+ * ``bundle_names`` (Dag processor only): Dag bundles that processor is
configured to parse, or ``null`` when unset
+
+ * For HA deployments, prefer ``detailed_status`` and ``instances`` when you
need to see every scheduler,
+ triggerer, or Dag processor. The top-level ``status`` remains useful for
simple probes that only care
+ whether at least one instance is healthy.
Please keep in mind that the HTTP response code of ``/api/v2/monitor/health``
endpoint **should not** be used to determine the health
status of the application. The return code is only indicative of the state of
the rest call (200 for success).
diff --git a/airflow-core/docs/migrations-ref.rst
b/airflow-core/docs/migrations-ref.rst
index 8e03f37ef54..e8bef749a6e 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
|
+=========================+==================+===================+==============================================================+
-| ``8d3f1a6b2c47`` (head) | ``c7f0a5d2e9b4`` | ``3.4.0`` | Add
team_name to log. |
+| ``f8c2a1d94e03`` (head) | ``8d3f1a6b2c47`` | ``3.4.0`` | Add
team_name and bundle_names scope columns to job table. |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``8d3f1a6b2c47`` | ``c7f0a5d2e9b4`` | ``3.4.0`` | Add
team_name to log. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| ``c7f0a5d2e9b4`` | ``76c46545c91e`` | ``3.4.0`` | Lower case
team names. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
diff --git a/airflow-core/src/airflow/api/common/airflow_health.py
b/airflow-core/src/airflow/api/common/airflow_health.py
index 1086e22df57..e95a2966b3a 100644
--- a/airflow-core/src/airflow/api/common/airflow_health.py
+++ b/airflow-core/src/airflow/api/common/airflow_health.py
@@ -16,77 +16,159 @@
# 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"
+DOWN = "down"
+
+
+@provide_session
+def get_jobs_health(job_runner_class, *, session: Session = NEW_SESSION) ->
list[Job]:
+ """Return unfinished 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.end_date.is_(None),
+ )
+ .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), 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
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
+ triggerer_status = UNHEALTHY
+ dag_processor_status = UNHEALTHY
+
+ scheduler_detailed_status = DOWN
+ triggerer_detailed_status = DOWN
+ dag_processor_detailed_status = DOWN
try:
- latest_scheduler_job = SchedulerJobRunner.most_recent_job()
+ scheduler_jobs = get_jobs_health(SchedulerJobRunner)
+ scheduler_status = _legacy_status(scheduler_jobs)
+ scheduler_detailed_status = _aggregate_detailed_status(scheduler_jobs)
+ if 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()
-
- 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)
+ 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()
except Exception:
metadatabase_status = UNHEALTHY
+ triggerer_status = UNHEALTHY
+ triggerer_detailed_status = DOWN
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)
+ 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()
except Exception:
metadatabase_status = UNHEALTHY
+ dag_processor_status = UNHEALTHY
+ dag_processor_detailed_status = DOWN
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,
},
}
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 f97be99191a..8e316d7b2e3 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 3bfc5425fd7..5bf60bd83bd 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,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):
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 6a8af48ecf6..25b50d9e825 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
@@ -3864,6 +3864,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 0eb4d2423dc..099e1b9cb5e 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
@@ -13938,12 +13938,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:
@@ -14838,6 +14883,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
@@ -15459,12 +15516,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:
@@ -16590,12 +16684,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/cli/commands/dag_processor_command.py
b/airflow-core/src/airflow/cli/commands/dag_processor_command.py
index 04e410a5d66..7e7df45a8d6 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 7e1d98f44db..87200ff57be 100644
--- a/airflow-core/src/airflow/cli/commands/triggerer_command.py
+++ b/airflow-core/src/airflow/cli/commands/triggerer_command.py
@@ -62,7 +62,9 @@ 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,
)
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 b19cd103de9..fa59a440abc 100644
--- a/airflow-core/src/airflow/jobs/job.py
+++ b/airflow-core/src/airflow/jobs/job.py
@@ -24,7 +24,7 @@ from functools import cached_property, lru_cache
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.log.logging_mixin import LoggingMixin
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 8f8e930b07f..6af0d1d5dfc 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -202,7 +202,6 @@ class TriggererJobRunner(BaseJobRunner, LoggingMixin):
job: Job,
capacity=None,
queues: set[str] | None = None,
- team_name: str | None = None,
):
super().__init__(job)
if capacity is None:
@@ -212,7 +211,6 @@ class TriggererJobRunner(BaseJobRunner, LoggingMixin):
else:
raise ValueError(f"Capacity number {capacity!r} is invalid")
self.queues = queues
- self.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
@@ -274,7 +272,7 @@ class TriggererJobRunner(BaseJobRunner, LoggingMixin):
capacity=self.capacity,
logger=log,
queues=self.queues,
- team_name=self.team_name,
+ team_name=self.job.team_name,
)
# Run the main DB comms loop in this process
self.trigger_runner.run()
diff --git
a/airflow-core/src/airflow/migrations/versions/0133_3_4_0_add_scope_columns_to_job.py
b/airflow-core/src/airflow/migrations/versions/0133_3_4_0_add_scope_columns_to_job.py
new file mode 100644
index 00000000000..ba8fa778610
--- /dev/null
+++
b/airflow-core/src/airflow/migrations/versions/0133_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: 8d3f1a6b2c47
+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 = "8d3f1a6b2c47"
+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")
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 ca7eee96aee..888427fae8e 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
@@ -4411,14 +4411,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: {
@@ -5697,6 +5778,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: [
{
@@ -6605,14 +6711,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: {
@@ -8387,14 +8560,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 1b280d7a567..be90b462bdb 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
@@ -1141,6 +1141,18 @@ export type DAGWarningResponse = {
export type DagProcessorInfoResponse = {
status: string | null;
latest_dag_processor_heartbeat: string | null;
+ detailed_status: string | null;
+ instances?: Array<DagProcessorInstanceInfoResponse> | 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;
};
/**
@@ -1492,6 +1504,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;
};
@@ -1724,6 +1738,17 @@ export type ReprocessBehavior = 'failed' | 'completed' |
'none';
export type SchedulerInfoResponse = {
status: string | null;
latest_scheduler_heartbeat: string | null;
+ detailed_status: string | null;
+ instances?: Array<SchedulerInstanceInfoResponse> | null;
+};
+
+/**
+ * Scheduler instance info serializer for responses.
+ */
+export type SchedulerInstanceInfoResponse = {
+ status: string | null;
+ hostname: string | null;
+ latest_scheduler_heartbeat: string | null;
};
/**
@@ -2070,6 +2095,18 @@ export type TriggerResponse = {
export type TriggererInfoResponse = {
status: string | null;
latest_triggerer_heartbeat: string | null;
+ detailed_status: string | null;
+ instances?: Array<TriggererInstanceInfoResponse> | 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;
};
/**
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 71ec2d2004b..f0779c99690 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -117,7 +117,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.1.8": "509b94a1042d",
"3.2.0": "1d6611b6ab7c",
"3.3.0": "d2f4e1b3c5a7",
- "3.4.0": "8d3f1a6b2c47",
+ "3.4.0": "f8c2a1d94e03",
}
# Prefix used to identify tables holding data moved during migration.
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 43c17e41088..79648ac99ac 100644
--- a/airflow-core/tests/unit/api/common/test_airflow_health.py
+++ b/airflow-core/tests/unit/api/common/test_airflow_health.py
@@ -16,127 +16,471 @@
# under the License.
from __future__ import annotations
-from datetime import datetime
+from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch
import pytest
+from airflow._shared.timezones import timezone
from airflow.api.common.airflow_health import (
+ DEGRADED,
+ DOWN,
HEALTHY,
UNHEALTHY,
get_airflow_health,
+ get_jobs_health,
)
-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 provide_session
+
+from tests_common.test_utils.db import clear_db_jobs
pytestmark = pytest.mark.db_test
+STALE_HEARTBEAT_AGE = timedelta(minutes=5)
-@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,
-):
- 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},
+
+def _mock_job(
+ *,
+ hostname: str,
+ heartbeat: datetime,
+ alive: bool,
+ team_name: str | None = None,
+ bundle_names: list[str] | None = None,
+) -> MagicMock:
+ job = MagicMock(spec=Job)
+ job.hostname = hostname
+ job.latest_heartbeat = heartbeat
+ job.is_alive = MagicMock(return_value=alive)
+ job.team_name = team_name
+ job.bundle_names = bundle_names
+ return job
+
+
+def _empty_component(heartbeat_field: str) -> dict:
+ return {
+ "status": UNHEALTHY,
+ heartbeat_field: None,
+ "detailed_status": DOWN,
+ "instances": None,
}
- assert health_status == expected_status
+
+def _create_job(
+ session,
+ runner_class,
+ *,
+ hostname: str,
+ heartbeat: datetime,
+ state: JobState = JobState.RUNNING,
+ end_date: datetime | None = None,
+ team_name: str | None = None,
+ bundle_names: list[str] | None = None,
+) -> Job:
+ job = Job(
+ state=state,
+ latest_heartbeat=heartbeat,
+ hostname=hostname,
+ end_date=end_date,
+ team_name=team_name,
+ bundle_names=bundle_names,
+ )
+ if runner_class is TriggererJobRunner:
+ runner_class(job=job, capacity=1)
+ else:
+ runner_class(job=job)
+ session.add(job)
+ return job
+
+
+ALIVE_SCHEDULER_JOB_MOCK = _mock_job(hostname="scheduler-alive",
heartbeat=datetime(2024, 2, 1), alive=True)
+STALE_SCHEDULER_JOB_MOCK = _mock_job(hostname="scheduler-stale",
heartbeat=datetime(2024, 1, 1), alive=False)
+STALE_SCHEDULER_JOB_MOCK_2 = _mock_job(
+ hostname="scheduler-stale-2", heartbeat=datetime(2023, 12, 1), alive=False
+)
+STALE_SCHEDULER_JOB_MOCK_3 = _mock_job(
+ hostname="scheduler-stale-3", heartbeat=datetime(2023, 11, 1), alive=False
+)
+
+ALIVE_TRIGGERER_JOB_MOCK = _mock_job(
+ hostname="triggerer-alive",
+ heartbeat=datetime(2024, 2, 1),
+ alive=True,
+ team_name="team-a",
+)
+STALE_TRIGGERER_JOB_MOCK = _mock_job(
+ hostname="triggerer-stale",
+ heartbeat=datetime(2024, 1, 1),
+ alive=False,
+ team_name="team-b",
+)
+
+ALIVE_DAG_PROCESSOR_JOB_MOCK = _mock_job(
+ hostname="dag-processor-host",
+ heartbeat=datetime(2024, 1, 3),
+ alive=True,
+ bundle_names=["bundle-a"],
+)
-@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")
+def test_get_airflow_health_no_jobs(mock_get_jobs_health):
+ mock_get_jobs_health.side_effect = [[], [], []]
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},
+ assert health_status == {
+ "metadatabase": {"status": HEALTHY},
+ "scheduler": _empty_component("latest_scheduler_heartbeat"),
+ "triggerer": _empty_component("latest_triggerer_heartbeat"),
+ "dag_processor": _empty_component("latest_dag_processor_heartbeat"),
}
- assert health_status == expected_status
+@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()
-LATEST_SCHEDULER_JOB_MOCK = MagicMock(spec=Job)
-LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime.now()
-LATEST_SCHEDULER_JOB_MOCK.is_alive = MagicMock(return_value=True)
+ assert health_status == {
+ "metadatabase": {"status": UNHEALTHY},
+ "scheduler": _empty_component("latest_scheduler_heartbeat"),
+ "triggerer": _empty_component("latest_triggerer_heartbeat"),
+ "dag_processor": _empty_component("latest_dag_processor_heartbeat"),
+ }
-@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_one_alive_job(mock_get_jobs_health):
+ mock_get_jobs_health.side_effect = [[ALIVE_SCHEDULER_JOB_MOCK], [], []]
health_status = get_airflow_health()
- expected_status = {
+ assert health_status == {
"metadatabase": {"status": HEALTHY},
"scheduler": {
"status": HEALTHY,
- "latest_scheduler_heartbeat":
LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(),
+ "latest_scheduler_heartbeat":
ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(),
+ "detailed_status": HEALTHY,
+ "instances": [
+ {
+ "hostname": ALIVE_SCHEDULER_JOB_MOCK.hostname,
+ "status": HEALTHY,
+ "latest_scheduler_heartbeat":
ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(),
+ }
+ ],
},
- "triggerer": {"status": None, "latest_triggerer_heartbeat": None},
- "dag_processor": {"status": None, "latest_dag_processor_heartbeat":
None},
+ "triggerer": _empty_component("latest_triggerer_heartbeat"),
+ "dag_processor": _empty_component("latest_dag_processor_heartbeat"),
}
- assert health_status == expected_status
+@patch("airflow.api.common.airflow_health.get_jobs_health")
+def test_get_airflow_health_mixed_alive_and_stale_jobs(mock_get_jobs_health):
+ mock_get_jobs_health.side_effect = [
+ [ALIVE_SCHEDULER_JOB_MOCK, STALE_SCHEDULER_JOB_MOCK,
STALE_SCHEDULER_JOB_MOCK_2],
+ [],
+ [],
+ ]
+ health_status = get_airflow_health()
-LATEST_TRIGGERER_JOB_MOCK = MagicMock(spec=Job)
-LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat = datetime.now()
-LATEST_TRIGGERER_JOB_MOCK.is_alive = MagicMock(return_value=True)
+ assert health_status["scheduler"]["status"] == HEALTHY
+ assert health_status["scheduler"]["detailed_status"] == DEGRADED
+ assert health_status["scheduler"]["instances"] == [
+ {
+ "hostname": ALIVE_SCHEDULER_JOB_MOCK.hostname,
+ "status": HEALTHY,
+ "latest_scheduler_heartbeat":
ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(),
+ },
+ {
+ "hostname": STALE_SCHEDULER_JOB_MOCK.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat":
STALE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(),
+ },
+ {
+ "hostname": STALE_SCHEDULER_JOB_MOCK_2.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat":
STALE_SCHEDULER_JOB_MOCK_2.latest_heartbeat.isoformat(),
+ },
+ ]
+ assert (
+ health_status["scheduler"]["latest_scheduler_heartbeat"]
+ == ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat()
+ )
+ assert health_status["triggerer"] ==
_empty_component("latest_triggerer_heartbeat")
+ assert health_status["dag_processor"] ==
_empty_component("latest_dag_processor_heartbeat")
-LATEST_DAG_PROCESSOR_JOB_MOCK = MagicMock(spec=Job)
-LATEST_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat = datetime.now()
-LATEST_DAG_PROCESSOR_JOB_MOCK.is_alive = MagicMock(return_value=True)
+@patch("airflow.api.common.airflow_health.get_jobs_health")
+def test_get_airflow_health_all_stale_jobs(mock_get_jobs_health):
+ mock_get_jobs_health.side_effect = [
+ [STALE_SCHEDULER_JOB_MOCK, STALE_SCHEDULER_JOB_MOCK_2,
STALE_SCHEDULER_JOB_MOCK_3],
+ [],
+ [],
+ ]
+ health_status = get_airflow_health()
-@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,
-):
+ assert health_status["scheduler"]["status"] == UNHEALTHY
+ assert health_status["scheduler"]["detailed_status"] == DOWN
+ assert health_status["scheduler"]["instances"] == [
+ {
+ "hostname": STALE_SCHEDULER_JOB_MOCK.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat":
STALE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(),
+ },
+ {
+ "hostname": STALE_SCHEDULER_JOB_MOCK_2.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat":
STALE_SCHEDULER_JOB_MOCK_2.latest_heartbeat.isoformat(),
+ },
+ {
+ "hostname": STALE_SCHEDULER_JOB_MOCK_3.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat":
STALE_SCHEDULER_JOB_MOCK_3.latest_heartbeat.isoformat(),
+ },
+ ]
+
+
+@patch("airflow.api.common.airflow_health.get_jobs_health")
+def
test_get_airflow_health_mixed_triggerers_include_team_name(mock_get_jobs_health):
+ mock_get_jobs_health.side_effect = [[], [ALIVE_TRIGGERER_JOB_MOCK,
STALE_TRIGGERER_JOB_MOCK], []]
health_status = get_airflow_health()
- expected_status = {
+ assert health_status["triggerer"]["status"] == HEALTHY
+ assert health_status["triggerer"]["detailed_status"] == DEGRADED
+ assert health_status["triggerer"]["instances"] == [
+ {
+ "hostname": ALIVE_TRIGGERER_JOB_MOCK.hostname,
+ "status": HEALTHY,
+ "latest_triggerer_heartbeat":
ALIVE_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(),
+ "team_name": ALIVE_TRIGGERER_JOB_MOCK.team_name,
+ },
+ {
+ "hostname": STALE_TRIGGERER_JOB_MOCK.hostname,
+ "status": UNHEALTHY,
+ "latest_triggerer_heartbeat":
STALE_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(),
+ "team_name": STALE_TRIGGERER_JOB_MOCK.team_name,
+ },
+ ]
+ assert (
+ health_status["triggerer"]["latest_triggerer_heartbeat"]
+ == ALIVE_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat()
+ )
+
+
+@patch("airflow.api.common.airflow_health.get_jobs_health")
+def
test_get_airflow_health_triggerer_and_dag_processor_healthy(mock_get_jobs_health):
+ mock_get_jobs_health.side_effect = [[], [ALIVE_TRIGGERER_JOB_MOCK],
[ALIVE_DAG_PROCESSOR_JOB_MOCK]]
+ health_status = get_airflow_health()
+
+ assert health_status == {
"metadatabase": {"status": HEALTHY},
- "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None},
+ "scheduler": _empty_component("latest_scheduler_heartbeat"),
"triggerer": {
"status": HEALTHY,
- "latest_triggerer_heartbeat":
LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(),
+ "latest_triggerer_heartbeat":
ALIVE_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(),
+ "detailed_status": HEALTHY,
+ "instances": [
+ {
+ "hostname": ALIVE_TRIGGERER_JOB_MOCK.hostname,
+ "status": HEALTHY,
+ "latest_triggerer_heartbeat":
ALIVE_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(),
+ "team_name": ALIVE_TRIGGERER_JOB_MOCK.team_name,
+ }
+ ],
},
"dag_processor": {
"status": HEALTHY,
- "latest_dag_processor_heartbeat":
LATEST_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat.isoformat(),
+ "latest_dag_processor_heartbeat":
ALIVE_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat.isoformat(),
+ "detailed_status": HEALTHY,
+ "instances": [
+ {
+ "hostname": ALIVE_DAG_PROCESSOR_JOB_MOCK.hostname,
+ "status": HEALTHY,
+ "latest_dag_processor_heartbeat":
ALIVE_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat.isoformat(),
+ "bundle_names": ALIVE_DAG_PROCESSOR_JOB_MOCK.bundle_names,
+ }
+ ],
},
}
- assert health_status == expected_status
+
+class TestAirflowHealthFromDb:
+ @pytest.fixture(autouse=True)
+ def cleanup_jobs(self):
+ clear_db_jobs()
+ yield
+ clear_db_jobs()
+
+ @provide_session
+ def
test_get_jobs_health_returns_unfinished_jobs_ordered_by_heartbeat(self, *,
session):
+ older = timezone.utcnow() - timedelta(minutes=5)
+ newer = timezone.utcnow()
+ older_job = _create_job(session, SchedulerJobRunner, hostname="older",
heartbeat=older)
+ newer_job = _create_job(session, SchedulerJobRunner, hostname="newer",
heartbeat=newer)
+ _create_job(
+ session,
+ SchedulerJobRunner,
+ hostname="ended",
+ heartbeat=newer,
+ end_date=newer,
+ )
+ unfinished_failed = _create_job(
+ session,
+ SchedulerJobRunner,
+ hostname="failed-unfinished",
+ heartbeat=older - timedelta(minutes=1),
+ state=JobState.FAILED,
+ )
+ _create_job(session, TriggererJobRunner, hostname="triggerer",
heartbeat=newer)
+ session.commit()
+
+ jobs = get_jobs_health(SchedulerJobRunner, session=session)
+
+ assert [job.hostname for job in jobs] == [
+ newer_job.hostname,
+ older_job.hostname,
+ unfinished_failed.hostname,
+ ]
+
+ def test_get_airflow_health_no_jobs(self):
+ health_status = get_airflow_health()
+
+ assert health_status == {
+ "metadatabase": {"status": HEALTHY},
+ "scheduler": _empty_component("latest_scheduler_heartbeat"),
+ "triggerer": _empty_component("latest_triggerer_heartbeat"),
+ "dag_processor":
_empty_component("latest_dag_processor_heartbeat"),
+ }
+
+ @provide_session
+ def test_get_airflow_health_one_alive_job(self, *, session):
+ heartbeat = timezone.utcnow()
+ job = _create_job(session, SchedulerJobRunner,
hostname="scheduler-alive", heartbeat=heartbeat)
+ _create_job(
+ session,
+ SchedulerJobRunner,
+ hostname="scheduler-ended",
+ heartbeat=heartbeat,
+ end_date=heartbeat,
+ )
+ session.commit()
+
+ health_status = get_airflow_health()
+
+ assert health_status["metadatabase"]["status"] == HEALTHY
+ assert health_status["scheduler"]["status"] == HEALTHY
+ assert health_status["scheduler"]["detailed_status"] == HEALTHY
+ assert health_status["scheduler"]["instances"] == [
+ {
+ "hostname": job.hostname,
+ "status": HEALTHY,
+ "latest_scheduler_heartbeat": heartbeat.isoformat(),
+ }
+ ]
+ assert health_status["scheduler"]["latest_scheduler_heartbeat"] ==
heartbeat.isoformat()
+ assert health_status["triggerer"] ==
_empty_component("latest_triggerer_heartbeat")
+ assert health_status["dag_processor"] ==
_empty_component("latest_dag_processor_heartbeat")
+
+ @provide_session
+ def test_get_airflow_health_mixed_alive_and_stale_jobs(self, *, session):
+ alive_heartbeat = timezone.utcnow()
+ stale_heartbeat = timezone.utcnow() - STALE_HEARTBEAT_AGE
+ older_stale_heartbeat = stale_heartbeat - timedelta(minutes=1)
+ alive = _create_job(
+ session, SchedulerJobRunner, hostname="scheduler-alive",
heartbeat=alive_heartbeat
+ )
+ stale = _create_job(
+ session, SchedulerJobRunner, hostname="scheduler-stale",
heartbeat=stale_heartbeat
+ )
+ older_stale = _create_job(
+ session, SchedulerJobRunner, hostname="scheduler-stale-2",
heartbeat=older_stale_heartbeat
+ )
+ session.commit()
+
+ health_status = get_airflow_health()
+
+ assert health_status["scheduler"]["status"] == HEALTHY
+ assert health_status["scheduler"]["detailed_status"] == DEGRADED
+ assert health_status["scheduler"]["instances"] == [
+ {
+ "hostname": alive.hostname,
+ "status": HEALTHY,
+ "latest_scheduler_heartbeat": alive_heartbeat.isoformat(),
+ },
+ {
+ "hostname": stale.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat": stale_heartbeat.isoformat(),
+ },
+ {
+ "hostname": older_stale.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat":
older_stale_heartbeat.isoformat(),
+ },
+ ]
+ assert health_status["scheduler"]["latest_scheduler_heartbeat"] ==
alive_heartbeat.isoformat()
+
+ @provide_session
+ def test_get_airflow_health_all_stale_jobs(self, *, session):
+ first = timezone.utcnow() - STALE_HEARTBEAT_AGE
+ second = first - timedelta(minutes=1)
+ third = second - timedelta(minutes=1)
+ jobs = [
+ _create_job(session, SchedulerJobRunner, hostname="stale-1",
heartbeat=first),
+ _create_job(session, SchedulerJobRunner, hostname="stale-2",
heartbeat=second),
+ _create_job(session, SchedulerJobRunner, hostname="stale-3",
heartbeat=third),
+ ]
+ session.commit()
+
+ health_status = get_airflow_health()
+
+ assert health_status["scheduler"]["status"] == UNHEALTHY
+ assert health_status["scheduler"]["detailed_status"] == DOWN
+ assert health_status["scheduler"]["instances"] == [
+ {
+ "hostname": job.hostname,
+ "status": UNHEALTHY,
+ "latest_scheduler_heartbeat": job.latest_heartbeat.isoformat(),
+ }
+ for job in jobs
+ ]
+
+ @provide_session
+ def test_get_airflow_health_mixed_triggerers_include_team_name(self,
testing_team, *, session):
+ alive_heartbeat = timezone.utcnow()
+ stale_heartbeat = timezone.utcnow() - STALE_HEARTBEAT_AGE
+ alive = _create_job(
+ session,
+ TriggererJobRunner,
+ hostname="triggerer-alive",
+ heartbeat=alive_heartbeat,
+ team_name=testing_team.name,
+ )
+ stale = _create_job(
+ session,
+ TriggererJobRunner,
+ hostname="triggerer-stale",
+ heartbeat=stale_heartbeat,
+ team_name=testing_team.name,
+ )
+ session.commit()
+
+ health_status = get_airflow_health()
+
+ assert health_status["triggerer"]["status"] == HEALTHY
+ assert health_status["triggerer"]["detailed_status"] == DEGRADED
+ assert health_status["triggerer"]["instances"] == [
+ {
+ "hostname": alive.hostname,
+ "status": HEALTHY,
+ "latest_triggerer_heartbeat": alive_heartbeat.isoformat(),
+ "team_name": testing_team.name,
+ },
+ {
+ "hostname": stale.hostname,
+ "status": UNHEALTHY,
+ "latest_triggerer_heartbeat": stale_heartbeat.isoformat(),
+ "team_name": testing_team.name,
+ },
+ ]
+ assert health_status["triggerer"]["latest_triggerer_heartbeat"] ==
alive_heartbeat.isoformat()
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 8e9538d3f63..cb4c7254492 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 @@ class TestGetJobs(TestJobEndpoint):
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 @@ class TestGetJobs(TestJobEndpoint):
"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
@@ -187,6 +189,25 @@ class TestGetJobs(TestJobEndpoint):
assert response_json["total_entries"] == 1
assert response_json["jobs"][0]["dag_id"] == "target_dag"
+ def test_get_jobs_includes_team_name_and_bundle_names(self, test_client,
session: Session, testing_team):
+ clear_db_jobs()
+ job = Job(
+ state=JobState.RUNNING,
+ job_type="TriggererJob",
+ team_name=testing_team.name,
+ bundle_names=["bundle-a", "bundle-b"],
+ )
+ session.add(job)
+ session.commit()
+
+ response = test_client.get("/jobs")
+
+ assert response.status_code == 200
+ response_json = response.json()
+ assert response_json["total_entries"] == 1
+ assert response_json["jobs"][0]["team_name"] == testing_team.name
+ assert response_json["jobs"][0]["bundle_names"] == ["bundle-a",
"bundle-b"]
+
def test_should_raises_401_unauthenticated(self,
unauthenticated_test_client):
response = unauthenticated_test_client.get("/jobs")
assert response.status_code == 401
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 d4b3d97f932..e696ec43811 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 @@ class TestGetHealth(TestMonitorEndpoint):
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 @@ class TestGetHealth(TestMonitorEndpoint):
"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,
},
}
@@ -133,6 +136,58 @@ class TestGetHealth(TestMonitorEndpoint):
assert body["scheduler"]["status"] == HEALTHY
assert body["triggerer"]["status"] == HEALTHY
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.monitor.get_airflow_health")
+ def test_health_includes_instance_team_name_and_bundle_names(self,
mock_get_airflow_health, test_client):
+ mock_get_airflow_health.return_value = {
+ "metadatabase": {"status": HEALTHY},
+ "scheduler": {
+ "status": HEALTHY,
+ "latest_scheduler_heartbeat":
"2024-11-23T11:09:16.663124+00:00",
+ "detailed_status": HEALTHY,
+ "instances": [
+ {
+ "status": HEALTHY,
+ "hostname": "scheduler-1",
+ "latest_scheduler_heartbeat":
"2024-11-23T11:09:16.663124+00:00",
+ }
+ ],
+ },
+ "triggerer": {
+ "status": HEALTHY,
+ "latest_triggerer_heartbeat":
"2024-11-23T11:09:15.815483+00:00",
+ "detailed_status": HEALTHY,
+ "instances": [
+ {
+ "status": HEALTHY,
+ "hostname": "triggerer-1",
+ "latest_triggerer_heartbeat":
"2024-11-23T11:09:15.815483+00:00",
+ "team_name": "team-a",
+ }
+ ],
+ },
+ "dag_processor": {
+ "status": HEALTHY,
+ "latest_dag_processor_heartbeat":
"2024-11-23T11:09:15.815483+00:00",
+ "detailed_status": HEALTHY,
+ "instances": [
+ {
+ "status": HEALTHY,
+ "hostname": "dag-processor-1",
+ "latest_dag_processor_heartbeat":
"2024-11-23T11:09:15.815483+00:00",
+ "bundle_names": ["bundle-a"],
+ }
+ ],
+ },
+ }
+
+ response = test_client.get("/monitor/health")
+
+ assert response.status_code == 200
+ body = response.json()
+ assert body["triggerer"]["instances"][0]["team_name"] == "team-a"
+ assert body["dag_processor"]["instances"][0]["bundle_names"] ==
["bundle-a"]
+ assert body["scheduler"]["instances"][0]["hostname"] == "scheduler-1"
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.monitor.get_airflow_health")
def test_health_without_dag_processor(self, mock_get_airflow_health,
test_client):
mock_get_airflow_health.return_value = {
@@ -140,10 +195,12 @@ class TestGetHealth(TestMonitorEndpoint):
"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,
},
}
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 7c945edf954..3784a57f062 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
@@ -456,11 +456,13 @@ class TestGetTaskInstance(TestTaskInstanceEndpoint):
"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,
diff --git a/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
b/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
index abdf52ad125..6b6e429b600 100644
--- a/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_dag_processor_command.py
@@ -56,6 +56,7 @@ class TestDagProcessorCommand:
with configure_testing_dag_bundle(os.devnull):
dag_processor_command.dag_processor(args)
assert mock_runner.call_args.kwargs["processor"].bundle_names_to_parse
== ["testing"]
+ assert mock_runner.call_args.kwargs["job"].bundle_names == ["testing"]
@conf_vars({("core", "load_examples"): "False"})
@mock.patch("airflow.cli.commands.dag_processor_command.DagProcessorJobRunner")
diff --git a/airflow-core/tests/unit/cli/commands/test_triggerer_command.py
b/airflow-core/tests/unit/cli/commands/test_triggerer_command.py
index 0cc9a10b52a..f88dc90ea5c 100644
--- a/airflow-core/tests/unit/cli/commands/test_triggerer_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_triggerer_command.py
@@ -51,9 +51,8 @@ class TestTriggererCommand:
triggerer_command.triggerer(args)
mock_serve.return_value.__enter__.assert_called_once()
mock_serve.return_value.__exit__.assert_called_once()
- mock_triggerer_job_runner.assert_called_once_with(
- job=mock.ANY, capacity=42, queues=None, team_name=None
- )
+ mock_triggerer_job_runner.assert_called_once_with(job=mock.ANY,
capacity=42, queues=None)
+ assert mock_triggerer_job_runner.call_args.kwargs["job"].team_name is
None
@conf_vars({("triggerer", "queues_enabled"): "True"})
@mock.patch("airflow.cli.commands.triggerer_command.TriggererJobRunner")
@@ -66,8 +65,9 @@ class TestTriggererCommand:
mock_serve.return_value.__enter__.assert_called_once()
mock_serve.return_value.__exit__.assert_called_once()
mock_triggerer_job_runner.assert_called_once_with(
- job=mock.ANY, capacity=4, queues=set(["my_queue", "other_queue"]),
team_name=None
+ job=mock.ANY, capacity=4, queues=set(["my_queue", "other_queue"])
)
+ assert mock_triggerer_job_runner.call_args.kwargs["job"].team_name is
None
@mock.patch("airflow.cli.commands.triggerer_command.TriggererJobRunner")
@mock.patch("airflow.cli.commands.triggerer_command.run_job")
@@ -120,22 +120,20 @@ class TestTriggererCommand:
@mock.patch("airflow.cli.commands.triggerer_command.TriggererJobRunner")
@mock.patch("airflow.cli.commands.triggerer_command._serve_logs")
def test_team_name_passed_through(self, mock_serve,
mock_triggerer_job_runner, mock_get_team):
- """--team-name should be passed to TriggererJobRunner when valid"""
+ """--team-name should be set on the Job when valid"""
mock_triggerer_job_runner.return_value.job_type = "TriggererJob"
args = self.parser.parse_args(["triggerer", "--team-name", "team_a"])
triggerer_command.triggerer(args)
- mock_triggerer_job_runner.assert_called_once_with(
- job=mock.ANY, capacity=mock.ANY, queues=None, team_name="team_a"
- )
+ mock_triggerer_job_runner.assert_called_once_with(job=mock.ANY,
capacity=mock.ANY, queues=None)
+ assert mock_triggerer_job_runner.call_args.kwargs["job"].team_name ==
"team_a"
@conf_vars({("core", "multi_team"): "False"})
@mock.patch("airflow.cli.commands.triggerer_command.TriggererJobRunner")
@mock.patch("airflow.cli.commands.triggerer_command._serve_logs")
def test_no_team_name_passes_none(self, mock_serve,
mock_triggerer_job_runner):
- """Without --team-name, team_name=None is passed"""
+ """Without --team-name, Job.team_name is None"""
mock_triggerer_job_runner.return_value.job_type = "TriggererJob"
args = self.parser.parse_args(["triggerer"])
triggerer_command.triggerer(args)
- mock_triggerer_job_runner.assert_called_once_with(
- job=mock.ANY, capacity=mock.ANY, queues=None, team_name=None
- )
+ mock_triggerer_job_runner.assert_called_once_with(job=mock.ANY,
capacity=mock.ANY, queues=None)
+ assert mock_triggerer_job_runner.call_args.kwargs["job"].team_name is
None
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index 07f861f4f91..5c4c247d2a0 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -265,11 +265,22 @@ def test_capacity_decode():
@pytest.mark.parametrize("team_name", ["team_a", None])
-def test_triggerer_job_runner_stores_team_name(team_name):
- """TriggererJobRunner stores team_name as-is (validated at CLI layer)."""
- job = Job()
- runner = TriggererJobRunner(job, capacity=10, team_name=team_name)
- assert runner.team_name == team_name
[email protected](TriggerRunnerSupervisor, "start")
+def test_execute_passes_job_team_name_to_supervisor(mock_supervisor_start,
team_name):
+ mock_supervisor = MagicMock(spec=TriggerRunnerSupervisor)
+ mock_supervisor._exit_code = 0
+ mock_supervisor_start.return_value = mock_supervisor
+
+ job = Job(team_name=team_name)
+ job_runner = TriggererJobRunner(job)
+ with (
+ patch.object(job_runner, "register_signals"),
+ patch("airflow.jobs.triggerer_job_runner.stats.initialize"),
+ ):
+ job_runner._execute()
+
+ mock_supervisor_start.assert_called_once()
+ assert mock_supervisor_start.call_args.kwargs["team_name"] == team_name
@pytest.mark.parametrize("platform_uses_exec", [True, False])
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index b5ef5665ac0..f2e0546f264 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -507,13 +507,15 @@ 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")]
+ hostname: Annotated[str | None, Field(title="Hostname")]
latest_dag_processor_heartbeat: Annotated[str | None, Field(title="Latest
Dag Processor Heartbeat")]
+ bundle_names: Annotated[list[str] | None, Field(title="Bundle Names")]
class DagRunAssetReference(BaseModel):
@@ -785,6 +787,8 @@ class JobResponse(BaseModel):
executor_class: Annotated[str | None, Field(title="Executor Class")]
hostname: Annotated[str | None, Field(title="Hostname")]
unixname: Annotated[str | None, Field(title="Unixname")]
+ 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
@@ -975,12 +979,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")]
+ hostname: Annotated[str | None, Field(title="Hostname")]
latest_scheduler_heartbeat: Annotated[str | None, Field(title="Latest
Scheduler Heartbeat")]
@@ -1188,13 +1193,15 @@ class TriggerResponse(BaseModel):
triggerer_id: Annotated[int | None, Field(title="Triggerer Id")]
-class TriggererInfoResponse(BaseModel):
+class TriggererInstanceInfoResponse(BaseModel):
"""
- Triggerer info serializer for responses.
+ Triggerer instance info serializer for responses.
"""
status: Annotated[str | None, Field(title="Status")]
+ hostname: Annotated[str | None, Field(title="Hostname")]
latest_triggerer_heartbeat: Annotated[str | None, Field(title="Latest
Triggerer Heartbeat")]
+ team_name: Annotated[str | None, Field(title="Team Name")]
class UpdateHITLDetailPayload(BaseModel):
@@ -1923,6 +1930,17 @@ 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")]
+ latest_dag_processor_heartbeat: Annotated[str | None, Field(title="Latest
Dag Processor Heartbeat")]
+ detailed_status: Annotated[str | None, Field(title="Detailed Status")]
+ instances: Annotated[list[DagProcessorInstanceInfoResponse] | None,
Field(title="Instances")] = None
+
+
class DagStatsResponse(BaseModel):
"""
Dag Stats serializer for responses.
@@ -1985,17 +2003,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.
@@ -2098,6 +2105,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")]
+ latest_scheduler_heartbeat: Annotated[str | None, Field(title="Latest
Scheduler Heartbeat")]
+ detailed_status: Annotated[str | None, Field(title="Detailed Status")]
+ instances: Annotated[list[SchedulerInstanceInfoResponse] | None,
Field(title="Instances")] = None
+
+
class TaskDependencyCollectionResponse(BaseModel):
"""
Task scheduling dependencies collection serializer for responses.
@@ -2226,6 +2244,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")]
+ latest_triggerer_heartbeat: Annotated[str | None, Field(title="Latest
Triggerer Heartbeat")]
+ detailed_status: Annotated[str | None, Field(title="Detailed Status")]
+ instances: Annotated[list[TriggererInstanceInfoResponse] | None,
Field(title="Instances")] = None
+
+
class VariableCollectionResponse(BaseModel):
"""
Variable Collection serializer for responses.
@@ -2491,6 +2520,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.
