Repository: apache/airflow
Pull request: AIP84:  Check standalone_dag_processor config in get_airflow_health()  and update health endpoint
closes: https://github.com/apache/airflow/issues/44253

### Motivation 
Right now the UI has to check the scheduler config for standalone_dag_processor in order to determine if the dag_processor health check should be rendered.

This PR does that check-in get_airflow_health() and also handles this inthe  ` public/health` endpoint.


### Testing
**Response with DAG Processor enabled**

<img width="1458" alt="image" src="https://github.com/user-attachments/assets/0488152f-a59c-427d-9995-9b3cc15fb9b6">


**Response with DAG Processor disabled**

<img width="1398" alt="image" src="https://github.com/user-attachments/assets/af3f2666-c373-46e7-a303-4e5161d571e7">


<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.

Feel free to ping committers for the review!

In case of an existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->



<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).


Problem statement:
Check standalone_dag_processor config in get_airflow_health()
Right now the UI has to check the scheduler config for `standalone_dag_processor` in order to determin if the dag_processor health  check should be rendered.

It would be better if we checked for `standalone_dag_processor` in the backend and simply not return a `dag_processor` object as part of the `public/health` endpoint if False

Hints from issue discussion:
@bbovenzi would like to work on this. Can I assign it to myself?
@vatsrahul1001 all yours!

Commit message: Merge branch 'AIP84-health-endpoint-skip-dag-processor' of github.com:astronomer/airflow into AIP84-health-endpoint-skip-dag-processor

Patch to review:
diff --git a/airflow/api/common/airflow_health.py b/airflow/api/common/airflow_health.py
index 5d37de540a498..043557fa0dbe3 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.configuration import conf
 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
@@ -28,13 +29,13 @@
 
 def get_airflow_health() -> dict[str, Any]:
     """Get the health for Airflow metadatabase, scheduler and triggerer."""
+    dag_processor_enabled = conf.getboolean("scheduler", "standalone_dag_processor")
     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()
@@ -58,18 +59,6 @@ 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": {
@@ -80,10 +69,27 @@ def get_airflow_health() -> dict[str, Any]:
             "status": triggerer_status,
             "latest_triggerer_heartbeat": latest_triggerer_heartbeat,
         },
-        "dag_processor": {
+    }
+
+    if dag_processor_enabled:
+        latest_dag_processor_heartbeat = None
+        dag_processor_status: str | None = 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["dag_processor"] = {
             "status": dag_processor_status,
             "latest_dag_processor_heartbeat": latest_dag_processor_heartbeat,
-        },
-    }
+        }
 
     return airflow_health_status
diff --git a/airflow/api_fastapi/core_api/datamodels/monitor.py b/airflow/api_fastapi/core_api/datamodels/monitor.py
index fbaf40b4e8416..6909729b80d87 100644
--- a/airflow/api_fastapi/core_api/datamodels/monitor.py
+++ b/airflow/api_fastapi/core_api/datamodels/monitor.py
@@ -16,6 +16,8 @@
 # under the License.
 from __future__ import annotations
 
+from pydantic import model_serializer
+
 from airflow.api_fastapi.core_api.base import BaseModel
 
 
@@ -49,4 +51,8 @@ class HealthInfoSchema(BaseModel):
     metadatabase: BaseInfoSchema
     scheduler: SchedulerInfoSchema
     triggerer: TriggererInfoSchema
-    dag_processor: DagProcessorInfoSchema
+    dag_processor: DagProcessorInfoSchema | None = None
+
+    @model_serializer()
+    def serialize(self):
+        return {k: v for k, v in self if not (k == "dag_processor" and v is None)}
diff --git a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
index 25ade1d917195..06e98fc87ffeb 100644
--- a/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
+++ b/airflow/api_fastapi/core_api/openapi/v1-generated.yaml
@@ -7437,13 +7437,14 @@ components:
         triggerer:
           $ref: '#/components/schemas/TriggererInfoSchema'
         dag_processor:
