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 c9b1966714af4..091b1399e303b 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"``. + When no running ``triggerer`` or ``dag_processor`` jobs exist, ``status`` is ``null`` 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 + * ``"unhealthy"`` — no running instance is alive + * ``null`` — no running ``triggerer`` or ``dag_processor`` jobs exist for that component + + * ``latest_*_heartbeat``: the most recent heartbeat among running jobs of that type (ordered by heartbeat descending). + 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 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/common/airflow_health.py b/airflow-core/src/airflow/api/common/airflow_health.py index 1086e22df576a..ed0c8d5b907f0 100644 --- a/airflow-core/src/airflow/api/common/airflow_health.py +++ b/airflow-core/src/airflow/api/common/airflow_health.py @@ -16,77 +16,164 @@ # under the License. from __future__ import annotations -from typing import Any +from typing import TYPE_CHECKING, Any + +from sqlalchemy import select from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner +from airflow.jobs.job import Job, JobState from airflow.jobs.scheduler_job_runner import SchedulerJobRunner from airflow.jobs.triggerer_job_runner import TriggererJobRunner +from airflow.utils.session import NEW_SESSION, provide_session + +if TYPE_CHECKING: + from sqlalchemy.orm import Session HEALTHY = "healthy" UNHEALTHY = "unhealthy" +DEGRADED = "degraded" + + +@provide_session +def get_jobs_health(job_runner_class, *, session: Session = NEW_SESSION) -> list[Job]: + """Return all running jobs for the runner class, ordered by latest heartbeat.""" + return list( + session.scalars( + select(Job) + .where( + Job.job_type == job_runner_class.job_type, + Job.state == JobState.RUNNING, + ) + .order_by(Job.latest_heartbeat.desc()) + ) + ) + + +def _job_instance_health(job: Job, heartbeat_field_name: str) -> dict[str, Any]: + heartbeat = job.latest_heartbeat.isoformat() if job.latest_heartbeat else None + return { + "hostname": job.hostname, + "status": HEALTHY if job.is_alive() else UNHEALTHY, + heartbeat_field_name: heartbeat, + } + + +def _legacy_status(jobs: list[Job]) -> str: + """Top-level status: healthy if any instance is alive.""" + return HEALTHY if any(job.is_alive() for job in jobs) else UNHEALTHY + + +def _triggerer_instance_health(job: Job) -> dict[str, Any]: + return { + **_job_instance_health(job, "latest_triggerer_heartbeat"), + "team_name": job.team_name, + } + + +def _dag_processor_instance_health(job: Job) -> dict[str, Any]: + return { + **_job_instance_health(job, "latest_dag_processor_heartbeat"), + "bundle_names": job.bundle_names, + } + + +def _aggregate_detailed_status(jobs: list[Job]) -> str: + """detailed_status: healthy (all alive), degraded (some alive), unhealthy (none alive).""" + alive_count = sum(1 for job in jobs if job.is_alive()) + if alive_count == 0: + return UNHEALTHY + if alive_count == len(jobs): + return HEALTHY + return DEGRADED def get_airflow_health() -> dict[str, Any]: - """Get the health for Airflow metadatabase, scheduler and triggerer.""" + """Get the health for Airflow metadatabase, scheduler, triggerer, and dag processor.""" metadatabase_status = HEALTHY + latest_scheduler_heartbeat = None latest_triggerer_heartbeat = None latest_dag_processor_heartbeat = None + scheduler_instances: list[dict[str, Any]] | None = None + triggerer_instances: list[dict[str, Any]] | None = None + dag_processor_instances: list[dict[str, Any]] | None = None + scheduler_status = UNHEALTHY triggerer_status: str | None = UNHEALTHY dag_processor_status: str | None = UNHEALTHY + scheduler_detailed_status = UNHEALTHY + triggerer_detailed_status: str | None = UNHEALTHY + dag_processor_detailed_status: str | None = UNHEALTHY + try: - latest_scheduler_job = SchedulerJobRunner.most_recent_job() + scheduler_jobs = get_jobs_health(SchedulerJobRunner) + if scheduler_jobs: + scheduler_status = _legacy_status(scheduler_jobs) # The top sorted job is confirmed alive + scheduler_detailed_status = _aggregate_detailed_status(scheduler_jobs) + scheduler_instances = [ + _job_instance_health(job, "latest_scheduler_heartbeat") for job in scheduler_jobs + ] - if latest_scheduler_job: - if latest_scheduler_job.latest_heartbeat: - latest_scheduler_heartbeat = latest_scheduler_job.latest_heartbeat.isoformat() - if latest_scheduler_job.is_alive(): - scheduler_status = HEALTHY + if scheduler_jobs[0].latest_heartbeat: + latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat() except Exception: metadatabase_status = UNHEALTHY try: - latest_triggerer_job = TriggererJobRunner.most_recent_job() + triggerer_jobs = get_jobs_health(TriggererJobRunner) + if triggerer_jobs: + triggerer_status = _legacy_status(triggerer_jobs) + triggerer_detailed_status = _aggregate_detailed_status(triggerer_jobs) + triggerer_instances = [_triggerer_instance_health(job) for job in triggerer_jobs] - if latest_triggerer_job: - if latest_triggerer_job.latest_heartbeat: - latest_triggerer_heartbeat = latest_triggerer_job.latest_heartbeat.isoformat() - if latest_triggerer_job.is_alive(): - triggerer_status = HEALTHY + if triggerer_jobs[0].latest_heartbeat: + latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat() else: triggerer_status = None + triggerer_detailed_status = None except Exception: metadatabase_status = UNHEALTHY + triggerer_status = UNHEALTHY + triggerer_detailed_status = UNHEALTHY try: - latest_dag_processor_job = DagProcessorJobRunner.most_recent_job() + dag_processor_jobs = get_jobs_health(DagProcessorJobRunner) + if dag_processor_jobs: + dag_processor_status = _legacy_status(dag_processor_jobs) + dag_processor_detailed_status = _aggregate_detailed_status(dag_processor_jobs) + dag_processor_instances = [_dag_processor_instance_health(job) for job in dag_processor_jobs] - if latest_dag_processor_job: - if latest_dag_processor_job.latest_heartbeat: - latest_dag_processor_heartbeat = latest_dag_processor_job.latest_heartbeat.isoformat() - if latest_dag_processor_job.is_alive(): - dag_processor_status = HEALTHY + if dag_processor_jobs[0].latest_heartbeat: + latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat() else: dag_processor_status = None + dag_processor_detailed_status = None except Exception: metadatabase_status = UNHEALTHY + dag_processor_status = UNHEALTHY + dag_processor_detailed_status = UNHEALTHY airflow_health_status = { "metadatabase": {"status": metadatabase_status}, "scheduler": { "status": scheduler_status, "latest_scheduler_heartbeat": latest_scheduler_heartbeat, + "detailed_status": scheduler_detailed_status, + "instances": scheduler_instances, }, "triggerer": { "status": triggerer_status, "latest_triggerer_heartbeat": latest_triggerer_heartbeat, + "detailed_status": triggerer_detailed_status, + "instances": triggerer_instances, }, "dag_processor": { "status": dag_processor_status, "latest_dag_processor_heartbeat": latest_dag_processor_heartbeat, + "detailed_status": dag_processor_detailed_status, + "instances": dag_processor_instances, }, } 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 3bfc5425fd7e6..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,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 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/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/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 new file mode 100644 index 0000000000000..b9968d8c2146c --- /dev/null +++ b/airflow-core/src/airflow/migrations/versions/0128_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: 7a98f1b7dbd3 +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 = "7a98f1b7dbd3" +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 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; }; /** 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. 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..a0c1e050f206c 100644 --- a/airflow-core/tests/unit/api/common/test_airflow_health.py +++ b/airflow-core/tests/unit/api/common/test_airflow_health.py @@ -16,76 +16,96 @@ # 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, 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 -@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,50 +113,148 @@ 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, + } + ], }, } assert health_status == expected_status + + +ALIVE_SCHEDULER_JOB_MOCK = MagicMock(spec=Job) +ALIVE_SCHEDULER_JOB_MOCK.hostname = "scheduler-alive" +ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime(2024, 2, 1) +ALIVE_SCHEDULER_JOB_MOCK.is_alive = MagicMock(return_value=True) + +DEAD_SCHEDULER_JOB_MOCK = MagicMock(spec=Job) +DEAD_SCHEDULER_JOB_MOCK.hostname = "scheduler-dead" +DEAD_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime(2024, 1, 1) +DEAD_SCHEDULER_JOB_MOCK.is_alive = MagicMock(return_value=False) + + +@patch("airflow.api.common.airflow_health.get_jobs_health") +def test_get_airflow_health_degraded_when_some_instances_unhealthy(mock_get_jobs_health): + mock_get_jobs_health.side_effect = [ + [ALIVE_SCHEDULER_JOB_MOCK, DEAD_SCHEDULER_JOB_MOCK], + [], + [], + ] + 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_SCHEDULER_JOB_MOCK.hostname, + "status": HEALTHY, + "latest_scheduler_heartbeat": ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(), + }, + { + "hostname": DEAD_SCHEDULER_JOB_MOCK.hostname, + "status": UNHEALTHY, + "latest_scheduler_heartbeat": DEAD_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(), + }, + ] + assert ( + health_status["scheduler"]["latest_scheduler_heartbeat"] + == ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat() + ) + + +@provide_session +def test_get_jobs_health_returns_running_jobs_ordered_by_heartbeat(session): + clear_db_jobs() + try: + older = timezone.utcnow() - timedelta(minutes=5) + newer = timezone.utcnow() + older_job = Job(state=JobState.RUNNING, latest_heartbeat=older, hostname="older") + newer_job = Job(state=JobState.RUNNING, latest_heartbeat=newer, hostname="newer") + failed_job = Job(state=JobState.FAILED, latest_heartbeat=newer, hostname="failed") + other_type_job = Job(state=JobState.RUNNING, latest_heartbeat=newer, hostname="triggerer") + SchedulerJobRunner(job=older_job) + SchedulerJobRunner(job=newer_job) + SchedulerJobRunner(job=failed_job) + TriggererJobRunner(job=other_type_job, capacity=1) + session.add_all([older_job, newer_job, failed_job, other_type_job]) + session.commit() + + jobs = get_jobs_health(SchedulerJobRunner, session=session) + + assert [job.hostname for job in jobs] == ["newer", "older"] + finally: + clear_db_jobs() 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..cb4c725449241 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 @@ -187,6 +189,25 @@ def test_get_jobs_filters_by_dag_id(self, test_client, session: Session): 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 d4b3d97f93252..e696ec43811b3 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, }, } @@ -133,6 +136,58 @@ def test_health_with_dag_processor(self, mock_get_airflow_health, test_client): 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 @@ 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, }, } 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, 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 abdf52ad125a1..6b6e429b600c9 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 @@ def test_bundle_names_passed(self, mock_runner, configure_testing_dag_bundle): 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 0cc9a10b52a6b..1ab983ba50a84 100644 --- a/airflow-core/tests/unit/cli/commands/test_triggerer_command.py +++ b/airflow-core/tests/unit/cli/commands/test_triggerer_command.py @@ -127,6 +127,7 @@ def test_team_name_passed_through(self, mock_serve, mock_triggerer_job_runner, m mock_triggerer_job_runner.assert_called_once_with( job=mock.ANY, capacity=mock.ANY, queues=None, team_name="team_a" ) + 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") diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py index 31a77b50b6592..75fd23a0dbe70 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py @@ -243,6 +243,14 @@ def test_triggerer_job_runner_stores_team_name(team_name): job = Job() runner = TriggererJobRunner(job, capacity=10, team_name=team_name) assert runner.team_name == team_name + assert job.team_name == team_name + + +def test_triggerer_job_runner_preserves_existing_job_team_name(): + job = Job(team_name="existing-team") + runner = TriggererJobRunner(job, capacity=10, team_name="other-team") + assert runner.team_name == "other-team" + assert job.team_name == "existing-team" @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 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.