From e7ce77dd6e2426e121d37a14f18238ec761e3740 Mon Sep 17 00:00:00 2001 From: Daniel Anya <49489918+daniel-anya@users.noreply.github.com> Date: Thu, 17 Nov 2022 14:28:36 -0600 Subject: [PATCH 1/7] Updates health check endpoint to include `triggerer` status --- .../endpoints/health_endpoint.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/airflow/api_connexion/endpoints/health_endpoint.py b/airflow/api_connexion/endpoints/health_endpoint.py index f833a5d72815b..22152787abe39 100644 --- a/airflow/api_connexion/endpoints/health_endpoint.py +++ b/airflow/api_connexion/endpoints/health_endpoint.py @@ -19,6 +19,7 @@ from airflow.api_connexion.schemas.health_schema import health_schema from airflow.api_connexion.types import APIResponse from airflow.jobs.scheduler_job import SchedulerJob +from airflow.jobs.triggerer_job import TriggererJob HEALTHY = "healthy" UNHEALTHY = "unhealthy" @@ -28,7 +29,9 @@ def get_health() -> APIResponse: """Return the health of the airflow scheduler and metadatabase.""" metadatabase_status = HEALTHY latest_scheduler_heartbeat = None + latest_triggerer_heartbeat = None scheduler_status = UNHEALTHY + triggerer_status = None try: scheduler_job = SchedulerJob.most_recent_job() @@ -38,6 +41,18 @@ def get_health() -> APIResponse: scheduler_status = HEALTHY except Exception: metadatabase_status = UNHEALTHY + + try: + triggerer_job = TriggererJob.most_recent_job() + + if triggerer_job: + latest_triggerer_heartbeat = triggerer_job.latest_heartbeat.isoformat() + if triggerer_job.is_alive(): + triggerer_status = HEALTHY + else: + triggerer_status = UNHEALTHY + except Exception: + metadatabase_status = UNHEALTHY payload = { "metadatabase": {"status": metadatabase_status}, @@ -45,6 +60,10 @@ def get_health() -> APIResponse: "status": scheduler_status, "latest_scheduler_heartbeat": latest_scheduler_heartbeat, }, + "triggerer": { + "status": triggerer_status, + "latest_triggerer_heartbeat": latest_triggerer_heartbeat, + }, } return health_schema.dump(payload) From 328e11ca64468f2b05ca55de4e670e1f5f31179c Mon Sep 17 00:00:00 2001 From: Daniel Anya <49489918+daniel-anya@users.noreply.github.com> Date: Thu, 17 Nov 2022 15:28:54 -0600 Subject: [PATCH 2/7] Remove redundant space to fix linting check failure --- airflow/api_connexion/endpoints/health_endpoint.py | 1 - 1 file changed, 1 deletion(-) diff --git a/airflow/api_connexion/endpoints/health_endpoint.py b/airflow/api_connexion/endpoints/health_endpoint.py index 22152787abe39..99f79c2786f63 100644 --- a/airflow/api_connexion/endpoints/health_endpoint.py +++ b/airflow/api_connexion/endpoints/health_endpoint.py @@ -41,7 +41,6 @@ def get_health() -> APIResponse: scheduler_status = HEALTHY except Exception: metadatabase_status = UNHEALTHY - try: triggerer_job = TriggererJob.most_recent_job() From fc4db411e36357f792e7757bc479634f83b1641c Mon Sep 17 00:00:00 2001 From: daniel-anya Date: Thu, 17 Nov 2022 20:12:15 -0600 Subject: [PATCH 3/7] Updates documentation with new health check response schema --- docs/apache-airflow/logging-monitoring/check-health.rst | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/docs/apache-airflow/logging-monitoring/check-health.rst b/docs/apache-airflow/logging-monitoring/check-health.rst index 4e0777eccbf89..2aa4386df5bd7 100644 --- a/docs/apache-airflow/logging-monitoring/check-health.rst +++ b/docs/apache-airflow/logging-monitoring/check-health.rst @@ -47,6 +47,10 @@ To check the health status of your Airflow instance, you can simply access the e "scheduler":{ "status":"healthy", "latest_scheduler_heartbeat":"2018-12-26 17:15:11+00:00" + }, + "triggerer":{ + "status":"healthy", + "latest_triggerer_heartbeat":"2018-12-26 17:16:12+00:00" } } @@ -63,6 +67,10 @@ To check the health status of your Airflow instance, you can simply access the e * 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 + * 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. + 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). From 9940a6a13c51e53fc6859ae1cf2429e73b45bb30 Mon Sep 17 00:00:00 2001 From: daniel-anya Date: Fri, 18 Nov 2022 09:14:57 -0600 Subject: [PATCH 4/7] Updates health schema with triggerer info --- airflow/api_connexion/schemas/health_schema.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/airflow/api_connexion/schemas/health_schema.py b/airflow/api_connexion/schemas/health_schema.py index 67155406c1a79..5391ce5ae235c 100644 --- a/airflow/api_connexion/schemas/health_schema.py +++ b/airflow/api_connexion/schemas/health_schema.py @@ -30,16 +30,23 @@ class MetaDatabaseInfoSchema(BaseInfoSchema): class SchedulerInfoSchema(BaseInfoSchema): - """Schema for Metadatabase info.""" + """Schema for Scheduler info.""" latest_scheduler_heartbeat = fields.String(dump_only=True) +class TriggererInfoSchema(BaseInfoSchema): + """Schema for Triggerer info.""" + + latest_triggerer_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) health_schema = HealthInfoSchema() From b48346552556e7a70da60444475dd64ab6137b68 Mon Sep 17 00:00:00 2001 From: daniel-anya Date: Tue, 22 Nov 2022 12:21:51 -0600 Subject: [PATCH 5/7] Updates 'triggerer_status' logic for setting unhealth/None --- airflow/api_connexion/endpoints/health_endpoint.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/airflow/api_connexion/endpoints/health_endpoint.py b/airflow/api_connexion/endpoints/health_endpoint.py index 99f79c2786f63..8b889c869bdec 100644 --- a/airflow/api_connexion/endpoints/health_endpoint.py +++ b/airflow/api_connexion/endpoints/health_endpoint.py @@ -31,7 +31,7 @@ def get_health() -> APIResponse: latest_scheduler_heartbeat = None latest_triggerer_heartbeat = None scheduler_status = UNHEALTHY - triggerer_status = None + triggerer_status = UNHEALTHY try: scheduler_job = SchedulerJob.most_recent_job() @@ -48,8 +48,8 @@ def get_health() -> APIResponse: latest_triggerer_heartbeat = triggerer_job.latest_heartbeat.isoformat() if triggerer_job.is_alive(): triggerer_status = HEALTHY - else: - triggerer_status = UNHEALTHY + else: + triggerer_status = None except Exception: metadatabase_status = UNHEALTHY From 4bf5fbcc332d9ec89fef86dce3c1aa2de319f743 Mon Sep 17 00:00:00 2001 From: daniel-anya Date: Tue, 22 Nov 2022 13:45:37 -0600 Subject: [PATCH 6/7] Fix 'triggerer_status' type --- airflow/api_connexion/endpoints/health_endpoint.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/airflow/api_connexion/endpoints/health_endpoint.py b/airflow/api_connexion/endpoints/health_endpoint.py index 8b889c869bdec..b89af33b4bc9e 100644 --- a/airflow/api_connexion/endpoints/health_endpoint.py +++ b/airflow/api_connexion/endpoints/health_endpoint.py @@ -16,6 +16,8 @@ # under the License. from __future__ import annotations +from typing import Optional + from airflow.api_connexion.schemas.health_schema import health_schema from airflow.api_connexion.types import APIResponse from airflow.jobs.scheduler_job import SchedulerJob @@ -31,7 +33,7 @@ def get_health() -> APIResponse: latest_scheduler_heartbeat = None latest_triggerer_heartbeat = None scheduler_status = UNHEALTHY - triggerer_status = UNHEALTHY + triggerer_status: Optional[str] = UNHEALTHY try: scheduler_job = SchedulerJob.most_recent_job() From 006919c5cc278bcbed70b62e8104925e6668a54e Mon Sep 17 00:00:00 2001 From: daniel-anya Date: Wed, 23 Nov 2022 11:57:32 -0600 Subject: [PATCH 7/7] Fix CI linting errors --- airflow/api_connexion/endpoints/health_endpoint.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/airflow/api_connexion/endpoints/health_endpoint.py b/airflow/api_connexion/endpoints/health_endpoint.py index b89af33b4bc9e..7fa7f21cd9e28 100644 --- a/airflow/api_connexion/endpoints/health_endpoint.py +++ b/airflow/api_connexion/endpoints/health_endpoint.py @@ -16,8 +16,6 @@ # under the License. from __future__ import annotations -from typing import Optional - from airflow.api_connexion.schemas.health_schema import health_schema from airflow.api_connexion.types import APIResponse from airflow.jobs.scheduler_job import SchedulerJob @@ -33,7 +31,7 @@ def get_health() -> APIResponse: latest_scheduler_heartbeat = None latest_triggerer_heartbeat = None scheduler_status = UNHEALTHY - triggerer_status: Optional[str] = UNHEALTHY + triggerer_status: str | None = UNHEALTHY try: scheduler_job = SchedulerJob.most_recent_job()