prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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)
  
  
  </issue>
  <code>
  [start of airflow/api_connexion/schemas/health_schema.py]
  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
  [end of airflow/api_connexion/schemas/health_schema.py]
  [start of airflow/api/common/airflow_health.py]
  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 of airflow/api/common/airflow_health.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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()
  
verification_info: {"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"}]}