We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add DagProcessor status to health endpoint.
### Description

Add DagProcessor status including latest heartbeat to health endpoint similar to Triggerer status added recently. Related PRs.

https://github.com/apache/airflow/pull/31529
https://github.com/apache/airflow/pull/27755

### Use case/motivation

It helps in dag processor monitoring 

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/api_connexion/schemas/health_schema.py`
Content:
```
1 # Licensed to the Apache Software Foundation (ASF) under one
2 # or more contributor license agreements.  See the NOTICE file
3 # distributed with this work for additional information
4 # regarding copyright ownership.  The ASF licenses this file
5 # to you under the Apache License, Version 2.0 (the
6 # "License"); you may not use this file except in compliance
7 # with the License.  You may obtain a copy of the License at
8 #
9 #   http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 from __future__ import annotations
18 
19 from marshmallow import Schema, fields
20 
21 
22 class BaseInfoSchema(Schema):
23     """Base status field for metadatabase and scheduler."""
24 
25     status = fields.String(dump_only=True)
26 
27 
28 class MetaDatabaseInfoSchema(BaseInfoSchema):
29     """Schema for Metadatabase info."""
30 
31 
32 class SchedulerInfoSchema(BaseInfoSchema):
33     """Schema for Scheduler info."""
34 
35     latest_scheduler_heartbeat = fields.String(dump_only=True)
36 
37 
38 class TriggererInfoSchema(BaseInfoSchema):
39     """Schema for Triggerer info."""
40 
41     latest_triggerer_heartbeat = fields.String(dump_only=True)
42 
43 
44 class HealthInfoSchema(Schema):
45     """Schema for the Health endpoint."""
46 
47     metadatabase = fields.Nested(MetaDatabaseInfoSchema)
48     scheduler = fields.Nested(SchedulerInfoSchema)
49     triggerer = fields.Nested(TriggererInfoSchema)
50 
51 
52 health_schema = HealthInfoSchema()
53
```

Path: `airflow/api/common/airflow_health.py`
Content:
```
1 # Licensed to the Apache Software Foundation (ASF) under one
2 # or more contributor license agreements.  See the NOTICE file
3 # distributed with this work for additional information
4 # regarding copyright ownership.  The ASF licenses this file
5 # to you under the Apache License, Version 2.0 (the
6 # "License"); you may not use this file except in compliance
7 # with the License.  You may obtain a copy of the License at
8 #
9 #   http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing,
12 # software distributed under the License is distributed on an
13 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 # KIND, either express or implied.  See the License for the
15 # specific language governing permissions and limitations
16 # under the License.
17 from __future__ import annotations
18 
19 from typing import Any
20 
21 from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
22 from airflow.jobs.triggerer_job_runner import TriggererJobRunner
23 
24 HEALTHY = "healthy"
25 UNHEALTHY = "unhealthy"
26 
27 
28 def get_airflow_health() -> dict[str, Any]:
29     """Get the health for Airflow metadatabase, scheduler and triggerer."""
30     metadatabase_status = HEALTHY
31     latest_scheduler_heartbeat = None
32     latest_triggerer_heartbeat = None
33     scheduler_status = UNHEALTHY
34     triggerer_status: str | None = UNHEALTHY
35 
36     try:
37         latest_scheduler_job = SchedulerJobRunner.most_recent_job()
38 
39         if latest_scheduler_job:
40             latest_scheduler_heartbeat = latest_scheduler_job.latest_heartbeat.isoformat()
41             if latest_scheduler_job.is_alive():
42                 scheduler_status = HEALTHY
43     except Exception:
44         metadatabase_status = UNHEALTHY
45 
46     try:
47         latest_triggerer_job = TriggererJobRunner.most_recent_job()
48 
49         if latest_triggerer_job:
50             latest_triggerer_heartbeat = latest_triggerer_job.latest_heartbeat.isoformat()
51             if latest_triggerer_job.is_alive():
52                 triggerer_status = HEALTHY
53         else:
54             triggerer_status = None
55     except Exception:
56         metadatabase_status = UNHEALTHY
57 
58     airflow_health_status = {
59         "metadatabase": {"status": metadatabase_status},
60         "scheduler": {
61             "status": scheduler_status,
62             "latest_scheduler_heartbeat": latest_scheduler_heartbeat,
63         },
64         "triggerer": {
65             "status": triggerer_status,
66             "latest_triggerer_heartbeat": latest_triggerer_heartbeat,
67         },
68     }
69 
70     return airflow_health_status
71
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/api/common/airflow_health.py b/airflow/api/common/airflow_health.py
--- 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 @@
     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 @@
     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 @@
             "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/schemas/health_schema.py b/airflow/api_connexion/schemas/health_schema.py
