You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Lee-W (via GitHub)" <gi...@apache.org> on 2023/06/07 09:55:02 UTC
[GitHub] [airflow] Lee-W opened a new pull request, #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Lee-W opened a new pull request, #31757:
URL: https://github.com/apache/airflow/pull/31757
<!--
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/
-->
address
---
**^ 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
[GitHub] [airflow] Lee-W commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231928871
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,13 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ if timeout := trigger.task_instance.trigger_timeout:
+ timeout = timeout.replace(tzinfo=timezone.utc) if not timeout.tzinfo else timeout
+ if timeout < timezone.utcnow():
+ self.log.error("Trigger cancelled due to timeout")
Review Comment:
Yes, it logs the timeout error and the original `asyncio.CancelledError` message
--
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] syedahsn commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231525460
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
This looks really good! Do we want that log level to be INFO though? I think it would be more accurate to log that as an ERROR
--
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] dstandish commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1228503159
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,14 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout and trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
i think this needs to be moved inside the `except asyncio.CancelledError as err` block
--
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] uranusjr commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231746274
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,18 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.error("Trigger cancelled due to timeout")
+ raise
+
+ self.log.info("Trigger cancelled; message=%s", str(err))
Review Comment:
```suggestion
self.log.info("Trigger cancelled; message=%s", err)
```
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231735365
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
Sounds good. I was not sure what would be the best for this case. Thus, I kept it INFO, but let me update it 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
[GitHub] [airflow] o-nikolas commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1228440901
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
Logging the exception that way isn't super user friendly, but at least it'd be accurate. Can you get an example of what that'd look like and paste it 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
[GitHub] [airflow] Lee-W commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231925296
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,17 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
Review Comment:
Yep, I encounter the timezone issue after I change the fallback [here](https://github.com/apache/airflow/blob/6becb7031618867bc253aefc9e3e216629575d2d/airflow/jobs/scheduler_job_runner.py#L940) to 1.0 for testing.
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,17 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
Review Comment:
Yep, I encountered the timezone issue after I change the fallback [here](https://github.com/apache/airflow/blob/6becb7031618867bc253aefc9e3e216629575d2d/airflow/jobs/scheduler_job_runner.py#L940) to 1.0 for testing.
--
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] dstandish merged pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish merged PR #31757:
URL: https://github.com/apache/airflow/pull/31757
--
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] o-nikolas commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1222177108
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
The issue that this was trying to address was communicating to the user that there was a timeout. If I was a user and saw "Trigger cancelled" I would be confused because I had not cancelled anything and I would have to trace through the code base (if I'm able as a code-y user) to find out that it was actually a timeout.
--
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] jedcunningham commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231187333
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -586,3 +613,9 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+ listener = setup_queue_listener()
+ assert handler not in non_pytest_handlers(log.handlers)
+ qh = log.handlers[-1]
+ assert qh.__class__ == LocalQueueHandler
+ assert qh.queue == listener.queue
+ listener.stop()
Review Comment:
I suspect duplicating these lines wasn't intentional, yeah?
```suggestion
```
--
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] dstandish commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231890523
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,17 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
Review Comment:
curious why the the tzinfo line is necessary. it seems that it should already be UTC. are you just being defensive or did you encounter an issue?
--
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] phanikumv commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1227447750
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,9 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
Review Comment:
let's add a unit test for this.
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1227449243
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,9 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
Review Comment:
Sure!
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231763415
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,18 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.error("Trigger cancelled due to timeout")
+ raise
+
+ self.log.info("Trigger cancelled; message=%s", str(err))
Review Comment:
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
[GitHub] [airflow] dstandish commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231892590
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,17 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.error("Trigger cancelled due to timeout")
+
+ self.log.error("Trigger cancelled; message=%s", err)
+ raise
Review Comment:
this suggestion is a bit golfy, but hey
```suggestion
except asyncio.CancelledError as err:
if timeout := trigger.task_instance.trigger_timeout:
timeout = timeout.replace(tzinfo=timezone.utc) if not timeout.tzinfo else timeout
if timeout < timezone.utcnow():
self.log.error("Trigger cancelled due to timeout")
self.log.error("Trigger cancelled; message=%s", err)
raise
```
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1229456886
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,9 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
Review Comment:
Just added unit tests today. 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
[GitHub] [airflow] Lee-W commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1229223195
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
I've modified the code a bit and this is how it looks like when it comes to timeout
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1229488432
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,18 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
After a second thought, I think what you suggest makes more sense. Just updated it. 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
[GitHub] [airflow] Lee-W commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1225238078
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
At first, I believed it to be the case. However, upon further consideration, we may come across different scenarios as well. How about we log the exception itself? like
```python
except asyncio.CancelledError as err:
self.log.info("Trigger cancelled due to {str(err)}")
```
--
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] uranusjr commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1229459801
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,18 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
Should this re-raise the caught CancelledError instead of create a new 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
[GitHub] [airflow] Lee-W commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231728679
##########
tests/jobs/test_triggerer_job.py:
##########
@@ -586,3 +613,9 @@ def non_pytest_handlers(val):
assert qh.__class__ == LocalQueueHandler
assert qh.queue == listener.queue
listener.stop()
+ listener = setup_queue_listener()
+ assert handler not in non_pytest_handlers(log.handlers)
+ qh = log.handlers[-1]
+ assert qh.__class__ == LocalQueueHandler
+ assert qh.queue == listener.queue
+ listener.stop()
Review Comment:
Yep, thanks for pointing it out. I didn't notice 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
[GitHub] [airflow] syedahsn commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231926391
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,13 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ if timeout := trigger.task_instance.trigger_timeout:
+ timeout = timeout.replace(tzinfo=timezone.utc) if not timeout.tzinfo else timeout
+ if timeout < timezone.utcnow():
+ self.log.error("Trigger cancelled due to timeout")
Review Comment:
In the timeout case, does it log the error twice?
--
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] uranusjr commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1222505521
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
Also do we want to re-raise the error?
--
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] o-nikolas commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1224844883
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
But is it always true that the Cancelled exception means a timeout? Or are there cases where it's actually cancelled by the user or for some other reason?
--
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] dstandish commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1228513098
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,14 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout and trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
it's pretty easy to test you can use example_time_delta_sensor_async.py and temporarily set its defer timeout to 2 seconds or something.
--
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] dstandish commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1228503159
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,14 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout and trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
i think this needs to be moved inside the `except asyncio.CancelledError as err` block; i don't think it will be reached otherwise.
otherwise, looks good. good idea checking the time. it may not be the most "direct" way but it's a lot nicer from implementation perspective than trying to send the "why" information to the database (since the timeout decision is actually made by the scheduler) and read that in the triggerer.
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1223836324
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
Hi @o-nikolas , @uranusjr , thanks for your suggestions! I tried to extend the message as "Trigger cancelled due to timeout" and re-raise this error
--
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] dstandish commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1228510338
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,14 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout and trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
+ except asyncio.CancelledError as err:
+ self.log.info("Trigger cancelled due to %s", str(err))
Review Comment:
```suggestion
self.log.info("Trigger cancelled; message=%s", str(err))
```
--
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] uranusjr commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231746991
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,18 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
On my second though Iām now wondering if this should raise at all! If this `raise` is removed, the block trickles down to the `raise` below anyway (line 623); and that gives us one additional log about the exception itself.
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1229463728
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,18 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
The reason why I raise a new one is to add the message to the exception. But if the log in line 619 is enough, I think reraise makes more sense
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231910034
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,17 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.error("Trigger cancelled due to timeout")
+
+ self.log.error("Trigger cancelled; message=%s", err)
+ raise
Review Comment:
Thanks for your suggestion! It looks great!
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1229460957
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,14 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout and trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
Hi, thanks for the suggestions for the changes and how to test them. I've made the changes accordingly and tested it locally.
--
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 pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #31757:
URL: https://github.com/apache/airflow/pull/31757#issuecomment-1589172963
@dstandish I tried to dig into the code a bit and found out the error message mentioned in https://github.com/apache/airflow/pull/30853#discussion_r1187018026 is from [check_trigger_timeouts](https://github.com/apache/airflow/blob/2940b9fa55a6a72c60c2162e541631addec3d6b8/airflow/jobs/scheduler_job_runner.py#L1660). Thus, I tried to add a similar timeout check [here](https://github.com/apache/airflow/pull/31757/commits/1f44a7b676682a62b522d35b5f758a0fc778cbc3#diff-9cea7921268261a177e82c16fd5111f8d3252e3ca0267bdfb397c379c5d70857R613-R616). I've not yet had time to test it today but would like to see what you think about it. 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
[GitHub] [airflow] syedahsn commented on a diff in pull request #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231525460
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +608,8 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError:
+ self.log.info("Trigger cancelled")
Review Comment:
Do we want that log level to be INFO? I think it would be more accurate to log that as an ERROR
--
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 #31757: feat(jobs/triggerer_job_runner): add triggerer canceled log
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #31757:
URL: https://github.com/apache/airflow/pull/31757#discussion_r1231748977
##########
airflow/jobs/triggerer_job_runner.py:
##########
@@ -608,6 +609,18 @@ async def run_trigger(self, trigger_id, trigger):
self.log.info("Trigger %s fired: %s", self.triggers[trigger_id]["name"], event)
self.triggers[trigger_id]["events"] += 1
self.events.append((trigger_id, event))
+ except asyncio.CancelledError as err:
+ trigger_timeout = trigger.task_instance.trigger_timeout
+ if trigger_timeout:
+ if not trigger_timeout.tzinfo:
+ trigger_timeout = trigger_timeout.replace(tzinfo=timezone.utc)
+
+ if trigger_timeout < timezone.utcnow():
+ self.log.info("Trigger cancelled due to timeout")
+ raise asyncio.CancelledError("Trigger cancelled due to timeout")
Review Comment:
This (not sure whether we should add that additional log) might be why I thought of reraising š¤ But I don't have a strong opinion on it. And I noticed it might be better to change the log level from info to error for other canceled errors as well. Let me make these changes 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