-          $ref: '#/components/schemas/DagProcessorInfoSchema'
+          anyOf:
+          - $ref: '#/components/schemas/DagProcessorInfoSchema'
+          - type: 'null'
       type: object
       required:
       - metadatabase
       - scheduler
       - triggerer
-      - dag_processor
       title: HealthInfoSchema
       description: Schema for the Health endpoint.
     HistoricalMetricDataResponse:
diff --git a/airflow/api_fastapi/core_api/routes/public/monitor.py b/airflow/api_fastapi/core_api/routes/public/monitor.py
index 38953a18b99fc..ef7c7b811404f 100644
--- a/airflow/api_fastapi/core_api/routes/public/monitor.py
+++ b/airflow/api_fastapi/core_api/routes/public/monitor.py
@@ -27,4 +27,4 @@
 @monitor_router.get("/health")
 def get_health() -> HealthInfoSchema:
     airflow_health_status = get_airflow_health()
-    return HealthInfoSchema.model_validate(airflow_health_status)
+    return HealthInfoSchema.model_validate(airflow_health_status, from_attributes=True)
diff --git a/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow/ui/openapi-gen/requests/schemas.gen.ts
index 84c63a7d7bc34..eab29a7fcffc0 100644
--- a/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -2823,11 +2823,18 @@ export const $HealthInfoSchema = {
       $ref: "#/components/schemas/TriggererInfoSchema",
     },
     dag_processor: {
-      $ref: "#/components/schemas/DagProcessorInfoSchema",
+      anyOf: [
+        {
+          $ref: "#/components/schemas/DagProcessorInfoSchema",
+        },
+        {
+          type: "null",
+        },
+      ],
     },
   },
   type: "object",
-  required: ["metadatabase", "scheduler", "triggerer", "dag_processor"],
+  required: ["metadatabase", "scheduler", "triggerer"],
   title: "HealthInfoSchema",
   description: "Schema for the Health endpoint.",
 } as const;
diff --git a/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow/ui/openapi-gen/requests/types.gen.ts
index 1adcdc8eba8f4..f60caae9b53ed 100644
--- a/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -684,7 +684,7 @@ export type HealthInfoSchema = {
   metadatabase: BaseInfoSchema;
   scheduler: SchedulerInfoSchema;
   triggerer: TriggererInfoSchema;
-  dag_processor: DagProcessorInfoSchema;
+  dag_processor?: DagProcessorInfoSchema | null;
 };
 
 /**
diff --git a/airflow/ui/src/pages/Dashboard/Health/Health.tsx b/airflow/ui/src/pages/Dashboard/Health/Health.tsx
index 0694283930ce9..29ec8602c28b9 100644
--- a/airflow/ui/src/pages/Dashboard/Health/Health.tsx
+++ b/airflow/ui/src/pages/Dashboard/Health/Health.tsx
@@ -54,15 +54,14 @@ export const Health = () => {
           status={data?.triggerer.status}
           title="Triggerer"
         />
-        {/* TODO: Update this to match the API when we move the config check to the API level */}
-        {data?.dag_processor.status === undefined ? undefined : (
+        {data && "dag_processor" in data ? (
           <HealthTag
             isLoading={isLoading}
-            latestHeartbeat={data.dag_processor.latest_dag_processor_heartbeat}
-            status={data.dag_processor.status}
+            latestHeartbeat={data.dag_processor?.latest_dag_processor_heartbeat}
+            status={data.dag_processor?.status}
             title="Dag Processor"
           />
-        )}
+        ) : undefined}
       </HStack>
     </Box>
   );
diff --git a/tests/api/common/test_airflow_health.py b/tests/api/common/test_airflow_health.py
index ebdc086c69277..b540b7e279801 100644
--- a/tests/api/common/test_airflow_health.py
+++ b/tests/api/common/test_airflow_health.py
@@ -33,8 +33,12 @@
 @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)
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=True)
 def test_get_airflow_health_only_metadatabase_healthy(
-    latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
 ):
     health_status = get_airflow_health()
     expected_status = {
@@ -50,8 +54,12 @@ def test_get_airflow_health_only_metadatabase_healthy(
 @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)
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=True)
 def test_get_airflow_health_metadatabase_unhealthy(
-    latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
 ):
     health_status = get_airflow_health()
 