--- a/airflow/api_connexion/schemas/health_schema.py
+++ b/airflow/api_connexion/schemas/health_schema.py
@@ -41,12 +41,19 @@
     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()

{"golden_diff": "diff --git a/airflow/api/common/airflow_health.py b/airflow/api/common/airflow_health.py\n--- a/airflow/api/common/airflow_health.py\n+++ b/airflow/api/common/airflow_health.py\n@@ -18,6 +18,7 @@\n \n from typing import Any\n \n+from airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner\n from airflow.jobs.scheduler_job_runner import SchedulerJobRunner\n from airflow.jobs.triggerer_job_runner import TriggererJobRunner\n \n@@ -30,8 +31,10 @@\n     metadatabase_status = HEALTHY\n     latest_scheduler_heartbeat = None\n     latest_triggerer_heartbeat = None\n+    latest_dag_processor_heartbeat = None\n     scheduler_status = UNHEALTHY\n     triggerer_status: str | None = UNHEALTHY\n+    dag_processor_status: str | None = UNHEALTHY\n \n     try:\n         latest_scheduler_job = SchedulerJobRunner.most_recent_job()\n@@ -55,6 +58,18 @@\n     except Exception:\n         metadatabase_status = UNHEALTHY\n \n+    try:\n+        latest_dag_processor_job = DagProcessorJobRunner.most_recent_job()\n+\n+        if latest_dag_processor_job:\n+            latest_dag_processor_heartbeat = latest_dag_processor_job.latest_heartbeat.isoformat()\n+            if latest_dag_processor_job.is_alive():\n+                dag_processor_status = HEALTHY\n+        else:\n+            dag_processor_status = None\n+    except Exception:\n+        metadatabase_status = UNHEALTHY\n+\n     airflow_health_status = {\n         \"metadatabase\": {\"status\": metadatabase_status},\n         \"scheduler\": {\n@@ -65,6 +80,10 @@\n             \"status\": triggerer_status,\n             \"latest_triggerer_heartbeat\": latest_triggerer_heartbeat,\n         },\n+        \"dag_processor\": {\n+            \"status\": dag_processor_status,\n+            \"latest_dag_processor_heartbeat\": latest_dag_processor_heartbeat,\n+        },\n     }\n \n     return airflow_health_status\ndiff --git a/airflow/api_connexion/schemas/health_schema.py b/airflow/api_connexion/schemas/health_schema.py\n--- a/airflow/api_connexion/schemas/health_schema.py\n+++ b/airflow/api_connexion/schemas/health_schema.py\n@@ -41,12 +41,19 @@\n     latest_triggerer_heartbeat = fields.String(dump_only=True)\n \n \n+class DagProcessorInfoSchema(BaseInfoSchema):\n+    \"\"\"Schema for DagProcessor info.\"\"\"\n+\n+    latest_dag_processor_heartbeat = fields.String(dump_only=True)\n+\n+\n class HealthInfoSchema(Schema):\n     \"\"\"Schema for the Health endpoint.\"\"\"\n \n     metadatabase = fields.Nested(MetaDatabaseInfoSchema)\n     scheduler = fields.Nested(SchedulerInfoSchema)\n     triggerer = fields.Nested(TriggererInfoSchema)\n+    dag_processor = fields.Nested(DagProcessorInfoSchema)\n \n \n health_schema = HealthInfoSchema()\n", "issue": "Add DagProcessor status to health endpoint.\n### Description\n\nAdd DagProcessor status including latest heartbeat to health endpoint similar to Triggerer status added recently. Related PRs.\r\n\r\nhttps://github.com/apache/airflow/pull/31529\r\nhttps://github.com/apache/airflow/pull/27755\n\n### Use case/motivation\n\nIt helps in dag processor monitoring \n\n### Related issues\n\n_No response_\n\n### Are you willing to submit a PR?\n\n- [X] Yes I am willing to submit a PR!\n\n### Code of Conduct\n\n- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)\n\n", "before_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nfrom __future__ import annotations\n\nfrom marshmallow import Schema, fields\n\n\nclass BaseInfoSchema(Schema):\n    \"\"\"Base status field for metadatabase and scheduler.\"\"\"\n\n    status = fields.String(dump_only=True)\n\n\nclass MetaDatabaseInfoSchema(BaseInfoSchema):\n    \"\"\"Schema for Metadatabase info.\"\"\"\n\n\nclass SchedulerInfoSchema(BaseInfoSchema):\n    \"\"\"Schema for Scheduler info.\"\"\"\n\n    latest_scheduler_heartbeat = fields.String(dump_only=True)\n\n\nclass TriggererInfoSchema(BaseInfoSchema):\n    \"\"\"Schema for Triggerer info.\"\"\"\n\n    latest_triggerer_heartbeat = fields.String(dump_only=True)\n\n\nclass HealthInfoSchema(Schema):\n    \"\"\"Schema for the Health endpoint.\"\"\"\n\n    metadatabase = fields.Nested(MetaDatabaseInfoSchema)\n    scheduler = fields.Nested(SchedulerInfoSchema)\n    triggerer = fields.Nested(TriggererInfoSchema)\n\n\nhealth_schema = HealthInfoSchema()\n", "path": "airflow/api_connexion/schemas/health_schema.py"}, {"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nfrom __future__ import annotations\n\nfrom typing import Any\n\nfrom airflow.jobs.scheduler_job_runner import SchedulerJobRunner\nfrom airflow.jobs.triggerer_job_runner import TriggererJobRunner\n\nHEALTHY = \"healthy\"\nUNHEALTHY = \"unhealthy\"\n\n\ndef get_airflow_health() -> dict[str, Any]:\n    \"\"\"Get the health for Airflow metadatabase, scheduler and triggerer.\"\"\"\n    metadatabase_status = HEALTHY\n    latest_scheduler_heartbeat = None\n    latest_triggerer_heartbeat = None\n    scheduler_status = UNHEALTHY\n    triggerer_status: str | None = UNHEALTHY\n\n    try:\n        latest_scheduler_job = SchedulerJobRunner.most_recent_job()\n\n        if latest_scheduler_job:\n            latest_scheduler_heartbeat = latest_scheduler_job.latest_heartbeat.isoformat()\n            if latest_scheduler_job.is_alive():\n                scheduler_status = HEALTHY\n    except Exception:\n        metadatabase_status = UNHEALTHY\n\n    try:\n        latest_triggerer_job = TriggererJobRunner.most_recent_job()\n\n        if latest_triggerer_job:\n            latest_triggerer_heartbeat = latest_triggerer_job.latest_heartbeat.isoformat()\n            if latest_triggerer_job.is_alive():\n                triggerer_status = HEALTHY\n        else:\n            triggerer_status = None\n    except Exception:\n        metadatabase_status = UNHEALTHY\n\n    airflow_health_status = {\n        \"metadatabase\": {\"status\": metadatabase_status},\n        \"scheduler\": {\n            \"status\": scheduler_status,\n            \"latest_scheduler_heartbeat\": latest_scheduler_heartbeat,\n        },\n        \"triggerer\": {\n            \"status\": triggerer_status,\n            \"latest_triggerer_heartbeat\": latest_triggerer_heartbeat,\n        },\n    }\n\n    return airflow_health_status\n", "path": "airflow/api/common/airflow_health.py"}], "after_files": [{"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nfrom __future__ import annotations\n\nfrom marshmallow import Schema, fields\n\n\nclass BaseInfoSchema(Schema):\n    \"\"\"Base status field for metadatabase and scheduler.\"\"\"\n\n    status = fields.String(dump_only=True)\n\n\nclass MetaDatabaseInfoSchema(BaseInfoSchema):\n    \"\"\"Schema for Metadatabase info.\"\"\"\n\n\nclass SchedulerInfoSchema(BaseInfoSchema):\n    \"\"\"Schema for Scheduler info.\"\"\"\n\n    latest_scheduler_heartbeat = fields.String(dump_only=True)\n\n\nclass TriggererInfoSchema(BaseInfoSchema):\n    \"\"\"Schema for Triggerer info.\"\"\"\n\n    latest_triggerer_heartbeat = fields.String(dump_only=True)\n\n\nclass DagProcessorInfoSchema(BaseInfoSchema):\n    \"\"\"Schema for DagProcessor info.\"\"\"\n\n    latest_dag_processor_heartbeat = fields.String(dump_only=True)\n\n\nclass HealthInfoSchema(Schema):\n    \"\"\"Schema for the Health endpoint.\"\"\"\n\n    metadatabase = fields.Nested(MetaDatabaseInfoSchema)\n    scheduler = fields.Nested(SchedulerInfoSchema)\n    triggerer = fields.Nested(TriggererInfoSchema)\n    dag_processor = fields.Nested(DagProcessorInfoSchema)\n\n\nhealth_schema = HealthInfoSchema()\n", "path": "airflow/api_connexion/schemas/health_schema.py"}, {"content": "# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nfrom __future__ import annotations\n\nfrom typing import Any\n\nfrom airflow.jobs.dag_processor_job_runner import DagProcessorJobRunner\nfrom airflow.jobs.scheduler_job_runner import SchedulerJobRunner\nfrom airflow.jobs.triggerer_job_runner import TriggererJobRunner\n\nHEALTHY = \"healthy\"\nUNHEALTHY = \"unhealthy\"\n\n\ndef get_airflow_health() -> dict[str, Any]:\n    \"\"\"Get the health for Airflow metadatabase, scheduler and triggerer.\"\"\"\n    metadatabase_status = HEALTHY\n    latest_scheduler_heartbeat = None\n    latest_triggerer_heartbeat = None\n    latest_dag_processor_heartbeat = None\n    scheduler_status = UNHEALTHY\n    triggerer_status: str | None = UNHEALTHY\n    dag_processor_status: str | None = UNHEALTHY\n\n    try:\n        latest_scheduler_job = SchedulerJobRunner.most_recent_job()\n\n        if latest_scheduler_job:\n            latest_scheduler_heartbeat = latest_scheduler_job.latest_heartbeat.isoformat()\n            if latest_scheduler_job.is_alive():\n                scheduler_status = HEALTHY\n    except Exception:\n        metadatabase_status = UNHEALTHY\n\n    try:\n        latest_triggerer_job = TriggererJobRunner.most_recent_job()\n\n        if latest_triggerer_job:\n            latest_triggerer_heartbeat = latest_triggerer_job.latest_heartbeat.isoformat()\n            if latest_triggerer_job.is_alive():\n                triggerer_status = HEALTHY\n        else:\n            triggerer_status = None\n    except Exception:\n        metadatabase_status = UNHEALTHY\n\n    try:\n        latest_dag_processor_job = DagProcessorJobRunner.most_recent_job()\n\n        if latest_dag_processor_job:\n            latest_dag_processor_heartbeat = latest_dag_processor_job.latest_heartbeat.isoformat()\n            if latest_dag_processor_job.is_alive():\n                dag_processor_status = HEALTHY\n        else:\n            dag_processor_status = None\n    except Exception:\n        metadatabase_status = UNHEALTHY\n\n    airflow_health_status = {\n        \"metadatabase\": {\"status\": metadatabase_status},\n        \"scheduler\": {\n            \"status\": scheduler_status,\n            \"latest_scheduler_heartbeat\": latest_scheduler_heartbeat,\n        },\n        \"triggerer\": {\n            \"status\": triggerer_status,\n            \"latest_triggerer_heartbeat\": latest_triggerer_heartbeat,\n        },\n        \"dag_processor\": {\n            \"status\": dag_processor_status,\n            \"latest_dag_processor_heartbeat\": latest_dag_processor_heartbeat,\n        },\n    }\n\n    return airflow_health_status\n", "path": "airflow/api/common/airflow_health.py"}]}