You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Bowrna (via GitHub)" <gi...@apache.org> on 2023/09/18 15:24:50 UTC
[GitHub] [airflow] Bowrna opened a new pull request, #34457: heartbeat recovery message
Bowrna opened a new pull request, #34457:
URL: https://github.com/apache/airflow/pull/34457
related: #31810
closed: #31810
this covers the point C of the issue
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
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.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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1463345865
##########
airflow/jobs/job.py:
##########
@@ -213,9 +218,12 @@ def heartbeat(
session.merge(self)
self.latest_heartbeat = timezone.utcnow()
session.commit()
+ # After committing verify if heartbeat has recovered after crossing health check threshold
Review Comment:
I am resolving this conversation for now @Lee-W. if you have any specific test recommendation that fits this case, please let me know. I will add it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1464523088
##########
airflow/jobs/job.py:
##########
@@ -126,7 +127,19 @@ def executor(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ @cached_property
Review Comment:
@amoghrajesh this function is already covered in the unit test. the right function for adding the unit test would be heartbeat(). Let me check that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Lee-W commented on a diff in pull request #34457: heartbeat recovery message
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1329844912
##########
airflow/jobs/job.py:
##########
@@ -128,6 +129,16 @@ def heartrate(self):
# cases continue to use that value for back compat
return conf.getfloat("scheduler", "JOB_HEARTBEAT_SEC")
+ @cached_property
+ def health_check_threshold(self, grace_multiplier=2.1):
Review Comment:
```suggestion
def health_check_threshold(self, grace_multiplier: float = 2.1) -> float:
```
##########
airflow/jobs/job.py:
##########
@@ -213,9 +218,12 @@ def heartbeat(
session.merge(self)
self.latest_heartbeat = timezone.utcnow()
session.commit()
+ # After committing verify if heartbeat has recovered after crossing health check threshold
Review Comment:
Should we add a test case for this behavior?
##########
airflow/jobs/job.py:
##########
@@ -128,6 +129,16 @@ def heartrate(self):
# cases continue to use that value for back compat
return conf.getfloat("scheduler", "JOB_HEARTBEAT_SEC")
+ @cached_property
+ def health_check_threshold(self, grace_multiplier=2.1):
+ if self.job_type == "SchedulerJob":
+ health_check_threshold_value: int = conf.getint("scheduler", "scheduler_health_check_threshold")
+ elif self.job_type == "TriggererJob":
+ health_check_threshold_value: int = conf.getint("triggerer", "triggerer_health_check_threshold")
+ else:
+ health_check_threshold_value: int = self.heartrate * grace_multiplier
Review Comment:
Should these `int` be `float` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1547161284
##########
airflow/jobs/job.py:
##########
@@ -131,22 +146,20 @@ def executors(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ def is_alive(self) -> bool:
"""
Is this job currently alive.
We define alive as in a state of RUNNING, and having sent a heartbeat
within a multiple of the heartrate (default of 2.1)
-
- :param grace_multiplier: multiplier of heartrate to require heart beat
- within
"""
+ threshold_value = health_check_threshold(
+ self.job_type, self.heartrate, self.grace_multiplier
Review Comment:
@potiuk when you get time, could you share your views on the above point
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-2036583335
All green now :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-2011247217
Test is failing in the health api call testcases. let me debug the issue and fix it by EOD
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1481704412
##########
tests/jobs/test_base_job.py:
##########
@@ -205,15 +205,17 @@ def test_is_alive_scheduler(self, job_type):
assert job.is_alive() is False, "Completed jobs even with recent heartbeat should not be alive"
def test_heartbeat_failed(self):
- when = timezone.utcnow() - datetime.timedelta(seconds=60)
- mock_session = Mock(name="MockSession")
- mock_session.commit.side_effect = OperationalError("Force fail", {}, None)
- job = Job(heartrate=10, state=State.RUNNING)
- job.latest_heartbeat = when
-
- job.heartbeat(heartbeat_callback=lambda: None, session=mock_session)
-
- assert job.latest_heartbeat == when, "attribute not updated when heartbeat fails"
+ try:
+ when = timezone.utcnow() - datetime.timedelta(seconds=60)
+ mock_session = Mock(name="MockSession")
+ mock_session.commit.side_effect = OperationalError("Force fail", {}, None)
+ job = Job(heartrate=10, state=State.RUNNING)
+ job.latest_heartbeat = when
+ job.heartbeat(heartbeat_callback=lambda: None, session=mock_session)
+ assert job.latest_heartbeat == when, "attribute not updated when heartbeat fails"
+ assert job.heartbeat_failed
+ except OperationalError as excinfo:
Review Comment:
Let me get back to you on this point, before this week ends @potiuk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-2008732165
@potiuk i made the changes regarding the grace_multiplier. If I remember right, the grace_multiplier was allowed to be set as a param value in config side and I don't see that part of the code currently. am I missing something here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1518546357
##########
tests/jobs/test_base_job.py:
##########
@@ -205,15 +205,17 @@ def test_is_alive_scheduler(self, job_type):
assert job.is_alive() is False, "Completed jobs even with recent heartbeat should not be alive"
def test_heartbeat_failed(self):
- when = timezone.utcnow() - datetime.timedelta(seconds=60)
- mock_session = Mock(name="MockSession")
- mock_session.commit.side_effect = OperationalError("Force fail", {}, None)
- job = Job(heartrate=10, state=State.RUNNING)
- job.latest_heartbeat = when
-
- job.heartbeat(heartbeat_callback=lambda: None, session=mock_session)
-
- assert job.latest_heartbeat == when, "attribute not updated when heartbeat fails"
+ try:
+ when = timezone.utcnow() - datetime.timedelta(seconds=60)
+ mock_session = Mock(name="MockSession")
+ mock_session.commit.side_effect = OperationalError("Force fail", {}, None)
+ job = Job(heartrate=10, state=State.RUNNING)
+ job.latest_heartbeat = when
+ job.heartbeat(heartbeat_callback=lambda: None, session=mock_session)
+ assert job.latest_heartbeat == when, "attribute not updated when heartbeat fails"
+ assert job.heartbeat_failed
+ except OperationalError as excinfo:
Review Comment:
@potiuk I wrongly thought that OperationalError will be thrown to this code, without checking that that exception is handled and the right error messages and params are set. I have removed that part of the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-1993761467
@potiuk i will add the changes that you have suggested
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna closed pull request #34457: heartbeat recovery message
URL: https://github.com/apache/airflow/pull/34457
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1542691829
##########
airflow/jobs/job.py:
##########
@@ -131,22 +146,20 @@ def executors(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ def is_alive(self) -> bool:
"""
Is this job currently alive.
We define alive as in a state of RUNNING, and having sent a heartbeat
within a multiple of the heartrate (default of 2.1)
-
- :param grace_multiplier: multiplier of heartrate to require heart beat
- within
"""
+ threshold_value = health_check_threshold(
+ self.job_type, self.heartrate, self.grace_multiplier
Review Comment:
@potiuk Guess I was able to identify the issue. I have added the grace_multiplier in the __init__ part of the code
https://github.com/Bowrna/airflow/blob/ddc7d44fecf217f56d55260dc5c997ecd4179b0b/airflow/jobs/job.py#L119-L135
But when invoked from health check API, it invokes the most_recent_job function which returns the Job object collected from the DB. There is a possibility in code, that init function is not invoked at all until this point. But as grace_multiplier is inside init it is throwing no attribute error. But other params like job_type( available as param of the class) and heartrate( available as response of the staticmethod) are available. How do I handle this case? @potiuk. I have tried keeping grace_multiplier outside any conditions too but it failed with same error as above
https://github.com/Bowrna/airflow/blob/ddc7d44fecf217f56d55260dc5c997ecd4179b0b/airflow/jobs/job.py#L376-L395
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1549269048
##########
airflow/jobs/job.py:
##########
@@ -131,22 +146,20 @@ def executors(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ def is_alive(self) -> bool:
"""
Is this job currently alive.
We define alive as in a state of RUNNING, and having sent a heartbeat
within a multiple of the heartrate (default of 2.1)
-
- :param grace_multiplier: multiplier of heartrate to require heart beat
- within
"""
+ threshold_value = health_check_threshold(
+ self.job_type, self.heartrate, self.grace_multiplier
Review Comment:
ok got it. i will do that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-1924290629
@amoghrajesh @potiuk do I have any feedback on the test case I have added?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #34457:
URL: https://github.com/apache/airflow/pull/34457
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-2040762752
🎉
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1458792412
##########
airflow/jobs/job.py:
##########
@@ -126,7 +127,19 @@ def executor(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ @cached_property
+ def health_check_threshold(self) -> int | float:
+ grace_multiplier: float = 2.1
Review Comment:
got it.. i will make this change @potiuk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-1986905542
this looks good @Bowrna (except the static check failing) - but I have one small proposal. It took me a bit of time to see why grace_multiplier is not passed to to the method in heartbeat - and I found out that even if it is specified in `is_alive` method, it's never set to anything diffferent than 2.1. And I think it's placed wrrongly. it should not be parameter of is_alive, but it should be - similarly as `heartrate` for example) added as Job's field in Job's constructor.
```
self.grace_multiplier = 2.1
```
And then - it should be taken from job whenever needed (similarly as we take job.job_type).
It would be great to change it here - because we are anyhow touching the same code, but If you think it's too much, I am fine to approve it even without it (and you could do it as a follow-up for example).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1456873011
##########
airflow/jobs/job.py:
##########
@@ -126,7 +127,19 @@ def executor(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ @cached_property
Review Comment:
Will we be able to add a unit test for this?
##########
airflow/jobs/job.py:
##########
@@ -126,7 +127,19 @@ def executor(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ @cached_property
+ def health_check_threshold(self) -> int | float:
+ grace_multiplier: float = 2.1
Review Comment:
+1 agree with you here
##########
airflow/jobs/job.py:
##########
@@ -201,12 +215,15 @@ def heartbeat(
job = Job._update_heartbeat(job=self, session=session)
self._merge_from(job)
-
+ time_since_last_heartbeat = (timezone.utcnow() - previous_heartbeat).total_seconds()
+ if time_since_last_heartbeat > self.health_check_threshold:
+ self.log.info("INFO: Heartbeat recovered after %.2f seconds", time_since_last_heartbeat)
Review Comment:
We do not need to add the "INFO:". It will be taken care by the logger
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-1870302012
Do you happen to have any updates/suggestions regarding this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1536847700
##########
airflow/jobs/job.py:
##########
@@ -131,22 +146,20 @@ def executors(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ def is_alive(self) -> bool:
"""
Is this job currently alive.
We define alive as in a state of RUNNING, and having sent a heartbeat
within a multiple of the heartrate (default of 2.1)
-
- :param grace_multiplier: multiplier of heartrate to require heart beat
- within
"""
+ threshold_value = health_check_threshold(
+ self.job_type, self.heartrate, self.grace_multiplier
Review Comment:
hmm i tried other way too.. let me check it again @potiuk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-2009185801
Looks like real issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-2008743614
> @potiuk i made the changes regarding the grace_multiplier. If I remember right, the grace_multiplier was allowed to be set as a param value in config side and I don't see that part of the code currently. am I missing something here?
Yes. Grace_multiplier is not set anywhere to a different value than default 2.1 now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1548320952
##########
airflow/jobs/job.py:
##########
@@ -131,22 +146,20 @@ def executors(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ def is_alive(self) -> bool:
"""
Is this job currently alive.
We define alive as in a state of RUNNING, and having sent a heartbeat
within a multiple of the heartrate (default of 2.1)
-
- :param grace_multiplier: multiplier of heartrate to require heart beat
- within
"""
+ threshold_value = health_check_threshold(
+ self.job_type, self.heartrate, self.grace_multiplier
Review Comment:
Yeah. That's the problem when we try to add parameters to a DB models that are not DB fields.
I think it's best to just remove the grace multiplier from init param and leave 2.1 as default value in "health_check_threshold" method (and do not pass any multiplier there - leave the default one).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34457:
URL: https://github.com/apache/airflow/pull/34457#issuecomment-2008744388
(or so I saw from reviewing the code myself). It's still good to keep it though in case we have a case we want to override it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1533445501
##########
airflow/jobs/job.py:
##########
@@ -131,22 +146,20 @@ def executors(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ def is_alive(self) -> bool:
"""
Is this job currently alive.
We define alive as in a state of RUNNING, and having sent a heartbeat
within a multiple of the heartrate (default of 2.1)
-
- :param grace_multiplier: multiplier of heartrate to require heart beat
- within
"""
+ threshold_value = health_check_threshold(
+ self.job_type, self.heartrate, self.grace_multiplier
Review Comment:
@potiuk I found this line is the cause of the issue. It throws `Facing AttributeError : 'Job' object has no attribute 'grace_multiplier'
while I have added in the init function the grace_multiplier. am I missing anything here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1379620724
##########
airflow/jobs/job.py:
##########
@@ -213,9 +218,12 @@ def heartbeat(
session.merge(self)
self.latest_heartbeat = timezone.utcnow()
session.commit()
+ # After committing verify if heartbeat has recovered after crossing health check threshold
Review Comment:
@Lee-W do you have any recommendation on the test case that I can add? I have tried to read test cases in `test_base_job.py` file but I don't know what part of this changes could be added as test in the file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1457670440
##########
airflow/jobs/job.py:
##########
@@ -201,12 +215,15 @@ def heartbeat(
job = Job._update_heartbeat(job=self, session=session)
self._merge_from(job)
-
+ time_since_last_heartbeat = (timezone.utcnow() - previous_heartbeat).total_seconds()
+ if time_since_last_heartbeat > self.health_check_threshold:
+ self.log.info("INFO: Heartbeat recovered after %.2f seconds", time_since_last_heartbeat)
Review Comment:
i will fix this part @amoghrajesh
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] Bowrna commented on a diff in pull request #34457: heartbeat recovery message
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1332510135
##########
airflow/jobs/job.py:
##########
@@ -213,9 +218,12 @@ def heartbeat(
session.merge(self)
self.latest_heartbeat = timezone.utcnow()
session.commit()
+ # After committing verify if heartbeat has recovered after crossing health check threshold
Review Comment:
yes let me check it @Lee-W thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1468854809
##########
airflow/jobs/job.py:
##########
@@ -126,7 +127,19 @@ def executor(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ @cached_property
Review Comment:
@amoghrajesh i have added the unit test for heartbeat function.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1478932569
##########
tests/jobs/test_base_job.py:
##########
@@ -205,15 +205,17 @@ def test_is_alive_scheduler(self, job_type):
assert job.is_alive() is False, "Completed jobs even with recent heartbeat should not be alive"
def test_heartbeat_failed(self):
- when = timezone.utcnow() - datetime.timedelta(seconds=60)
- mock_session = Mock(name="MockSession")
- mock_session.commit.side_effect = OperationalError("Force fail", {}, None)
- job = Job(heartrate=10, state=State.RUNNING)
- job.latest_heartbeat = when
-
- job.heartbeat(heartbeat_callback=lambda: None, session=mock_session)
-
- assert job.latest_heartbeat == when, "attribute not updated when heartbeat fails"
+ try:
+ when = timezone.utcnow() - datetime.timedelta(seconds=60)
+ mock_session = Mock(name="MockSession")
+ mock_session.commit.side_effect = OperationalError("Force fail", {}, None)
+ job = Job(heartrate=10, state=State.RUNNING)
+ job.latest_heartbeat = when
+ job.heartbeat(heartbeat_callback=lambda: None, session=mock_session)
+ assert job.latest_heartbeat == when, "attribute not updated when heartbeat fails"
+ assert job.heartbeat_failed
+ except OperationalError as excinfo:
Review Comment:
Why OperationalError here? I believe it is handled internally (when I remove it , the test continues to work BTW.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1535863595
##########
airflow/jobs/job.py:
##########
@@ -131,22 +146,20 @@ def executors(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ def is_alive(self) -> bool:
"""
Is this job currently alive.
We define alive as in a state of RUNNING, and having sent a heartbeat
within a multiple of the heartrate (default of 2.1)
-
- :param grace_multiplier: multiplier of heartrate to require heart beat
- within
"""
+ threshold_value = health_check_threshold(
+ self.job_type, self.heartrate, self.grace_multiplier
Review Comment:
It's only set conditionally if I see right, so yes - that looks ike a bug
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1438567596
##########
airflow/jobs/job.py:
##########
@@ -126,7 +127,19 @@ def executor(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ @cached_property
+ def health_check_threshold(self) -> int | float:
+ grace_multiplier: float = 2.1
Review Comment:
I think it's not exactly the same. Here `grace_multiplier` is hard-coded 2.1 but it can be overwritten in the `is_alive` method. Also this property is now duplicated in two places (Job and Job Pydantic)
What should happen is - similarly to other ORM/Pydantic classes this property/method should be chaned into a `@lru_cache` decorated fundtion as the top level function in job (taking job_type, grace_multiplier and heartrate as parameter) and it should be called in both Job and Job Pydantic. Decorating it with `@lru_cache` will cache the config retrieval as appropriate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] heartbeat recovery message [airflow]
Posted by "Bowrna (via GitHub)" <gi...@apache.org>.
Bowrna commented on code in PR #34457:
URL: https://github.com/apache/airflow/pull/34457#discussion_r1463353839
##########
airflow/jobs/job.py:
##########
@@ -126,7 +127,19 @@ def executor(self):
def heartrate(self) -> float:
return Job._heartrate(self.job_type)
- def is_alive(self, grace_multiplier=2.1) -> bool:
+ @cached_property
Review Comment:
let me check this one @amoghrajesh
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org