@@ -65,6 +73,47 @@ def test_get_airflow_health_metadatabase_unhealthy(
     assert health_status == expected_status
 
 
+@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)
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=False)
+def test_get_airflow_health_only_metadatabase_healthy_with_dag_processor_disabled(
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
+):
+    health_status = get_airflow_health()
+    expected_status = {
+        "metadatabase": {"status": HEALTHY},
+        "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None},
+        "triggerer": {"status": None, "latest_triggerer_heartbeat": None},
+    }
+
+    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)
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=False)
+def test_get_airflow_health_metadatabase_unhealthy_with_dag_processor_disabled(
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
+):
+    health_status = get_airflow_health()
+
+    expected_status = {
+        "metadatabase": {"status": UNHEALTHY},
+        "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None},
+        "triggerer": {"status": UNHEALTHY, "latest_triggerer_heartbeat": None},
+    }
+
+    assert health_status == expected_status
+
+
 LATEST_SCHEDULER_JOB_MOCK = MagicMock()
 LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat = datetime.now()
 LATEST_SCHEDULER_JOB_MOCK.is_alive = MagicMock(return_value=True)
@@ -76,8 +125,12 @@ def test_get_airflow_health_metadatabase_unhealthy(
 )
 @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)
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=True)
 def test_get_airflow_health_scheduler_healthy_no_triggerer(
-    latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
 ):
     health_status = get_airflow_health()
 
@@ -94,6 +147,33 @@ def test_get_airflow_health_scheduler_healthy_no_triggerer(
     assert health_status == expected_status
 
 
+@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)
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=False)
+def test_get_airflow_health_scheduler_healthy_no_triggerer__with_dag_processor_disabled(
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
+):
+    health_status = get_airflow_health()
+
+    expected_status = {
+        "metadatabase": {"status": HEALTHY},
+        "scheduler": {
+            "status": HEALTHY,
+            "latest_scheduler_heartbeat": LATEST_SCHEDULER_JOB_MOCK.latest_heartbeat.isoformat(),
+        },
+        "triggerer": {"status": None, "latest_triggerer_heartbeat": None},
+    }
+
+    assert health_status == expected_status
+
+
 LATEST_TRIGGERER_JOB_MOCK = MagicMock()
 LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat = datetime.now()
 LATEST_TRIGGERER_JOB_MOCK.is_alive = MagicMock(return_value=True)
@@ -112,8 +192,12 @@ def test_get_airflow_health_scheduler_healthy_no_triggerer(
     "airflow.api.common.airflow_health.DagProcessorJobRunner.most_recent_job",
     return_value=LATEST_DAG_PROCESSOR_JOB_MOCK,
 )
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=True)
 def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record(
-    latest_scheduler_job_mock, latest_triggerer_job_mock, latest_dag_processor_job_mock
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
 ):
     health_status = get_airflow_health()
 
@@ -131,3 +215,33 @@ def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record(
     }
 
     assert health_status == expected_status
+
+
+@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,
+)
+@patch("airflow.api.common.airflow_health.conf.getboolean", return_value=False)
+def test_get_airflow_health_triggerer_healthy_no_scheduler_job_record_with_dag_processor_disabled(
+    latest_scheduler_job_mock,
+    latest_triggerer_job_mock,
+    latest_dag_processor_job_mock,
+    dag_processor_enabled_mock,
+):
+    health_status = get_airflow_health()
+
+    expected_status = {
+        "metadatabase": {"status": HEALTHY},
+        "scheduler": {"status": UNHEALTHY, "latest_scheduler_heartbeat": None},
+        "triggerer": {
+            "status": HEALTHY,
+            "latest_triggerer_heartbeat": LATEST_TRIGGERER_JOB_MOCK.latest_heartbeat.isoformat(),
+        },
+    }
+
+    assert health_status == expected_status
diff --git a/tests/api_fastapi/core_api/routes/public/test_monitor.py b/tests/api_fastapi/core_api/routes/public/test_monitor.py
index f4cdfb7a5511a..7fc0f15bcae39 100644
--- a/tests/api_fastapi/core_api/routes/public/test_monitor.py
+++ b/tests/api_fastapi/core_api/routes/public/test_monitor.py
@@ -103,3 +103,57 @@ def test_unhealthy_metadatabase_status(self, most_recent_job_mock, test_client):
 
         assert "unhealthy" == body["metadatabase"]["status"]
         assert body["scheduler"]["latest_scheduler_heartbeat"] is None
