You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "mobuchowski (via GitHub)" <gi...@apache.org> on 2024/05/28 09:11:04 UTC
[PR] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
mobuchowski opened a new pull request, #39890:
URL: https://github.com/apache/airflow/pull/39890
(no comment)
--
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] local task job: add overtime mechanism, to not kill task prematurely when auxiliary processes are running [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1636631128
##########
tests/listeners/very_slow_listener.py:
##########
@@ -0,0 +1,26 @@
+# 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.
+from __future__ import annotations
+
+import time
+
+from airflow.listeners import hookimpl
+
+
+@hookimpl
+def on_task_instance_success(previous_state, task_instance, session):
+ time.sleep(30)
Review Comment:
Tried to naively use `frozen_sleep` and I don't think it plays nice with heartbeat mechanism.
```
INFO airflow.jobs.local_task_job_runner.LocalTaskJobRunner:local_task_job_runner.py:122 ::group::Pre task execution logs
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 420.04 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 750.01 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 665.00 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 560.00 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 330.00 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 220.00 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 345.00 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 360.00 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 460.01 seconds
INFO airflow.jobs.job.Job:job.py:225 Heartbeat recovered after 2170.02 seconds
ERROR airflow.jobs.local_task_job_runner.LocalTaskJobRunner:local_task_job_runner.py:215 Heartbeat time limit exceeded!
INFO airflow.jobs.local_task_job_runner.LocalTaskJobRunner:local_task_job_runner.py:223 ::endgroup::
```
However, got it down to `51.44s`.
--
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] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1621078161
##########
airflow/jobs/local_task_job_runner.py:
##########
@@ -303,6 +306,14 @@ def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
if dagrun_timeout and execution_time > dagrun_timeout:
self.log.warning("DagRun timed out after %s.", execution_time)
+ # If process still runs after being marked as success, let it run until configured overtime
+ # if there are configured listeners
+ if (
+ ti.state == TaskInstanceState.SUCCESS
Review Comment:
In this case, it's actually ran before state is formally changed in DB, so that heartbeat mechanism does not apply.
https://github.com/apache/airflow/blob/7dc2b5243621f7a15ed1b44e1bf6eb81670f68d2/airflow/models/taskinstance.py#L3130
--
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] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1623281077
##########
airflow/jobs/local_task_job_runner.py:
##########
@@ -303,6 +306,14 @@ def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
if dagrun_timeout and execution_time > dagrun_timeout:
self.log.warning("DagRun timed out after %s.", execution_time)
+ # If process still runs after being marked as success, let it run until configured overtime
+ # if there are configured listeners
+ if (
+ ti.state == TaskInstanceState.SUCCESS
+ and self._overtime < conf.getint("core", "task_listener_timeout")
+ and get_listener_manager().has_listeners
Review Comment:
Do we want to limit it to only case where we have listeners? I think very similar scenario might happen when there is a "mini-scheduler" running after task completes. I **think** that case would also benefit from eager heartbeat killing such task. Maybe then we should turn it into a generic "task_timeout" not "task_listener_timeout" and kill two birds with the same stone?
--
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] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1626332859
##########
airflow/jobs/local_task_job_runner.py:
##########
@@ -303,6 +306,14 @@ def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
if dagrun_timeout and execution_time > dagrun_timeout:
self.log.warning("DagRun timed out after %s.", execution_time)
+ # If process still runs after being marked as success, let it run until configured overtime
+ # if there are configured listeners
+ if (
+ ti.state == TaskInstanceState.SUCCESS
+ and self._overtime < conf.getint("core", "task_listener_timeout")
+ and get_listener_manager().has_listeners
Review Comment:
Ok - I've just removed the check for listeners and renamed it as `task_success_overtime` - should cover that needs.
--
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] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1621883599
##########
airflow/jobs/local_task_job_runner.py:
##########
@@ -303,6 +306,14 @@ def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
if dagrun_timeout and execution_time > dagrun_timeout:
self.log.warning("DagRun timed out after %s.", execution_time)
+ # If process still runs after being marked as success, let it run until configured overtime
+ # if there are configured listeners
+ if (
+ ti.state == TaskInstanceState.SUCCESS
Review Comment:
Maybe it's worth mentioning it somehow in the yaml ? Right now we have:
> task_listener_timeout:
> description: |
> Maximum possible time (in seconds) that task listener will have for their execution.
I know config is not the place to explain how airflow works, but after reading just that i had the impression that this timeout will work for all listener task related methods, not only success 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] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1621078161
##########
airflow/jobs/local_task_job_runner.py:
##########
@@ -303,6 +306,14 @@ def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
if dagrun_timeout and execution_time > dagrun_timeout:
self.log.warning("DagRun timed out after %s.", execution_time)
+ # If process still runs after being marked as success, let it run until configured overtime
+ # if there are configured listeners
+ if (
+ ti.state == TaskInstanceState.SUCCESS
Review Comment:
In this case, it's actually ran before state is formally changed in DB:
https://github.com/apache/airflow/blob/7dc2b5243621f7a15ed1b44e1bf6eb81670f68d2/airflow/models/taskinstance.py#L3130
--
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] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1623281077
##########
airflow/jobs/local_task_job_runner.py:
##########
@@ -303,6 +306,14 @@ def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
if dagrun_timeout and execution_time > dagrun_timeout:
self.log.warning("DagRun timed out after %s.", execution_time)
+ # If process still runs after being marked as success, let it run until configured overtime
+ # if there are configured listeners
+ if (
+ ti.state == TaskInstanceState.SUCCESS
+ and self._overtime < conf.getint("core", "task_listener_timeout")
+ and get_listener_manager().has_listeners
Review Comment:
Do we want to limit it to only case where we have listeners? I think very similar scenario might happen when there is a "mini-scheduler" running after task completes. I **think** that case would also benefit from preventing eager heartbeat killing such task. Maybe then we should turn it into a generic "task_timeout" not "task_listener_timeout" and kill two birds with the same stone?
--
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] local task job: add overtime mechanism, to not kill task prematurely when auxiliary processes are running [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39890:
URL: https://github.com/apache/airflow/pull/39890#issuecomment-2161295635
Nice and simple. I think we should have another committer to take a look as well.
--
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] local task job: add overtime mechanism, to not kill task prematurely when auxiliary processes are running [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1635395008
##########
tests/listeners/very_slow_listener.py:
##########
@@ -0,0 +1,26 @@
+# 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.
+from __future__ import annotations
+
+import time
+
+from airflow.listeners import hookimpl
+
+
+@hookimpl
+def on_task_instance_success(previous_state, task_instance, session):
+ time.sleep(30)
Review Comment:
Have not run the unit tests but will this slow down unit testing by 30 seconds? Any way to mock the delay w/o a real sleep?
--
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] local task job: add timeout, to not kill on_task_instance_success listener by heartbeat_callback [airflow]
Posted by "josh-fell (via GitHub)" <gi...@apache.org>.
josh-fell commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1621003623
##########
airflow/jobs/local_task_job_runner.py:
##########
@@ -303,6 +306,14 @@ def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
if dagrun_timeout and execution_time > dagrun_timeout:
self.log.warning("DagRun timed out after %s.", execution_time)
+ # If process still runs after being marked as success, let it run until configured overtime
+ # if there are configured listeners
+ if (
+ ti.state == TaskInstanceState.SUCCESS
Review Comment:
A naïve question, but can the scenario you described occur if a task is marked as failed as well?
--
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] local task job: add overtime mechanism, to not kill task prematurely when auxiliary processes are running [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1636562750
##########
tests/listeners/very_slow_listener.py:
##########
@@ -0,0 +1,26 @@
+# 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.
+from __future__ import annotations
+
+import time
+
+from airflow.listeners import hookimpl
+
+
+@hookimpl
+def on_task_instance_success(previous_state, task_instance, session):
+ time.sleep(30)
Review Comment:
Maybe, we can do it with `time_machine` ? It's already used in a number of our tests and it's very easy to mock time using 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] local task job: add overtime mechanism, to not kill task prematurely when auxiliary processes are running [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #39890:
URL: https://github.com/apache/airflow/pull/39890#discussion_r1636526065
##########
tests/listeners/very_slow_listener.py:
##########
@@ -0,0 +1,26 @@
+# 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.
+from __future__ import annotations
+
+import time
+
+from airflow.listeners import hookimpl
+
+
+@hookimpl
+def on_task_instance_success(previous_state, task_instance, session):
+ time.sleep(30)
Review Comment:
@jscheffl not exactly - this is specifically too long to make the timeout mechanism reliably kill the process. However, since this is running real timeouts that measure time, _some_ seconds spend running those tests are reasonable.
I've modified the values a bit and now `pytest tests/jobs/test_local_task_job.py` take
`66.58s` compared to `46.27s` - I don't think that's too bad.
--
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] local task job: add overtime mechanism, to not kill task prematurely when auxiliary processes are running [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski merged PR #39890:
URL: https://github.com/apache/airflow/pull/39890
--
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