You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ep...@apache.org on 2023/05/15 09:34:10 UTC
[airflow] branch v2-6-test updated (ddcfebdffe -> f97a484793)
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a change to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
from ddcfebdffe Update RELEASE_NOTES.rst
new 34b838f772 Fix calculation of health check threshold for SchedulerJob (#31277)
new f97a484793 Update RELEASE_NOTES.rst
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
RELEASE_NOTES.rst | 1 +
airflow/config_templates/config.yml | 3 ++-
airflow/config_templates/default_airflow.cfg | 3 ++-
airflow/jobs/job.py | 7 +++++--
airflow/jobs/scheduler_job_runner.py | 21 ---------------------
newsfragments/31277.significant.rst | 8 ++++++++
tests/jobs/test_base_job.py | 23 +++++++++++++++++++++++
7 files changed, 41 insertions(+), 25 deletions(-)
create mode 100644 newsfragments/31277.significant.rst
[airflow] 02/02: Update RELEASE_NOTES.rst
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit f97a484793ef0bf85bded735587e081513288bfa
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Mon May 15 10:33:37 2023 +0100
Update RELEASE_NOTES.rst
---
RELEASE_NOTES.rst | 1 +
1 file changed, 1 insertion(+)
diff --git a/RELEASE_NOTES.rst b/RELEASE_NOTES.rst
index edc3fd65a7..50397616eb 100644
--- a/RELEASE_NOTES.rst
+++ b/RELEASE_NOTES.rst
@@ -30,6 +30,7 @@ No significant changes.
Bug Fixes
^^^^^^^^^
+- Fix calculation of health check threshold for SchedulerJob (#31277)
- Fix timestamp parse failure for k8s executor pod tailing (#31175)
- Make sure that DAG processor job row has filled value in ``job_type`` column (#31182)
- Fix section name reference for ``api_client_retry_configuration`` (#31174)
[airflow] 01/02: Fix calculation of health check threshold for SchedulerJob (#31277)
Posted by ep...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 34b838f7727a2e41330788ccd71c06a975b05420
Author: Jarek Potiuk <ja...@potiuk.com>
AuthorDate: Mon May 15 10:31:14 2023 +0200
Fix calculation of health check threshold for SchedulerJob (#31277)
The change ##30302 split Job from JobRunner, but it missed the fact
that SchedulerJob had a special case of checking the threshold -
instead of using the standard grace multiplier, it used whatever
has been defined in the `scheduler_helth_check_threshold`. The
`is_alive` method in SchedulerJobRunner has remained unused, and
default 2.1 grace multiplier has been used for both /health
endpoint and `airflow jobs check`.
This PR brings the exception for SchedulerJob back and clarifies
that the same treshold is also used for airflow jobs check in
the documentation.
Fixes: #31200
(cherry picked from commit f366d955cd3be551c96ad7f794e0b8525900d13d)
---
airflow/config_templates/config.yml | 3 ++-
airflow/config_templates/default_airflow.cfg | 3 ++-
airflow/jobs/job.py | 7 +++++--
airflow/jobs/scheduler_job_runner.py | 21 ---------------------
newsfragments/31277.significant.rst | 8 ++++++++
tests/jobs/test_base_job.py | 23 +++++++++++++++++++++++
6 files changed, 40 insertions(+), 25 deletions(-)
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index e80b18cbe8..b54876f2ae 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -2212,7 +2212,8 @@ scheduler:
description: |
If the last scheduler heartbeat happened more than scheduler_health_check_threshold
ago (in seconds), scheduler is considered unhealthy.
- This is used by the health check in the "/health" endpoint
+ This is used by the health check in the "/health" endpoint and in `airflow jobs check` CLI
+ for SchedulerJob.
version_added: 1.10.2
type: string
example: ~
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index 684a78992d..dc4b8aa35c 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -1120,7 +1120,8 @@ pool_metrics_interval = 5.0
# If the last scheduler heartbeat happened more than scheduler_health_check_threshold
# ago (in seconds), scheduler is considered unhealthy.
-# This is used by the health check in the "/health" endpoint
+# This is used by the health check in the "/health" endpoint and in `airflow jobs check` CLI
+# for SchedulerJob.
scheduler_health_check_threshold = 30
# When you start a scheduler, airflow starts a tiny web server
diff --git a/airflow/jobs/job.py b/airflow/jobs/job.py
index 9867efa9ec..399fa39f25 100644
--- a/airflow/jobs/job.py
+++ b/airflow/jobs/job.py
@@ -130,10 +130,13 @@ class Job(Base, LoggingMixin):
:param grace_multiplier: multiplier of heartrate to require heart beat
within
"""
+ if self.job_type == "SchedulerJob":
+ health_check_threshold: int = conf.getint("scheduler", "scheduler_health_check_threshold")
+ else:
+ health_check_threshold: int = self.heartrate * grace_multiplier
return (
self.state == State.RUNNING
- and (timezone.utcnow() - self.latest_heartbeat).total_seconds()
- < self.heartrate * grace_multiplier
+ and (timezone.utcnow() - self.latest_heartbeat).total_seconds() < health_check_threshold
)
@provide_session
diff --git a/airflow/jobs/scheduler_job_runner.py b/airflow/jobs/scheduler_job_runner.py
index 15f9214063..50f8ab586c 100644
--- a/airflow/jobs/scheduler_job_runner.py
+++ b/airflow/jobs/scheduler_job_runner.py
@@ -264,27 +264,6 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
self.job.executor.debug_dump()
self.log.info("-" * 80)
- def is_alive(self, grace_multiplier: float | None = None) -> bool:
- """
- Whether the SchedulerJob is alive.
-
- We define alive as in a state of running and a heartbeat within the
- threshold defined in the ``scheduler_health_check_threshold`` config
- setting.
-
- ``grace_multiplier`` is accepted for compatibility with the parent class.
-
- """
- if grace_multiplier is not None:
- # Accept the same behaviour as superclass
- return self.job.is_alive(grace_multiplier=grace_multiplier)
- scheduler_health_check_threshold: int = conf.getint("scheduler", "scheduler_health_check_threshold")
- return (
- self.job.state == State.RUNNING
- and (timezone.utcnow() - self.job.latest_heartbeat).total_seconds()
- < scheduler_health_check_threshold
- )
-
def __get_concurrency_maps(self, states: Iterable[TaskInstanceState], session: Session) -> ConcurrencyMap:
"""
Get the concurrency maps.
diff --git a/newsfragments/31277.significant.rst b/newsfragments/31277.significant.rst
new file mode 100644
index 0000000000..d65329a6bc
--- /dev/null
+++ b/newsfragments/31277.significant.rst
@@ -0,0 +1,8 @@
+Clarifications of the external Health Check mechanism and using ``Job`` classes.
+
+In the past SchedulerJob and other ``*Job`` classes are known to have been used to perform
+external health checks for Airflow components. Those are, however, Airflow DB ORM related classes.
+The DB models and database structure of Airflow are considered as internal implementation detail, following
+`public interface <https://airflow.apache.org/docs/apache-airflow/stable/public-airflow-interface.html>`_).
+Therefore, they should not be used for external health checks. Instead, you should use the
+``airflow jobs check`` CLI command (introduced in Airflow 2.1) for that purpose.
diff --git a/tests/jobs/test_base_job.py b/tests/jobs/test_base_job.py
index fb415e31a3..c646979237 100644
--- a/tests/jobs/test_base_job.py
+++ b/tests/jobs/test_base_job.py
@@ -148,6 +148,29 @@ class TestJob:
job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
assert job.is_alive() is False, "Completed jobs even with recent heartbeat should not be alive"
+ def test_is_alive_scheduler(self):
+ job = Job(heartrate=10, state=State.RUNNING, job_type="SchedulerJob")
+ assert job.is_alive() is True
+
+ job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=20)
+ assert job.is_alive() is True
+
+ # default health-check grace period for scheduler job is not heartrate*2.1 but 30 seconds
+ job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=21)
+ assert job.is_alive() is True
+
+ job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=31)
+ assert job.is_alive() is False
+
+ # test because .seconds was used before instead of total_seconds
+ # internal repr of datetime is (days, seconds)
+ job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(days=1)
+ assert job.is_alive() is False
+
+ job.state = State.SUCCESS
+ job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
+ assert job.is_alive() is False, "Completed jobs even with recent heartbeat should not be alive"
+
@patch("airflow.jobs.job.create_session")
def test_heartbeat_failed(self, mock_create_session):
when = timezone.utcnow() - datetime.timedelta(seconds=60)