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 f42ea8702f439..5289e7f042c41 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 @@ -51,7 +51,6 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc "detailed_status": "healthy", "instances": [ { - "status": "healthy", "hostname": "scheduler-1.example.com", "latest_scheduler_heartbeat": "2018-12-26T17:15:11+00:00" } @@ -63,16 +62,9 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc "detailed_status": "degraded", "instances": [ { - "status": "healthy", "hostname": "triggerer-1.example.com", "latest_triggerer_heartbeat": "2018-12-26T17:16:12+00:00", "team_name": "team-a" - }, - { - "status": "unhealthy", - "hostname": "triggerer-2.example.com", - "latest_triggerer_heartbeat": "2018-12-26T17:10:00+00:00", - "team_name": null } ] }, @@ -82,7 +74,6 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc "detailed_status": "healthy", "instances": [ { - "status": "healthy", "hostname": "dag-processor-1.example.com", "latest_dag_processor_heartbeat": "2018-12-26T17:16:12+00:00", "bundle_names": ["dags-team-a"] @@ -100,11 +91,28 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc * ``status`` (legacy aggregate): ``"healthy"`` if **any** running instance is alive, otherwise ``"unhealthy"`` (including when no running jobs exist for that component). - * ``detailed_status``: reflects the full set of running instances: - - * ``"healthy"`` — every running instance is alive - * ``"degraded"`` — some instances are alive and some are not - * ``"down"`` — no running instance is alive (including when no jobs exist) + * ``detailed_status``: whether every part of that component's work is being covered by a live instance. + What counts as "every part" differs per component, because only some of them divide their work up: + + * **Dag processor** — the parts are the Dag bundles in ``[dag_processor] dag_bundle_config_list``. + A processor started without ``--bundle-name`` covers every configured bundle; + one started with it covers only the bundles it was given. + ``"healthy"`` when every configured bundle has a live processor, ``"degraded"`` when only some do, + ``"down"`` when none do. + * **Triggerer** — with ``[core] multi_team`` enabled, the parts are the teams those bundles are scoped to + (plus the unscoped bundles), because a triggerer only picks up triggers for its own team. + ``"healthy"`` when every team scope has a live triggerer, ``"degraded"`` when only some do, + ``"down"`` when none do. With multi-team disabled, no team filtering applies, so any live triggerer + covers everything: ``"healthy"`` if one is alive, ``"down"`` if none is. + * **Scheduler** — schedulers are symmetric and share no partitioned work, so there is nothing partial + to report: ``"healthy"`` if at least one is alive, ``"down"`` if none is. ``"degraded"`` is never + returned for the scheduler. Use ``instances`` to see how many replicas are up, and your orchestrator + or the ``scheduler_heartbeat`` metric to alert on reduced scheduling throughput. + + Because the expected set of work comes from configuration rather than from the job table, + ``detailed_status`` is unaffected by how instances come and go. Restarting an instance — including after + a ``SIGKILL``, an out-of-memory kill, or a node eviction, none of which let Airflow mark the old job row + as finished — does not report the component as ``"degraded"``. * ``latest_*_heartbeat``: the most recent heartbeat among running jobs of that type (ordered by heartbeat descending), or ``null`` when there are none. @@ -113,9 +121,13 @@ including per-instance details when multiple schedulers, triggerers, or Dag proc ``[triggerer] triggerer_health_check_threshold``, ``[dag_processor] health_check_threshold``). - * ``instances``: one entry per **running** job of that type (``null`` when there are none). Each entry includes: + * ``instances``: one entry per **live** instance of that type — ``null`` when none is live. Airflow cannot + mark a job as finished when its process is killed abruptly (``SIGKILL``, an out-of-memory kill, a node + eviction), so the job row of such an instance is never closed; listing it would show a host that is gone + and, under an orchestrator that assigns a fresh hostname on each restart, will never come back. + ``status`` and ``latest_*_heartbeat`` are still derived from every unfinished job, so how long a dead + component has been silent stays visible after it drops out of ``instances``. Each entry includes: - * ``status``: ``"healthy"`` or ``"unhealthy"`` for that instance * ``hostname``: host where the component is running * the corresponding ``latest_*_heartbeat`` for that instance * ``team_name`` (triggerer only): team the triggerer is scoped to, or ``null`` when unscoped diff --git a/airflow-core/src/airflow/api/common/airflow_health.py b/airflow-core/src/airflow/api/common/airflow_health.py index e95a2966b3acc..def85190e4401 100644 --- a/airflow-core/src/airflow/api/common/airflow_health.py +++ b/airflow-core/src/airflow/api/common/airflow_health.py @@ -16,10 +16,14 @@ # under the License. from __future__ import annotations +import logging +from enum import Enum from typing import TYPE_CHECKING, Any from sqlalchemy import select +from airflow.configuration import conf +from airflow.dag_processing.bundles.manager import _get_configured_bundle_team_names from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner from airflow.jobs.job import Job from airflow.jobs.scheduler_job_runner import SchedulerJobRunner @@ -29,10 +33,22 @@ if TYPE_CHECKING: from sqlalchemy.orm import Session -HEALTHY = "healthy" -UNHEALTHY = "unhealthy" -DEGRADED = "degraded" -DOWN = "down" +log = logging.getLogger(__name__) + + +class HealthStatus(str, Enum): + """Aggregate health of a component: whether it has at least one live instance.""" + + HEALTHY = "healthy" + UNHEALTHY = "unhealthy" + + +class DetailedHealthStatus(str, Enum): + """How much of a component's work has a live instance covering it.""" + + HEALTHY = "healthy" + DEGRADED = "degraded" + DOWN = "down" @provide_session @@ -54,14 +70,26 @@ def _job_instance_health(job: Job, heartbeat_field_name: str) -> dict[str, Any]: heartbeat = job.latest_heartbeat.isoformat() if job.latest_heartbeat else None return { "hostname": job.hostname, - "status": HEALTHY if job.is_alive() else UNHEALTHY, heartbeat_field_name: heartbeat, } -def _legacy_status(jobs: list[Job]) -> str: +def _live_jobs(jobs: list[Job]) -> list[Job]: + """ + Narrow unfinished job rows down to the replicas that are actually running. + + ``instances`` describes the deployment as it is now, so a row left behind by a replica that + never got to write its ``end_date`` must not be listed: the host it names is gone, and under an + orchestrator that assigns a fresh hostname per restart it will never come back. The unfinished + rows are still what ``status`` and ``latest_*_heartbeat`` are derived from, so how long a dead + component has been silent stays visible even once it drops out of ``instances``. + """ + return [job for job in jobs if job.is_alive()] + + +def _legacy_status(jobs: list[Job]) -> HealthStatus: """Top-level status: healthy if any instance is alive.""" - return HEALTHY if any(job.is_alive() for job in jobs) else UNHEALTHY + return HealthStatus.HEALTHY if any(job.is_alive() for job in jobs) else HealthStatus.UNHEALTHY def _triggerer_instance_health(job: Job) -> dict[str, Any]: @@ -78,19 +106,74 @@ def _dag_processor_instance_health(job: Job) -> dict[str, Any]: } -def _aggregate_detailed_status(jobs: list[Job]) -> str: - """detailed_status: healthy (all alive), degraded (some alive), down (none alive).""" - alive_count = sum(1 for job in jobs if job.is_alive()) - if alive_count == 0: - return DOWN - if alive_count == len(jobs): - return HEALTHY - return DEGRADED +# ``detailed_status`` answers "is every part of this component's work being done", which needs a +# denominator. Counting job rows cannot supply one: ``end_date`` is only written by a cooperative +# shutdown, so a replica lost to SIGKILL, an OOM kill, or a node eviction leaves an unfinished row +# behind forever and a restarted replica adds a second one. The denominator is therefore taken from +# the declared work partition instead, which is unaffected by how replicas come and go: +# +# * Dag processor -- the bundles in ``[dag_processor] dag_bundle_config_list``. +# * Triggerer -- the team scopes those bundles declare, since a triggerer only picks up triggers for +# its own team (see ``Trigger.ids_for_triggerer``). +# * Scheduler -- schedulers are symmetric, so there is no partition and no partial state to report. + + +def _configured_bundle_teams() -> dict[str, str | None]: + """Map every configured Dag bundle to the team owning it, empty when the config is unreadable.""" + try: + return _get_configured_bundle_team_names() + except Exception: + # A health probe must not fail on malformed bundle config; callers fall back to liveness only. + log.warning("Could not read the Dag bundle configuration", exc_info=True) + return {} + + +def _liveness_status(jobs: list[Job]) -> DetailedHealthStatus: + """Status for a component with no declared work partition: one live replica covers everything.""" + return DetailedHealthStatus.HEALTHY if any(job.is_alive() for job in jobs) else DetailedHealthStatus.DOWN + + +def _coverage_status(expected: set[Any], covered: set[Any]) -> DetailedHealthStatus: + """Status from how much of a component's declared work partition its live replicas cover.""" + if not expected - covered: + return DetailedHealthStatus.HEALTHY + if expected & covered: + return DetailedHealthStatus.DEGRADED + return DetailedHealthStatus.DOWN + + +def _dag_processor_detailed_status(jobs: list[Job]) -> DetailedHealthStatus: + """Status from bundle coverage, which partitions processor work with or without multi-team mode.""" + expected = set(_configured_bundle_teams()) + if not expected: + return _liveness_status(jobs) + + covered: set[str] = set() + for job in jobs: + if job.is_alive(): + # A processor started without ``--bundle-name`` parses every configured bundle. + covered |= expected if not job.bundle_names else expected & set(job.bundle_names) + return _coverage_status(expected, covered) + + +def _triggerer_detailed_status(jobs: list[Job]) -> DetailedHealthStatus: + """Status from team coverage, since a triggerer only picks up triggers for its own team.""" + if not conf.getboolean("core", "multi_team"): + # Outside multi-team mode no team filter is applied, so any live triggerer serves every trigger. + return _liveness_status(jobs) + + # ``None`` is a scope in its own right: triggers from bundles that declare no team are only picked + # up by a triggerer started without ``--team-name``, so it must stay in the expected set. + expected = set(_configured_bundle_teams().values()) + if not expected: + return _liveness_status(jobs) + + return _coverage_status(expected, {job.team_name for job in jobs if job.is_alive()}) def get_airflow_health() -> dict[str, Any]: """Get the health for Airflow metadatabase, scheduler, triggerer, and dag processor.""" - metadatabase_status = HEALTHY + metadatabase_status = HealthStatus.HEALTHY latest_scheduler_heartbeat = None latest_triggerer_heartbeat = None @@ -100,55 +183,52 @@ def get_airflow_health() -> dict[str, Any]: triggerer_instances: list[dict[str, Any]] | None = None dag_processor_instances: list[dict[str, Any]] | None = None - scheduler_status = UNHEALTHY - triggerer_status = UNHEALTHY - dag_processor_status = UNHEALTHY + scheduler_status = HealthStatus.UNHEALTHY + triggerer_status = HealthStatus.UNHEALTHY + dag_processor_status = HealthStatus.UNHEALTHY - scheduler_detailed_status = DOWN - triggerer_detailed_status = DOWN - dag_processor_detailed_status = DOWN + scheduler_detailed_status = DetailedHealthStatus.DOWN + triggerer_detailed_status = DetailedHealthStatus.DOWN + dag_processor_detailed_status = DetailedHealthStatus.DOWN try: scheduler_jobs = get_jobs_health(SchedulerJobRunner) scheduler_status = _legacy_status(scheduler_jobs) - scheduler_detailed_status = _aggregate_detailed_status(scheduler_jobs) - if scheduler_jobs: + scheduler_detailed_status = _liveness_status(scheduler_jobs) + if scheduler_jobs and scheduler_jobs[0].latest_heartbeat: + latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat() + if live_scheduler_jobs := _live_jobs(scheduler_jobs): scheduler_instances = [ - _job_instance_health(job, "latest_scheduler_heartbeat") for job in scheduler_jobs + _job_instance_health(job, "latest_scheduler_heartbeat") for job in live_scheduler_jobs ] - - if scheduler_jobs[0].latest_heartbeat: - latest_scheduler_heartbeat = scheduler_jobs[0].latest_heartbeat.isoformat() except Exception: - metadatabase_status = UNHEALTHY + metadatabase_status = HealthStatus.UNHEALTHY try: triggerer_jobs = get_jobs_health(TriggererJobRunner) triggerer_status = _legacy_status(triggerer_jobs) - triggerer_detailed_status = _aggregate_detailed_status(triggerer_jobs) - if triggerer_jobs: - triggerer_instances = [_triggerer_instance_health(job) for job in triggerer_jobs] - - if triggerer_jobs[0].latest_heartbeat: - latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat() + triggerer_detailed_status = _triggerer_detailed_status(triggerer_jobs) + if triggerer_jobs and triggerer_jobs[0].latest_heartbeat: + latest_triggerer_heartbeat = triggerer_jobs[0].latest_heartbeat.isoformat() + if live_triggerer_jobs := _live_jobs(triggerer_jobs): + triggerer_instances = [_triggerer_instance_health(job) for job in live_triggerer_jobs] except Exception: - metadatabase_status = UNHEALTHY - triggerer_status = UNHEALTHY - triggerer_detailed_status = DOWN + metadatabase_status = HealthStatus.UNHEALTHY + triggerer_status = HealthStatus.UNHEALTHY + triggerer_detailed_status = DetailedHealthStatus.DOWN try: dag_processor_jobs = get_jobs_health(DagProcessorJobRunner) dag_processor_status = _legacy_status(dag_processor_jobs) - dag_processor_detailed_status = _aggregate_detailed_status(dag_processor_jobs) - if dag_processor_jobs: - dag_processor_instances = [_dag_processor_instance_health(job) for job in dag_processor_jobs] - - if dag_processor_jobs[0].latest_heartbeat: - latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat() + dag_processor_detailed_status = _dag_processor_detailed_status(dag_processor_jobs) + if dag_processor_jobs and dag_processor_jobs[0].latest_heartbeat: + latest_dag_processor_heartbeat = dag_processor_jobs[0].latest_heartbeat.isoformat() + if live_dag_processor_jobs := _live_jobs(dag_processor_jobs): + dag_processor_instances = [_dag_processor_instance_health(job) for job in live_dag_processor_jobs] except Exception: - metadatabase_status = UNHEALTHY - dag_processor_status = UNHEALTHY - dag_processor_detailed_status = DOWN + metadatabase_status = HealthStatus.UNHEALTHY + dag_processor_status = HealthStatus.UNHEALTHY + dag_processor_detailed_status = DetailedHealthStatus.DOWN airflow_health_status = { "metadatabase": {"status": metadatabase_status}, 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 5bf60bd83bd05..442a5e9f44aa3 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 @@ -16,23 +16,26 @@ # under the License. from __future__ import annotations +from airflow.api.common.airflow_health import DetailedHealthStatus, HealthStatus from airflow.api_fastapi.core_api.base import BaseModel class BaseInfoResponse(BaseModel): """Base info serializer for responses.""" - status: str | None + status: HealthStatus | None -class SchedulerInstanceInfoResponse(BaseInfoResponse): +# Instances carry no status of their own: only running replicas are listed, and how healthy the set +# of them is together is what the component's ``status`` and ``detailed_status`` report. +class SchedulerInstanceInfoResponse(BaseModel): """Scheduler instance info serializer for responses.""" hostname: str | None latest_scheduler_heartbeat: str | None -class TriggererInstanceInfoResponse(BaseInfoResponse): +class TriggererInstanceInfoResponse(BaseModel): """Triggerer instance info serializer for responses.""" hostname: str | None @@ -40,7 +43,7 @@ class TriggererInstanceInfoResponse(BaseInfoResponse): team_name: str | None -class DagProcessorInstanceInfoResponse(BaseInfoResponse): +class DagProcessorInstanceInfoResponse(BaseModel): """Dag processor instance info serializer for responses.""" hostname: str | None @@ -52,7 +55,7 @@ class SchedulerInfoResponse(BaseInfoResponse): """Scheduler info serializer for responses.""" latest_scheduler_heartbeat: str | None - detailed_status: str | None + detailed_status: DetailedHealthStatus | None instances: list[SchedulerInstanceInfoResponse] | None = None @@ -60,7 +63,7 @@ class TriggererInfoResponse(BaseInfoResponse): """Triggerer info serializer for responses.""" latest_triggerer_heartbeat: str | None - detailed_status: str | None + detailed_status: DetailedHealthStatus | None instances: list[TriggererInstanceInfoResponse] | None = None @@ -68,7 +71,7 @@ class DagProcessorInfoResponse(BaseInfoResponse): """DagProcessor info serializer for responses.""" latest_dag_processor_heartbeat: str | None - detailed_status: str | None + detailed_status: DetailedHealthStatus | None instances: list[DagProcessorInstanceInfoResponse] | None = None 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 e6c0274e61d79..6b6bae9b1dc18 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 @@ -11820,9 +11820,8 @@ components: properties: status: anyOf: - - type: string + - $ref: '#/components/schemas/HealthStatus' - type: 'null' - title: Status type: object required: - status @@ -14117,9 +14116,8 @@ components: properties: status: anyOf: - - type: string + - $ref: '#/components/schemas/HealthStatus' - type: 'null' - title: Status latest_dag_processor_heartbeat: anyOf: - type: string @@ -14127,9 +14125,8 @@ components: title: Latest Dag Processor Heartbeat detailed_status: anyOf: - - type: string + - $ref: '#/components/schemas/DetailedHealthStatus' - type: 'null' - title: Detailed Status instances: anyOf: - items: @@ -14146,11 +14143,6 @@ components: description: DagProcessor info serializer for responses. DagProcessorInstanceInfoResponse: properties: - status: - anyOf: - - type: string - - type: 'null' - title: Status hostname: anyOf: - type: string @@ -14170,7 +14162,6 @@ components: title: Bundle Names type: object required: - - status - hostname - latest_dag_processor_heartbeat - bundle_names @@ -14452,6 +14443,14 @@ components: This is the set of allowable values for the ``warning_type`` field in the DagWarning model.' + DetailedHealthStatus: + type: string + enum: + - healthy + - degraded + - down + title: DetailedHealthStatus + description: How much of a component's work has a live instance covering it. DryRunBackfillCollectionResponse: properties: backfills: @@ -14964,6 +14963,14 @@ components: - triggerer title: HealthInfoResponse description: Health serializer for responses. + HealthStatus: + type: string + enum: + - healthy + - unhealthy + title: HealthStatus + description: 'Aggregate health of a component: whether it has at least one live + instance.' ImportErrorCollectionResponse: properties: import_errors: @@ -15703,9 +15710,8 @@ components: properties: status: anyOf: - - type: string + - $ref: '#/components/schemas/HealthStatus' - type: 'null' - title: Status latest_scheduler_heartbeat: anyOf: - type: string @@ -15713,9 +15719,8 @@ components: title: Latest Scheduler Heartbeat detailed_status: anyOf: - - type: string + - $ref: '#/components/schemas/DetailedHealthStatus' - type: 'null' - title: Detailed Status instances: anyOf: - items: @@ -15732,11 +15737,6 @@ components: description: Scheduler info serializer for responses. SchedulerInstanceInfoResponse: properties: - status: - anyOf: - - type: string - - type: 'null' - title: Status hostname: anyOf: - type: string @@ -15749,7 +15749,6 @@ components: title: Latest Scheduler Heartbeat type: object required: - - status - hostname - latest_scheduler_heartbeat title: SchedulerInstanceInfoResponse @@ -16871,9 +16870,8 @@ components: properties: status: anyOf: - - type: string + - $ref: '#/components/schemas/HealthStatus' - type: 'null' - title: Status latest_triggerer_heartbeat: anyOf: - type: string @@ -16881,9 +16879,8 @@ components: title: Latest Triggerer Heartbeat detailed_status: anyOf: - - type: string + - $ref: '#/components/schemas/DetailedHealthStatus' - type: 'null' - title: Detailed Status instances: anyOf: - items: @@ -16900,11 +16897,6 @@ components: description: Triggerer info serializer for responses. TriggererInstanceInfoResponse: properties: - status: - anyOf: - - type: string - - type: 'null' - title: Status hostname: anyOf: - type: string @@ -16922,7 +16914,6 @@ components: title: Team Name type: object required: - - status - hostname - latest_triggerer_heartbeat - team_name diff --git a/airflow-core/src/airflow/dag_processing/bundles/manager.py b/airflow-core/src/airflow/dag_processing/bundles/manager.py index 292bb440a21e7..2c20f923739bd 100644 --- a/airflow-core/src/airflow/dag_processing/bundles/manager.py +++ b/airflow-core/src/airflow/dag_processing/bundles/manager.py @@ -130,6 +130,32 @@ def _parse_bundle_config(config_list) -> list[_ExternalBundleConfig]: return list(bundles.values()) +def _read_bundle_config_list() -> list[_ExternalBundleConfig]: + config_list = conf.getjson("dag_processor", "dag_bundle_config_list") + if not config_list: + return [] + if not isinstance(config_list, list): + raise AirflowConfigException( + "Section `dag_processor` key `dag_bundle_config_list` " + f"must be list but got {config_list.__class__}" + ) + return _parse_bundle_config(config_list) + + +def _get_configured_bundle_team_names() -> dict[str, str | None]: + """ + Get the team owning each explicitly configured Dag bundle. + + This reads the config rather than going through ``DagBundlesManager`` so that callers who only + need the declared bundle partition neither import every bundle class nor see the example-Dag + bundles that ``DagBundlesManager.parse_config`` injects when ``[core] load_examples`` is set -- + those are added by Airflow, not declared by the deployment. + + :return: mapping of bundle name to team name, ``None`` for bundles that are not team scoped. + """ + return {cfg.name: cfg.team_name for cfg in _read_bundle_config_list()} + + def _add_example_dag_bundle(bundle_config_list: list[_ExternalBundleConfig]): from airflow import example_dags @@ -274,15 +300,9 @@ def parse_config(self) -> None: if self._bundle_config: return - config_list = conf.getjson("dag_processor", "dag_bundle_config_list") - if not config_list: + bundle_config_list = _read_bundle_config_list() + if not bundle_config_list: return - if not isinstance(config_list, list): - raise AirflowConfigException( - "Section `dag_processor` key `dag_bundle_config_list` " - f"must be list but got {config_list.__class__}" - ) - bundle_config_list = _parse_bundle_config(config_list) if conf.getboolean("core", "LOAD_EXAMPLES"): _add_example_dag_bundle(bundle_config_list) _add_provider_example_dags_to_bundle(bundle_config_list) 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 075af8fe558a2..4771e7de12e8e 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 @@ -1023,13 +1023,12 @@ export const $BaseInfoResponse = { status: { anyOf: [ { - type: 'string' + '$ref': '#/components/schemas/HealthStatus' }, { type: 'null' } - ], - title: 'Status' + ] } }, type: 'object', @@ -4523,13 +4522,12 @@ export const $DagProcessorInfoResponse = { status: { anyOf: [ { - type: 'string' + '$ref': '#/components/schemas/HealthStatus' }, { type: 'null' } - ], - title: 'Status' + ] }, latest_dag_processor_heartbeat: { anyOf: [ @@ -4545,13 +4543,12 @@ export const $DagProcessorInfoResponse = { detailed_status: { anyOf: [ { - type: 'string' + '$ref': '#/components/schemas/DetailedHealthStatus' }, { type: 'null' } - ], - title: 'Detailed Status' + ] }, instances: { anyOf: [ @@ -4576,17 +4573,6 @@ export const $DagProcessorInfoResponse = { export const $DagProcessorInstanceInfoResponse = { properties: { - status: { - anyOf: [ - { - type: 'string' - }, - { - type: 'null' - } - ], - title: 'Status' - }, hostname: { anyOf: [ { @@ -4625,7 +4611,7 @@ export const $DagProcessorInstanceInfoResponse = { } }, type: 'object', - required: ['status', 'hostname', 'latest_dag_processor_heartbeat', 'bundle_names'], + required: ['hostname', 'latest_dag_processor_heartbeat', 'bundle_names'], title: 'DagProcessorInstanceInfoResponse', description: 'Dag processor instance info serializer for responses.' } as const; @@ -4957,6 +4943,13 @@ This is the set of allowable values for the \`\`warning_type\`\` field in the DagWarning model.` } as const; +export const $DetailedHealthStatus = { + type: 'string', + enum: ['healthy', 'degraded', 'down'], + title: 'DetailedHealthStatus', + description: "How much of a component's work has a live instance covering it." +} as const; + export const $DryRunBackfillCollectionResponse = { properties: { backfills: { @@ -5730,6 +5723,13 @@ export const $HealthInfoResponse = { description: 'Health serializer for responses.' } as const; +export const $HealthStatus = { + type: 'string', + enum: ['healthy', 'unhealthy'], + title: 'HealthStatus', + description: 'Aggregate health of a component: whether it has at least one live instance.' +} as const; + export const $ImportErrorCollectionResponse = { properties: { import_errors: { @@ -6830,13 +6830,12 @@ export const $SchedulerInfoResponse = { status: { anyOf: [ { - type: 'string' + '$ref': '#/components/schemas/HealthStatus' }, { type: 'null' } - ], - title: 'Status' + ] }, latest_scheduler_heartbeat: { anyOf: [ @@ -6852,13 +6851,12 @@ export const $SchedulerInfoResponse = { detailed_status: { anyOf: [ { - type: 'string' + '$ref': '#/components/schemas/DetailedHealthStatus' }, { type: 'null' } - ], - title: 'Detailed Status' + ] }, instances: { anyOf: [ @@ -6883,17 +6881,6 @@ export const $SchedulerInfoResponse = { export const $SchedulerInstanceInfoResponse = { properties: { - status: { - anyOf: [ - { - type: 'string' - }, - { - type: 'null' - } - ], - title: 'Status' - }, hostname: { anyOf: [ { @@ -6918,7 +6905,7 @@ export const $SchedulerInstanceInfoResponse = { } }, type: 'object', - required: ['status', 'hostname', 'latest_scheduler_heartbeat'], + required: ['hostname', 'latest_scheduler_heartbeat'], title: 'SchedulerInstanceInfoResponse', description: 'Scheduler instance info serializer for responses.' } as const; @@ -8679,13 +8666,12 @@ export const $TriggererInfoResponse = { status: { anyOf: [ { - type: 'string' + '$ref': '#/components/schemas/HealthStatus' }, { type: 'null' } - ], - title: 'Status' + ] }, latest_triggerer_heartbeat: { anyOf: [ @@ -8701,13 +8687,12 @@ export const $TriggererInfoResponse = { detailed_status: { anyOf: [ { - type: 'string' + '$ref': '#/components/schemas/DetailedHealthStatus' }, { type: 'null' } - ], - title: 'Detailed Status' + ] }, instances: { anyOf: [ @@ -8732,17 +8717,6 @@ export const $TriggererInfoResponse = { export const $TriggererInstanceInfoResponse = { properties: { - status: { - anyOf: [ - { - type: 'string' - }, - { - type: 'null' - } - ], - title: 'Status' - }, hostname: { anyOf: [ { @@ -8778,7 +8752,7 @@ export const $TriggererInstanceInfoResponse = { } }, type: 'object', - required: ['status', 'hostname', 'latest_triggerer_heartbeat', 'team_name'], + required: ['hostname', 'latest_triggerer_heartbeat', 'team_name'], title: 'TriggererInstanceInfoResponse', description: 'Triggerer instance info serializer for responses.' } as const; 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 f74d0a87cab8b..dc08fe4d80e2e 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 @@ -303,7 +303,7 @@ export type BackfillResponse = { * Base info serializer for responses. */ export type BaseInfoResponse = { - status: string | null; + status: HealthStatus | null; }; /** @@ -1181,9 +1181,9 @@ export type DagBundleResponse = { * DagProcessor info serializer for responses. */ export type DagProcessorInfoResponse = { - status: string | null; + status: HealthStatus | null; latest_dag_processor_heartbeat: string | null; - detailed_status: string | null; + detailed_status: DetailedHealthStatus | null; instances?: Array | null; }; @@ -1191,7 +1191,6 @@ export type DagProcessorInfoResponse = { * 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; @@ -1311,6 +1310,11 @@ export type DagVersionResponse = { */ export type DagWarningType = 'asset conflict' | 'duplicate dag id' | 'non-existent pool' | 'runtime varying value'; +/** + * How much of a component's work has a live instance covering it. + */ +export type DetailedHealthStatus = 'healthy' | 'degraded' | 'down'; + /** * Backfill collection serializer for responses in dry-run mode. */ @@ -1510,6 +1514,11 @@ export type HealthInfoResponse = { dag_processor?: DagProcessorInfoResponse | null; }; +/** + * Aggregate health of a component: whether it has at least one live instance. + */ +export type HealthStatus = 'healthy' | 'unhealthy'; + /** * Import Error Collection Response. */ @@ -1783,9 +1792,9 @@ export type ReprocessBehavior = 'failed' | 'completed' | 'none'; * Scheduler info serializer for responses. */ export type SchedulerInfoResponse = { - status: string | null; + status: HealthStatus | null; latest_scheduler_heartbeat: string | null; - detailed_status: string | null; + detailed_status: DetailedHealthStatus | null; instances?: Array | null; }; @@ -1793,7 +1802,6 @@ export type SchedulerInfoResponse = { * Scheduler instance info serializer for responses. */ export type SchedulerInstanceInfoResponse = { - status: string | null; hostname: string | null; latest_scheduler_heartbeat: string | null; }; @@ -2140,9 +2148,9 @@ export type TriggerResponse = { * Triggerer info serializer for responses. */ export type TriggererInfoResponse = { - status: string | null; + status: HealthStatus | null; latest_triggerer_heartbeat: string | null; - detailed_status: string | null; + detailed_status: DetailedHealthStatus | null; instances?: Array | null; }; @@ -2150,7 +2158,6 @@ export type TriggererInfoResponse = { * 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/tests/unit/api/common/test_airflow_health.py b/airflow-core/tests/unit/api/common/test_airflow_health.py index 79648ac99ac36..cb10bcb79acda 100644 --- a/airflow-core/tests/unit/api/common/test_airflow_health.py +++ b/airflow-core/tests/unit/api/common/test_airflow_health.py @@ -23,10 +23,11 @@ from airflow._shared.timezones import timezone from airflow.api.common.airflow_health import ( - DEGRADED, - DOWN, - HEALTHY, - UNHEALTHY, + DetailedHealthStatus, + HealthStatus, + _configured_bundle_teams, + _dag_processor_detailed_status, + _triggerer_detailed_status, get_airflow_health, get_jobs_health, ) @@ -35,6 +36,7 @@ from airflow.jobs.triggerer_job_runner import TriggererJobRunner from airflow.utils.session import provide_session +from tests_common.test_utils.config import conf_vars from tests_common.test_utils.db import clear_db_jobs pytestmark = pytest.mark.db_test @@ -42,6 +44,14 @@ STALE_HEARTBEAT_AGE = timedelta(minutes=5) +def _patch_bundles(bundle_teams: dict[str, str | None]): + """Patch the declared bundle/team partition that ``detailed_status`` is measured against.""" + return patch( + "airflow.api.common.airflow_health._configured_bundle_teams", + return_value=bundle_teams, + ) + + def _mock_job( *, hostname: str, @@ -61,9 +71,9 @@ def _mock_job( def _empty_component(heartbeat_field: str) -> dict: return { - "status": UNHEALTHY, + "status": HealthStatus.UNHEALTHY, heartbeat_field: None, - "detailed_status": DOWN, + "detailed_status": DetailedHealthStatus.DOWN, "instances": None, } @@ -131,7 +141,7 @@ def test_get_airflow_health_no_jobs(mock_get_jobs_health): health_status = get_airflow_health() assert health_status == { - "metadatabase": {"status": HEALTHY}, + "metadatabase": {"status": HealthStatus.HEALTHY}, "scheduler": _empty_component("latest_scheduler_heartbeat"), "triggerer": _empty_component("latest_triggerer_heartbeat"), "dag_processor": _empty_component("latest_dag_processor_heartbeat"), @@ -143,7 +153,7 @@ def test_get_airflow_health_metadatabase_unhealthy(mock_get_jobs_health): health_status = get_airflow_health() assert health_status == { - "metadatabase": {"status": UNHEALTHY}, + "metadatabase": {"status": HealthStatus.UNHEALTHY}, "scheduler": _empty_component("latest_scheduler_heartbeat"), "triggerer": _empty_component("latest_triggerer_heartbeat"), "dag_processor": _empty_component("latest_dag_processor_heartbeat"), @@ -156,15 +166,14 @@ def test_get_airflow_health_one_alive_job(mock_get_jobs_health): health_status = get_airflow_health() assert health_status == { - "metadatabase": {"status": HEALTHY}, + "metadatabase": {"status": HealthStatus.HEALTHY}, "scheduler": { - "status": HEALTHY, + "status": HealthStatus.HEALTHY, "latest_scheduler_heartbeat": ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(), - "detailed_status": HEALTHY, + "detailed_status": DetailedHealthStatus.HEALTHY, "instances": [ { "hostname": ALIVE_SCHEDULER_JOB_MOCK.hostname, - "status": HEALTHY, "latest_scheduler_heartbeat": ALIVE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(), } ], @@ -183,24 +192,15 @@ def test_get_airflow_health_mixed_alive_and_stale_jobs(mock_get_jobs_health): ] health_status = get_airflow_health() - assert health_status["scheduler"]["status"] == HEALTHY - assert health_status["scheduler"]["detailed_status"] == DEGRADED + assert health_status["scheduler"]["status"] == HealthStatus.HEALTHY + # Schedulers are symmetric, so a stale row alongside a live one is not a partial outage. + assert health_status["scheduler"]["detailed_status"] == DetailedHealthStatus.HEALTHY + # The rows the two dead replicas left behind name hosts that are gone, so they are not listed. 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"] @@ -219,25 +219,14 @@ def test_get_airflow_health_all_stale_jobs(mock_get_jobs_health): ] health_status = get_airflow_health() - 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(), - }, - ] + assert health_status["scheduler"]["status"] == HealthStatus.UNHEALTHY + assert health_status["scheduler"]["detailed_status"] == DetailedHealthStatus.DOWN + assert health_status["scheduler"]["instances"] is None + # A component with no live replica still reports when it was last heard from. + assert ( + health_status["scheduler"]["latest_scheduler_heartbeat"] + == STALE_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat() + ) @patch("airflow.api.common.airflow_health.get_jobs_health") @@ -245,21 +234,15 @@ def test_get_airflow_health_mixed_triggerers_include_team_name(mock_get_jobs_hea mock_get_jobs_health.side_effect = [[], [ALIVE_TRIGGERER_JOB_MOCK, STALE_TRIGGERER_JOB_MOCK], []] health_status = get_airflow_health() - assert health_status["triggerer"]["status"] == HEALTHY - assert health_status["triggerer"]["detailed_status"] == DEGRADED + assert health_status["triggerer"]["status"] == HealthStatus.HEALTHY + # Multi-team is off here, so every live triggerer serves every trigger regardless of team. + assert health_status["triggerer"]["detailed_status"] == DetailedHealthStatus.HEALTHY 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"] @@ -267,35 +250,34 @@ def test_get_airflow_health_mixed_triggerers_include_team_name(mock_get_jobs_hea ) +@_patch_bundles({"bundle-a": None}) @patch("airflow.api.common.airflow_health.get_jobs_health") -def test_get_airflow_health_triggerer_and_dag_processor_healthy(mock_get_jobs_health): +def test_get_airflow_health_triggerer_and_dag_processor_healthy(mock_get_jobs_health, _mock_bundles): 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}, + "metadatabase": {"status": HealthStatus.HEALTHY}, "scheduler": _empty_component("latest_scheduler_heartbeat"), "triggerer": { - "status": HEALTHY, + "status": HealthStatus.HEALTHY, "latest_triggerer_heartbeat": ALIVE_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(), - "detailed_status": HEALTHY, + "detailed_status": DetailedHealthStatus.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, + "status": HealthStatus.HEALTHY, "latest_dag_processor_heartbeat": ALIVE_DAG_PROCESSOR_JOB_MOCK.latest_heartbeat.isoformat(), - "detailed_status": HEALTHY, + "detailed_status": DetailedHealthStatus.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, } @@ -304,6 +286,198 @@ def test_get_airflow_health_triggerer_and_dag_processor_healthy(mock_get_jobs_he } +def _processor(*, alive: bool, bundle_names: list[str] | None) -> MagicMock: + return _mock_job( + hostname=f"dag-processor-{'alive' if alive else 'stale'}", + heartbeat=datetime(2024, 2, 1), + alive=alive, + bundle_names=bundle_names, + ) + + +def _triggerer(*, alive: bool, team_name: str | None) -> MagicMock: + return _mock_job( + hostname=f"triggerer-{team_name}-{'alive' if alive else 'stale'}", + heartbeat=datetime(2024, 2, 1), + alive=alive, + team_name=team_name, + ) + + +class TestDagProcessorDetailedStatus: + """``detailed_status`` measures which configured bundles a live processor is parsing.""" + + BUNDLES = {"bundle-a": None, "bundle-b": None} + + @pytest.mark.parametrize( + ("jobs", "expected"), + [ + pytest.param( + [_processor(alive=True, bundle_names=["bundle-a", "bundle-b"])], + DetailedHealthStatus.HEALTHY, + id="one_processor_parses_every_bundle", + ), + pytest.param( + [ + _processor(alive=True, bundle_names=["bundle-a"]), + _processor(alive=True, bundle_names=["bundle-b"]), + ], + DetailedHealthStatus.HEALTHY, + id="a_processor_per_bundle", + ), + pytest.param( + [_processor(alive=True, bundle_names=None)], + DetailedHealthStatus.HEALTHY, + id="processor_without_bundle_name_parses_all", + ), + pytest.param( + [_processor(alive=True, bundle_names=[])], + DetailedHealthStatus.HEALTHY, + id="empty_bundle_names_parses_all", + ), + pytest.param( + [_processor(alive=True, bundle_names=["bundle-a"])], + DetailedHealthStatus.DEGRADED, + id="one_bundle_left_unparsed", + ), + pytest.param( + [_processor(alive=False, bundle_names=["bundle-a", "bundle-b"])], + DetailedHealthStatus.DOWN, + id="only_processor_is_stale", + ), + pytest.param( + [_processor(alive=True, bundle_names=["bundle-removed-from-config"])], + DetailedHealthStatus.DOWN, + id="processor_parses_nothing_configured", + ), + pytest.param([], DetailedHealthStatus.DOWN, id="no_processor_at_all"), + ], + ) + def test_bundle_coverage(self, jobs, expected): + with _patch_bundles(self.BUNDLES): + assert _dag_processor_detailed_status(jobs) == expected + + def test_restarted_processor_leaves_the_component_healthy(self): + """A hard-killed replica keeps an unfinished job row; its replacement must still read healthy.""" + jobs = [ + _processor(alive=True, bundle_names=["bundle-a", "bundle-b"]), + _processor(alive=False, bundle_names=["bundle-a", "bundle-b"]), + ] + + with _patch_bundles(self.BUNDLES): + assert _dag_processor_detailed_status(jobs) == DetailedHealthStatus.HEALTHY + + @pytest.mark.parametrize( + ("jobs", "expected"), + [ + pytest.param( + [_processor(alive=True, bundle_names=None)], + DetailedHealthStatus.HEALTHY, + id="a_processor_is_alive", + ), + pytest.param( + [_processor(alive=False, bundle_names=None)], + DetailedHealthStatus.DOWN, + id="no_processor_is_alive", + ), + ], + ) + def test_falls_back_to_liveness_without_configured_bundles(self, jobs, expected): + with _patch_bundles({}): + assert _dag_processor_detailed_status(jobs) == expected + + +class TestTriggererDetailedStatus: + """Under multi-team a triggerer only serves its own team, so every team scope needs one.""" + + BUNDLES = {"bundle-a": "team-a", "bundle-b": "team-b", "bundle-shared": None} + + @pytest.mark.parametrize( + ("jobs", "expected"), + [ + pytest.param( + [ + _triggerer(alive=True, team_name="team-a"), + _triggerer(alive=True, team_name="team-b"), + _triggerer(alive=True, team_name=None), + ], + DetailedHealthStatus.HEALTHY, + id="every_team_scope_covered", + ), + pytest.param( + [ + _triggerer(alive=True, team_name="team-a"), + _triggerer(alive=True, team_name=None), + ], + DetailedHealthStatus.DEGRADED, + id="one_team_has_no_triggerer", + ), + pytest.param( + [ + _triggerer(alive=True, team_name="team-a"), + _triggerer(alive=True, team_name="team-b"), + ], + DetailedHealthStatus.DEGRADED, + id="unscoped_bundles_have_no_triggerer", + ), + pytest.param( + [_triggerer(alive=False, team_name="team-a")], + DetailedHealthStatus.DOWN, + id="no_triggerer_is_alive", + ), + pytest.param( + [_triggerer(alive=True, team_name="team-of-a-removed-bundle")], + DetailedHealthStatus.DOWN, + id="triggerer_serves_no_configured_team", + ), + pytest.param([], DetailedHealthStatus.DOWN, id="no_triggerer_at_all"), + ], + ) + def test_team_coverage(self, jobs, expected): + with conf_vars({("core", "multi_team"): "True"}), _patch_bundles(self.BUNDLES): + assert _triggerer_detailed_status(jobs) == expected + + def test_restarted_triggerer_leaves_the_component_healthy(self): + jobs = [ + _triggerer(alive=True, team_name="team-a"), + _triggerer(alive=False, team_name="team-a"), + ] + + with conf_vars({("core", "multi_team"): "True"}), _patch_bundles({"bundle-a": "team-a"}): + assert _triggerer_detailed_status(jobs) == DetailedHealthStatus.HEALTHY + + @pytest.mark.parametrize( + ("jobs", "expected"), + [ + pytest.param( + [_triggerer(alive=True, team_name="team-a"), _triggerer(alive=False, team_name="team-b")], + DetailedHealthStatus.HEALTHY, + id="a_triggerer_is_alive", + ), + pytest.param( + [_triggerer(alive=False, team_name=None)], + DetailedHealthStatus.DOWN, + id="no_triggerer_is_alive", + ), + ], + ) + def test_ignores_teams_without_multi_team(self, jobs, expected): + with conf_vars({("core", "multi_team"): "False"}), _patch_bundles(self.BUNDLES): + assert _triggerer_detailed_status(jobs) == expected + + def test_falls_back_to_liveness_without_configured_bundles(self): + with conf_vars({("core", "multi_team"): "True"}), _patch_bundles({}): + assert ( + _triggerer_detailed_status([_triggerer(alive=True, team_name="team-a")]) + == DetailedHealthStatus.HEALTHY + ) + + +@conf_vars({("dag_processor", "dag_bundle_config_list"): "not json"}) +def test_configured_bundle_teams_survives_unreadable_config(): + assert _configured_bundle_teams() == {} + + class TestAirflowHealthFromDb: @pytest.fixture(autouse=True) def cleanup_jobs(self): @@ -346,7 +520,7 @@ def test_get_airflow_health_no_jobs(self): health_status = get_airflow_health() assert health_status == { - "metadatabase": {"status": HEALTHY}, + "metadatabase": {"status": HealthStatus.HEALTHY}, "scheduler": _empty_component("latest_scheduler_heartbeat"), "triggerer": _empty_component("latest_triggerer_heartbeat"), "dag_processor": _empty_component("latest_dag_processor_heartbeat"), @@ -367,13 +541,12 @@ def test_get_airflow_health_one_alive_job(self, *, session): 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["metadatabase"]["status"] == HealthStatus.HEALTHY + assert health_status["scheduler"]["status"] == HealthStatus.HEALTHY + assert health_status["scheduler"]["detailed_status"] == DetailedHealthStatus.HEALTHY assert health_status["scheduler"]["instances"] == [ { "hostname": job.hostname, - "status": HEALTHY, "latest_scheduler_heartbeat": heartbeat.isoformat(), } ] @@ -389,34 +562,21 @@ def test_get_airflow_health_mixed_alive_and_stale_jobs(self, *, session): 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( + _create_job(session, SchedulerJobRunner, hostname="scheduler-stale", heartbeat=stale_heartbeat) + _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"]["status"] == HealthStatus.HEALTHY + assert health_status["scheduler"]["detailed_status"] == DetailedHealthStatus.HEALTHY 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() @@ -425,25 +585,17 @@ 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), - ] + _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 - ] + assert health_status["scheduler"]["status"] == HealthStatus.UNHEALTHY + assert health_status["scheduler"]["detailed_status"] == DetailedHealthStatus.DOWN + assert health_status["scheduler"]["instances"] is None + assert health_status["scheduler"]["latest_scheduler_heartbeat"] == first.isoformat() @provide_session def test_get_airflow_health_mixed_triggerers_include_team_name(self, testing_team, *, session): @@ -456,7 +608,7 @@ def test_get_airflow_health_mixed_triggerers_include_team_name(self, testing_tea heartbeat=alive_heartbeat, team_name=testing_team.name, ) - stale = _create_job( + _create_job( session, TriggererJobRunner, hostname="triggerer-stale", @@ -467,20 +619,13 @@ def test_get_airflow_health_mixed_triggerers_include_team_name(self, testing_tea health_status = get_airflow_health() - assert health_status["triggerer"]["status"] == HEALTHY - assert health_status["triggerer"]["detailed_status"] == DEGRADED + assert health_status["triggerer"]["status"] == HealthStatus.HEALTHY + assert health_status["triggerer"]["detailed_status"] == DetailedHealthStatus.HEALTHY 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/dag_processing/bundles/test_dag_bundle_manager.py b/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py index 08e63c3f77d11..7d23af9afefc0 100644 --- a/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py +++ b/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py @@ -29,7 +29,11 @@ from sqlalchemy import func, select, update from airflow.dag_processing.bundles.base import BaseDagBundle -from airflow.dag_processing.bundles.manager import DagBundlesManager, _guess_best_bundle_for_fileloc +from airflow.dag_processing.bundles.manager import ( + DagBundlesManager, + _get_configured_bundle_team_names, + _guess_best_bundle_for_fileloc, +) from airflow.exceptions import AirflowConfigException from airflow.models.dag import DagModel from airflow.models.dag_version import DagVersion @@ -141,6 +145,40 @@ def path(self) -> Path: } ] +TEAM_BUNDLE_CONFIG = [ + { + "name": "team-bundle", + "classpath": "unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle", + "kwargs": {"refresh_interval": 1}, + "team_name": "team-a", + }, + { + "name": "unscoped-bundle", + "classpath": "unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle", + "kwargs": {"refresh_interval": 1}, + }, +] + + +@pytest.mark.parametrize("load_examples", ["False", "True"]) +def test_get_configured_bundle_team_names(load_examples): + with conf_vars( + { + ("core", "load_examples"): load_examples, + ("core", "multi_team"): "True", + ("dag_processor", "dag_bundle_config_list"): json.dumps(TEAM_BUNDLE_CONFIG), + } + ): + assert _get_configured_bundle_team_names() == { + "team-bundle": "team-a", + "unscoped-bundle": None, + } + + +@conf_vars({("dag_processor", "dag_bundle_config_list"): "[]"}) +def test_get_configured_bundle_team_names_without_config(): + assert _get_configured_bundle_team_names() == {} + def test_get_bundle(): """Test that get_bundle builds and returns a bundle.""" diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py b/airflow-ctl/src/airflowctl/api/datamodels/generated.py index 60282ddef5d40..5e340d7667dd8 100644 --- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py +++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py @@ -130,14 +130,6 @@ class AsyncConnectionTestResponse(BaseModel): created_at: Annotated[datetime, Field(title="Created At")] -class BaseInfoResponse(BaseModel): - """ - Base info serializer for responses. - """ - - status: Annotated[str | None, Field(title="Status")] - - class BulkActionNotOnExistence(str, Enum): """ Bulk Action to be taken if the entity does not exist. @@ -548,7 +540,6 @@ class DagProcessorInstanceInfoResponse(BaseModel): 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")] @@ -705,6 +696,16 @@ class DagWarningType(str, Enum): RUNTIME_VARYING_VALUE = "runtime varying value" +class DetailedHealthStatus(str, Enum): + """ + How much of a component's work has a live instance covering it. + """ + + HEALTHY = "healthy" + DEGRADED = "degraded" + DOWN = "down" + + class DryRunBackfillResponse(BaseModel): """ Backfill serializer for responses in dry-run mode. @@ -806,6 +807,15 @@ class HTTPExceptionResponse(BaseModel): detail: Annotated[str | dict[str, Any], Field(title="Detail")] +class HealthStatus(str, Enum): + """ + Aggregate health of a component: whether it has at least one live instance. + """ + + HEALTHY = "healthy" + UNHEALTHY = "unhealthy" + + class ImportErrorResponse(BaseModel): """ Import Error Response. @@ -1030,7 +1040,6 @@ class SchedulerInstanceInfoResponse(BaseModel): 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")] @@ -1244,7 +1253,6 @@ class TriggererInstanceInfoResponse(BaseModel): 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")] @@ -1541,6 +1549,14 @@ class BackfillResponse(BaseModel): dag_display_name: Annotated[str, Field(title="Dag Display Name")] +class BaseInfoResponse(BaseModel): + """ + Base info serializer for responses. + """ + + status: HealthStatus | None + + class BulkCreateActionConnectionBody(BaseModel): model_config = ConfigDict( extra="forbid", @@ -2006,9 +2022,9 @@ class DagProcessorInfoResponse(BaseModel): DagProcessor info serializer for responses. """ - status: Annotated[str | None, Field(title="Status")] + status: HealthStatus | None latest_dag_processor_heartbeat: Annotated[str | None, Field(title="Latest Dag Processor Heartbeat")] - detailed_status: Annotated[str | None, Field(title="Detailed Status")] + detailed_status: DetailedHealthStatus | None instances: Annotated[list[DagProcessorInstanceInfoResponse] | None, Field(title="Instances")] = None @@ -2181,9 +2197,9 @@ class SchedulerInfoResponse(BaseModel): Scheduler info serializer for responses. """ - status: Annotated[str | None, Field(title="Status")] + status: HealthStatus | None latest_scheduler_heartbeat: Annotated[str | None, Field(title="Latest Scheduler Heartbeat")] - detailed_status: Annotated[str | None, Field(title="Detailed Status")] + detailed_status: DetailedHealthStatus | None instances: Annotated[list[SchedulerInstanceInfoResponse] | None, Field(title="Instances")] = None @@ -2320,9 +2336,9 @@ class TriggererInfoResponse(BaseModel): Triggerer info serializer for responses. """ - status: Annotated[str | None, Field(title="Status")] + status: HealthStatus | None latest_triggerer_heartbeat: Annotated[str | None, Field(title="Latest Triggerer Heartbeat")] - detailed_status: Annotated[str | None, Field(title="Detailed Status")] + detailed_status: DetailedHealthStatus | None instances: Annotated[list[TriggererInstanceInfoResponse] | None, Field(title="Instances")] = None