From eb48444257531eff5aa36c18fc10c1a2242ea1a3 Mon Sep 17 00:00:00 2001 From: Karthikeyan Singaravelan Date: Thu, 6 Jul 2023 00:19:07 +0530 Subject: [PATCH] Updates health check endpoint to include dag_processor status. --- airflow/api/common/airflow_health.py | 19 +++++++++++++++++++ airflow/api_connexion/openapi/v1.yaml | 18 ++++++++++++++++++ .../api_connexion/schemas/health_schema.py | 7 +++++++ airflow/www/static/js/types/api-generated.ts | 17 +++++++++++++++++ .../logging-monitoring/check-health.rst | 8 ++++++++ tests/api/common/test_airflow_health.py | 17 ++++++++++++++++- 6 files changed, 85 insertions(+), 1 deletion(-) diff --git a/airflow/api/common/airflow_health.py b/airflow/api/common/airflow_health.py index 3174add64842d..5d37de540a498 100644 --- a/airflow/api/common/airflow_health.py +++ b/airflow/api/common/airflow_health.py @@ -18,6 +18,7 @@ from typing import Any +from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner from airflow.jobs.scheduler_job_runner import SchedulerJobRunner from airflow.jobs.triggerer_job_runner import TriggererJobRunner @@ -30,8 +31,10 @@ def get_airflow_health() -> dict[str, Any]: metadatabase_status = HEALTHY latest_scheduler_heartbeat = None latest_triggerer_heartbeat = None + latest_dag_processor_heartbeat = None scheduler_status = UNHEALTHY triggerer_status: str | None = UNHEALTHY + dag_processor_status: str | None = UNHEALTHY try: latest_scheduler_job = SchedulerJobRunner.most_recent_job() @@ -55,6 +58,18 @@ def get_airflow_health() -> dict[str, Any]: except Exception: metadatabase_status = UNHEALTHY + try: + latest_dag_processor_job = DagProcessorJobRunner.most_recent_job() + + if latest_dag_processor_job: + latest_dag_processor_heartbeat = latest_dag_processor_job.latest_heartbeat.isoformat() + if latest_dag_processor_job.is_alive(): + dag_processor_status = HEALTHY + else: + dag_processor_status = None + except Exception: + metadatabase_status = UNHEALTHY + airflow_health_status = { "metadatabase": {"status": metadatabase_status}, "scheduler": { @@ -65,6 +80,10 @@ def get_airflow_health() -> dict[str, Any]: "status": triggerer_status, "latest_triggerer_heartbeat": latest_triggerer_heartbeat, }, + "dag_processor": { + "status": dag_processor_status, + "latest_dag_processor_heartbeat": latest_dag_processor_heartbeat, + }, } return airflow_health_status diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml index 8bf976aa46272..3e1dcdedd7409 100644 --- a/airflow/api_connexion/openapi/v1.yaml +++ b/airflow/api_connexion/openapi/v1.yaml @@ -3084,6 +3084,8 @@ components: $ref: '#/components/schemas/SchedulerStatus' triggerer: $ref: '#/components/schemas/TriggererStatus' + dag_processor: + $ref: '#/components/schemas/DagProcessorStatus' MetadatabaseStatus: type: object @@ -3121,6 +3123,22 @@ components: readOnly: true nullable: true + DagProcessorStatus: + type: object + description: | + The status and the latest dag processor heartbeat. + + *New in version 2.6.3* + properties: + status: + $ref: '#/components/schemas/HealthStatus' + latest_dag_processor_heartbeat: + description: The time the dag processor last did a heartbeat. + type: string + format: datetime + readOnly: true + nullable: true + Pool: description: The pool type: object diff --git a/airflow/api_connexion/schemas/health_schema.py b/airflow/api_connexion/schemas/health_schema.py index 5391ce5ae235c..5e184171fb397 100644 --- a/airflow/api_connexion/schemas/health_schema.py +++ b/airflow/api_connexion/schemas/health_schema.py @@ -41,12 +41,19 @@ class TriggererInfoSchema(BaseInfoSchema): latest_triggerer_heartbeat = fields.String(dump_only=True) +class DagProcessorInfoSchema(BaseInfoSchema): + """Schema for DagProcessor info.""" + + latest_dag_processor_heartbeat = fields.String(dump_only=True) + + class HealthInfoSchema(Schema): """Schema for the Health endpoint.""" metadatabase = fields.Nested(MetaDatabaseInfoSchema) scheduler = fields.Nested(SchedulerInfoSchema) triggerer = fields.Nested(TriggererInfoSchema) + dag_processor = fields.Nested(DagProcessorInfoSchema) health_schema = HealthInfoSchema() diff --git a/airflow/www/static/js/types/api-generated.ts b/airflow/www/static/js/types/api-generated.ts index 347f76e08f1e9..aa9bebf4d6dec 100644 --- a/airflow/www/static/js/types/api-generated.ts +++ b/airflow/www/static/js/types/api-generated.ts @@ -1192,6 +1192,7 @@ export interface components { metadatabase?: components["schemas"]["MetadatabaseStatus"]; scheduler?: components["schemas"]["SchedulerStatus"]; triggerer?: components["schemas"]["TriggererStatus"]; + dag_processor?: components["schemas"]["DagProcessorStatus"]; }; /** @description The status of the metadatabase. */ MetadatabaseStatus: { @@ -1219,6 +1220,19 @@ export interface components { */ latest_triggerer_heartbeat?: string | null; }; + /** + * @description The status and the latest dag processor heartbeat. + * + * *New in version 2.6.3* + */ + DagProcessorStatus: { + status?: components["schemas"]["HealthStatus"]; + /** + * Format: datetime + * @description The time the dag processor last did a heartbeat. + */ + latest_dag_processor_heartbeat?: string | null; + }; /** @description The pool */ Pool: { /** @description The name of pool. */ @@ -4667,6 +4681,9 @@ export type SchedulerStatus = CamelCasedPropertiesDeep< export type TriggererStatus = CamelCasedPropertiesDeep< components["schemas"]["TriggererStatus"] >; +export type DagProcessorStatus = CamelCasedPropertiesDeep< + components["schemas"]["DagProcessorStatus"] +>; export type Pool = CamelCasedPropertiesDeep; export type PoolCollection = CamelCasedPropertiesDeep< components["schemas"]["PoolCollection"] diff --git a/docs/apache-airflow/administration-and-deployment/logging-monitoring/check-health.rst b/docs/apache-airflow/administration-and-deployment/logging-monitoring/check-health.rst index 798343f9d1e0e..58e8834078668 100644 --- a/docs/apache-airflow/administration-and-deployment/logging-monitoring/check-health.rst +++ b/docs/apache-airflow/administration-and-deployment/logging-monitoring/check-health.rst @@ -51,6 +51,10 @@ To check the health status of your Airflow instance, you can simply access the e "triggerer":{ "status":"healthy", "latest_triggerer_heartbeat":"2018-12-26 17:16:12+00:00" + }, + "dag_processor":{ + "status":"healthy", + "latest_dag_processor_heartbeat":"2018-12-26 17:16:12+00:00" } } @@ -71,6 +75,10 @@ To check the health status of your Airflow instance, you can simply access the e 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. + * 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. + Please keep in mind that the HTTP response code of ``/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/tests/api/common/test_airflow_health.py b/tests/api/common/test_airflow_health.py index 1f778f7a25076..20bd101cabe84 100644 --- a/tests/api/common/test_airflow_health.py +++ b/tests/api/common/test_airflow_health.py @@ -22,6 +22,7 @@ from airflow.api.common.airflow_health import ( HEALTHY, UNHEALTHY, + DagProcessorJobRunner, SchedulerJobRunner, TriggererJobRunner, get_airflow_health, @@ -31,13 +32,14 @@ def test_get_airflow_health_only_metadatabase_healthy(): SchedulerJobRunner.most_recent_job = MagicMock(return_value=None) TriggererJobRunner.most_recent_job = MagicMock(return_value=None) - + DagProcessorJobRunner.most_recent_job = MagicMock(return_value=None) 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}, } assert health_status == expected_status @@ -46,6 +48,7 @@ def test_get_airflow_health_only_metadatabase_healthy(): def test_get_airflow_health_metadatabase_unhealthy(): SchedulerJobRunner.most_recent_job = MagicMock(side_effect=Exception) TriggererJobRunner.most_recent_job = MagicMock(side_effect=Exception) + DagProcessorJobRunner.most_recent_job = MagicMock(side_effect=Exception) health_status = get_airflow_health() @@ -53,6 +56,7 @@ def test_get_airflow_health_metadatabase_unhealthy(): "metadatabase": {"status": UNHEALTHY}, "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None}, "triggerer": {"status": UNHEALTHY, "latest_triggerer_heartbeat": None}, + "dag_processor": {"status": UNHEALTHY, "latest_dag_processor_heartbeat": None}, } assert health_status == expected_status @@ -64,6 +68,7 @@ def test_get_airflow_health_scheduler_healthy_no_triggerer(): latest_scheduler_job_mock.is_alive = MagicMock(return_value=True) SchedulerJobRunner.most_recent_job = MagicMock(return_value=latest_scheduler_job_mock) TriggererJobRunner.most_recent_job = MagicMock(return_value=None) + DagProcessorJobRunner.most_recent_job = MagicMock(return_value=None) health_status = get_airflow_health() @@ -74,6 +79,7 @@ def test_get_airflow_health_scheduler_healthy_no_triggerer(): "latest_scheduler_heartbeat": latest_scheduler_job_mock.latest_heartbeat.isoformat(), }, "triggerer": {"status": None, "latest_triggerer_heartbeat": None}, + "dag_processor": {"status": None, "latest_dag_processor_heartbeat": None}, } assert health_status == expected_status @@ -83,8 +89,13 @@ def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record(): latest_triggerer_job_mock = MagicMock() latest_triggerer_job_mock.latest_heartbeat = datetime.now() latest_triggerer_job_mock.is_alive = MagicMock(return_value=True) + latest_dag_processor_job_mock = MagicMock() + latest_dag_processor_job_mock.latest_heartbeat = datetime.now() + latest_dag_processor_job_mock.is_alive = MagicMock(return_value=True) + SchedulerJobRunner.most_recent_job = MagicMock(return_value=None) TriggererJobRunner.most_recent_job = MagicMock(return_value=latest_triggerer_job_mock) + DagProcessorJobRunner.most_recent_job = MagicMock(return_value=latest_dag_processor_job_mock) health_status = get_airflow_health() @@ -95,6 +106,10 @@ def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record(): "status": HEALTHY, "latest_triggerer_heartbeat": latest_triggerer_job_mock.latest_heartbeat.isoformat(), }, + "dag_processor": { + "status": HEALTHY, + "latest_dag_processor_heartbeat": latest_dag_processor_job_mock.latest_heartbeat.isoformat(), + }, } assert health_status == expected_status