Timeout Airflow TI and Sensors when using deferable mode - #32990
Timeout Airflow TI and Sensors when using deferable mode#32990hussein-awala wants to merge 23 commits into
Conversation
boring-cyborg
Bot
added
area:core-operators
area:providers
area:Scheduler
hussein-awala
changed the title
|
You are assuming that the defer timeout is always |
|
I should add... I believe (or at least i have always assumed) that the overall task timeout has always already been enforced. but the trigger timeout is something different that you may optionally use if you want to put a timeout on the specific deferral. So if I understand correctly, you're simply removing the feature to optionally put a timeout on a deferral. Is that correct? Is it really necessary / desirable? |
There was a problem hiding this comment.
Why do we need to move this? The import is done locally anyway and from what I can tell it can live where it has been.
There was a problem hiding this comment.
Currently I duplicated it, but I think we should move it to TI (or base operator class), since it is useful in all the operators and not just in Python operators. WDYT?
There was a problem hiding this comment.
This entire block is redundant now.
There was a problem hiding this comment.
Oops, I will fix it
|
How would removing the |
If we want to apply the same logic used in the normal mode for the execution_timeout the answer will be yes.
I can remove the deprecation warning and keep the support of custom timeout for the trigger
You assumption is not completely correct; there was already a timeout enforced equals to execution_timeout, but:
|
You're right, it is not b/c, I will update it. |
| trigger_timeout: datetime.datetime | None = None, | ||
| trigger_timeout_reason: str | None = None, |
There was a problem hiding this comment.
this seems to me like breaking backcompat
There was a problem hiding this comment.
specifically, you are removing the timeout param.
why do we need to remove that?
why do we need to add trigger_ prefix? isn't it implied?
There was a problem hiding this comment.
I agree, I ignored the possibility that the user might use it in his operators or unit tests. I will replace trigger_timeout and trigger_timeout_reason by timeout and timeout_reason.
|
So, what was the case before this change? IIUC:
do i have that right? it makes me a bit uncomfortable to remove the feature to be able to set a specific deferral timeout on a given deferral. not that it's necessarily all that useful, but it's there. now as to your change, let me see if i understand the new behavior correctly.
do i have all of that correct? or if i got something wrong, please correct. thanks. |
yes, but when it's reached, we don't execute
Non, we take the min between deferral timeout and execution_timeout if they are both present
Non, when we deferred the task, we calculated timeout as: # Calculate timeout too if it was passed
if defer.timeout is not None:
self.trigger_timeout = timezone.utcnow() + defer.timeout
else:
self.trigger_timeout = None
# If an execution_timeout is set, set the timeout to the minimum of
# it and the trigger timeout
execution_timeout = self.task.execution_timeout
if execution_timeout:
if self.trigger_timeout:
self.trigger_timeout = min(self.start_date + execution_timeout, self.trigger_timeout)
else:
self.trigger_timeout = self.start_date + execution_timeoutAnd when we resumed the task, we calculated the timeout duration as: timeout_seconds = (
task_to_execute.execution_timeout - (timezone.utcnow() - self.start_date)
).total_seconds()The main issue in the operators (not sensors) was with how we handle this timeout, where it was considered as a normal failure in the trigger.
It's still possible, with the possibility to control the callback when this timeout is reached
yes but this was the case before
20 minutes not 30 (1h - 10m - 30m)
if it's the earliest timeout moment, it will be used, and the The most important thing is the sensor timeout; In sync mode, when this timeout is reached, we fail the task regardless the number of remaining attempts, and this is not the case in the current async mode. This PR fixes this bug. So in summary, this PR:
|
hussein-awala
removed
provider:google
d71f56a to
98a2166
Compare
There was a problem hiding this comment.
you are changing data type from timedelta to datetime? that would seem to break backcompat.
There was a problem hiding this comment.
why doesn't this make more sense on the trigger model?
There was a problem hiding this comment.
i see there's already a trigger_timeout column on TI.
does seem to make more sense on trigger. but not sure if we should break with what's there already. just, ti tends to be one of the larger tables, and it's already pretty wide.
There was a problem hiding this comment.
does seem to make more sense on trigger. but not sure if we should break with what's there already. just, ti tends to be one of the larger tables, and it's already pretty wide.
it depends on if we decide to handle the timeout in the task instance as I do now, or the trigger cleanup as you propose.
There was a problem hiding this comment.
would seem we should keep this?
There was a problem hiding this comment.
I totally agree, I'll update it
There was a problem hiding this comment.
it's good to document the params. not necessarily obvious what the param "timeout" means given that, apparently, it can mean different things
There was a problem hiding this comment.
perhaps instead of adding a column to task instance, sensors could use a subclass of BaseTrigger which would be handled differently, and for that class of trigger, the timeouts would always result in an immediate task fail with no retries.
|
It feels like we may be rolling too much into one PR here, too many different decisions. In part, it's a fix for sensors. In part it's a change to the interface for deferrables. In part it's ensuring on_kill is called when trigger is cancelled due to execution_timeout. Regarding on_kill -- are we sure that on_kill should be called? The trigger class has a cleanup method which is called when it is cancelled. Should we not rely on this method instead? Or perhaps it should be delegated to the trigger to determine whether on_kill should be called in this circumstance. I'm also not sure it is really necessary to change the interface for deferrables and add columns to task_instance in order to fix the issue with sensor retries. It seems possible to fix that with perhaps less invasive changes. But having all of these changes in one PR makes it feel like it's all or nothing, and hard to assess whether the changes are the right way to solve these different problems. As usual, I won't stand in the way if others disagree, but if it's not too burdensome, it might be easier to make these decisions if we split up the PRs. |
There was a problem hiding this comment.
do we really need to merge these timeouts in this way?
the execution_timeout is still applied to the trigger, it seems.
so, it seems that sensors could always use the sensor timeout value to determine the appropriate deferral timeout to use. and such could be handled differentially by creating a new trigger subclass. and let execution_timeout continue to be applied independenty.
it feels a bit wrong to muddle them together.
I don't see a big risk to to all of that in the same time, because this help to avoid changing the interface when we want to fix the other problems (in case we found that there is something we missed during the first PR). But yes this could be split in two PR, one for on_kill (which needs changing in the interface), and as second one for sensor issue.
IMHO making async and sync mode consistent is the best option, but that's of course debatable.
Not sure about deferrables interfaces, but for the new column it is possible, but too complicated, where we should rerun the methods to re-find the reason, I started with this solution before adding the column.
I'm open to split them, do you have any suggestion to how we should do that? |
This one also depends on how we decide to handle the timeout, if we decide to handle it in the triggerer (which I don't prefer). But for me, I imagine the TI tell the trigger: "hey, I need your to do X, and you have Y seconds to do that, after that I will not need you anymore", and the Trigger should not care about the reason, it should just do it's job, and stop after the timeout. WDYT? |
If it doesn't seem practical, don't worry about it. |
98a2166 to
8ccf600
Compare
|
Ok, I put together what is a bit of a simpler and less invasive approach here: #33718 Please take a look when you have a chance. Essentially, for sensors, if I have it right, we can just use the existing trigger timeout feature to monitor for the sensor timeout, and we can raise AirflowSensorTimeout when that happens. |
|
suppressed by #33718 |
4 participants
closes: #32638
This PR handles the trigger timeout and raise
AirflowTaskTimeoutorAirflowSensorTimeoutinstead of failing the TI withTaskDeferralError.It deprecates providing a custom timeout, where it calculate the trigger timeout based on the execution timeout and/or the sensor timeout.
To test the sensor timeout case, I used:
and here is the log:
And it was passed directly to
Failedalthough there were 3 more attempts (this is the expected behavior).And for the execution_timeout:
and here is the log:
And it was passed to
up_for_retry.^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named
{pr_number}.significant.rstor{issue_number}.significant.rst, in newsfragments.