+
+
+class TestGetHealthDagProcessorEnabled(TestMonitorEndpoint):
+    @mock.patch("airflow.api_fastapi.core_api.routes.public.monitor.get_airflow_health")
+    def test_health_with_dag_processor(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",
+            },
+            "triggerer": {
+                "status": HEALTHY,
+                "latest_triggerer_heartbeat": "2024-11-23T11:09:15.815483+00:00",
+            },
+            "dag_processor": {
+                "status": HEALTHY,
+                "latest_dag_processor_heartbeat": "2024-11-23T11:09:15.815483+00:00",
+            },
+        }
+
+        response = test_client.get("/public/monitor/health")
+
+        assert response.status_code == 200
+        body = response.json()
+
+        assert "dag_processor" in body
+        assert body["metadatabase"]["status"] == HEALTHY
+        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_without_dag_processor(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",
+            },
+            "triggerer": {
+                "status": HEALTHY,
+                "latest_triggerer_heartbeat": "2024-11-23T11:09:15.815483+00:00",
+            },
+        }
+
+        response = test_client.get("/public/monitor/health")
+
+        assert response.status_code == 200
+        body = response.json()
+
+        assert "dag_processor" not in body
+        assert body["metadatabase"]["status"] == HEALTHY
+        assert body["scheduler"]["status"] == HEALTHY
+        assert body["triggerer"]["status"] == HEALTHY


Code review comments:
diff_hunk: @@ -27,4 +27,4 @@
 @monitor_router.get("/health")
 def get_health() -> HealthInfoSchema:
     airflow_health_status = get_airflow_health()
-    return HealthInfoSchema.model_validate(airflow_health_status)
+    return HealthInfoSchema.model_validate(airflow_health_status, from_attributes=True)
path: airflow/api_fastapi/core_api/routes/public/monitor.py
text: @user1:
```suggestion
    return airflow_health_status
```

@author:
I think we need to cast it to  `HealthInfoSchema`

@author:
mypy giving me below error

```
airflow/api_fastapi/core_api/routes/public/monitor.py:30: error: Incompatible
return value type (got "dict[str, Any]", expected "HealthInfoSchema")
[return-value]
        return airflow_health_status
               ^~~~~~~~~~~~~~~~~~~~~
Found 1 error in 1 file (checked 1193 source files)
```

@user2:
`from_attributes` should be removed.

Because it is now in the model configuration. This is done automatically.

diff_hunk: @@ -49,4 +51,8 @@ class HealthInfoSchema(BaseModel):
     metadatabase: BaseInfoSchema
     scheduler: SchedulerInfoSchema
     triggerer: TriggererInfoSchema
-    dag_processor: DagProcessorInfoSchema
+    dag_processor: DagProcessorInfoSchema | None = None
+
+    @model_serializer()
+    def serialize(self):
+        return {k: v for k, v in self if not (k == "dag_processor" and v is None)}
path: airflow/api_fastapi/core_api/datamodels/monitor.py
text: @user2:
@user1 , can we use this(model_serializer) to implement fields to return in list dag_runs?

@user1:
If we use `response_model_exclude_unset=True`, we won't need it anymore.

Also just a [field_serializer](https://docs.pydantic.dev/latest/api/functional_serializers/#pydantic.functional_serializers.field_serializer) on the `dag_processor` is most likely enough.

@user2:
Sorry. I meant, we did not include fields query param in list dag runs as we did not want to have model with dynamic fields.