You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/11/10 16:47:37 UTC
[GitHub] [airflow] hussein-awala opened a new pull request, #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
hussein-awala opened a new pull request, #27597:
URL: https://github.com/apache/airflow/pull/27597
We want to make use of `exponential_backoff` to increase wait times between pokes, but we need to control the maximum wait time as we do with tasks retries when we activate `retry_exponential_backoff` ,where we can set `max_retry_delay` to use a fixed delay after a certain number of retries.
--
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] hussein-awala commented on a diff in pull request #27597: Add max_wait for exponential_backoff in BaseSensor
Posted by GitBox <gi...@apache.org>.
hussein-awala commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019602217
##########
airflow/decorators/__init__.pyi:
##########
@@ -14,12 +14,12 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
Review Comment:
yes, I didn't change this line, but the hook `Run black (python formatter)` did
--
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 pull request #27597: Add max_wait for exponential_backoff in BaseSensor
Posted by GitBox <gi...@apache.org>.
jedcunningham commented on PR #27597:
URL: https://github.com/apache/airflow/pull/27597#issuecomment-1311114809
Thanks @hussein-awala! Congrats on your first commit 🎉
--
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 merged pull request #27597: Add max_wait for exponential_backoff in BaseSensor
Posted by GitBox <gi...@apache.org>.
jedcunningham merged PR #27597:
URL: https://github.com/apache/airflow/pull/27597
--
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] hussein-awala commented on a diff in pull request #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
hussein-awala commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019596600
##########
airflow/sensors/base.py:
##########
@@ -122,6 +135,7 @@ def __init__(
self.timeout = timeout
self.mode = mode
self.exponential_backoff = exponential_backoff
+ self.max_wait = max_wait if max_wait is None else coerce_timedelta(max_wait, key="max_retry_delay")
Review Comment:
while this method is used only for `max_wait`, I deleted the argument `key` and I renamed the method to `_coerce_max_wait`
--
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 #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019596769
##########
airflow/sensors/base.py:
##########
@@ -97,6 +97,9 @@ class BaseSensorOperator(BaseOperator, SkipMixin):
prevent too much load on the scheduler.
:param exponential_backoff: allow progressive longer waits between
pokes by using exponential backoff algorithm
+ :param max_wait: maximum wait interval between pokes, can be set as
+ ``timedelta`` or ``float`` seconds, which will be converted into
+ ``timedelta``.
Review Comment:
```suggestion
:param max_wait: maximum wait interval between pokes, can be ``timedelta`` or ``float`` seconds
```
--
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 #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019517527
##########
airflow/sensors/base.py:
##########
@@ -122,6 +135,7 @@ def __init__(
self.timeout = timeout
self.mode = mode
self.exponential_backoff = exponential_backoff
+ self.max_wait = max_wait if max_wait is None else coerce_timedelta(max_wait, key="max_retry_delay")
Review Comment:
i would put "if wait is none" logic in coerce timedelta so that ... if it is none, it will return none, wdyt?
##########
airflow/sensors/base.py:
##########
@@ -72,6 +73,14 @@ def __bool__(self) -> bool:
return self.is_done
+def coerce_timedelta(value: float | timedelta, *, key: str) -> timedelta:
+ if isinstance(value, timedelta):
+ return value
+ if isinstance(value, Number):
+ return timedelta(seconds=value)
+ raise ValueError(f"{key} should be timedelta object or a number of seconds")
Review Comment:
i think it's ok to fail here (though i'd add `if value is None: return None` at the top)
BUT, i think it would be best to help the user by failing fast i.e. check the type in operator init and raise accordingly (rather than waiting until code gets deployed and task runs)
##########
airflow/sensors/base.py:
##########
@@ -72,6 +73,14 @@ def __bool__(self) -> bool:
return self.is_done
+def coerce_timedelta(value: float | timedelta, *, key: str) -> timedelta:
+ if isinstance(value, timedelta):
+ return value
+ if isinstance(value, Number):
+ return timedelta(seconds=value)
+ raise ValueError(f"{key} should be timedelta object or a number of seconds")
Review Comment:
the other benefit is then... if we're failing fast... then we don't really need this `key` argument which, at call site, is a little confusing.
if you validate the param in `__init__` then you can immediately raise with a helpful message.
##########
airflow/sensors/base.py:
##########
@@ -122,6 +135,7 @@ def __init__(
self.timeout = timeout
self.mode = mode
self.exponential_backoff = exponential_backoff
+ self.max_wait = max_wait if max_wait is None else coerce_timedelta(max_wait, key="max_retry_delay")
Review Comment:
i would get rid of `key=...` it's confusing and not nec esp if you validate type in init
##########
airflow/sensors/base.py:
##########
@@ -72,6 +73,14 @@ def __bool__(self) -> bool:
return self.is_done
+def coerce_timedelta(value: float | timedelta, *, key: str) -> timedelta:
Review Comment:
let's make it private
##########
tests/sensors/test_base.py:
##########
@@ -509,6 +509,40 @@ def run_duration():
assert interval2 >= sensor.poke_interval
assert interval2 > interval1
+ def test_sensor_with_exponential_backoff_on_and_max_wait(self):
+
+ sensor = DummySensor(
+ task_id=SENSOR_OP,
+ return_value=None,
+ poke_interval=10,
+ timeout=60,
+ exponential_backoff=True,
+ max_wait=timedelta(seconds=30),
+ )
+
+ with patch("airflow.utils.timezone.utcnow") as mock_utctime:
+ mock_utctime.return_value = DEFAULT_DATE
+
+ started_at = timezone.utcnow() - timedelta(seconds=10)
+
+ def run_duration():
+ return (timezone.utcnow - started_at).total_seconds()
+
+ interval1 = sensor._get_next_poke_interval(started_at, run_duration, 1)
+ interval2 = sensor._get_next_poke_interval(started_at, run_duration, 2)
+ interval3 = sensor._get_next_poke_interval(started_at, run_duration, 3)
+ interval4 = sensor._get_next_poke_interval(started_at, run_duration, 4)
+
+ assert interval1 >= 0
+ assert interval1 <= sensor.poke_interval
+ assert interval2 >= sensor.poke_interval
+ assert interval2 > interval1
+ assert interval2 <= sensor.max_wait.total_seconds()
+ assert interval3 >= interval2
+ assert interval3 <= sensor.max_wait.total_seconds()
+ assert interval4 >= interval3
+ assert interval4 <= sensor.max_wait.total_seconds()
Review Comment:
```suggestion
for idx, expected in enumerate([2, 6, 13, 30, 30, 30, 30, 30]):
assert sensor._get_next_poke_interval(started_at, run_duration, idx) == expected
```
i don't _think_ this is fragile... but it's def more readable
--
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] boring-cyborg[bot] commented on pull request #27597: Add max_wait for exponential_backoff in BaseSensor
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on PR #27597:
URL: https://github.com/apache/airflow/pull/27597#issuecomment-1311114409
Awesome work, congrats on your first merged pull request!
--
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] hussein-awala commented on a diff in pull request #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
hussein-awala commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019513643
##########
airflow/decorators/__init__.pyi:
##########
@@ -444,6 +445,9 @@ class TaskDecoratorCollection:
prevent too much load on the scheduler.
:param exponential_backoff: allow progressive longer waits between
pokes by using exponential backoff algorithm
+ :param max_wait: maximum wait interval between pokes, can be set as
+ ``timedelta`` or ``float`` seconds, which will be converted into
Review Comment:
When `exponential_backoff` is False, the sensor pokes every `poke_interval` seconds, but when it's True, we [calculate](https://github.com/apache/airflow/pull/27597/files#diff-7486f32e385d7ad0376cccda08d80e54939aa901a24616d11fb1f5cba6af7f83L226) a new exponential interval based `poke_interval` and the try number. After some tries, if we have a big timeout, the new interval will be very big, so this new argument is the max wait interval for the sensor waits.
For example, if we have a sensor with `poke_interval=600` and `timeout=24*60*60`(24h)
- Without max_wait, the senor will poke at the each retry:
1. Try 1: 546s
2. Try 2: 773s
3. Try 3: 1875s
4. Try 4: 3986s
5. Try 5: 5912s
6. Try 6: 18896s
- With max_wait=2 hours, it will poke at the each retry:
1. Try 1: 546s
2. Try 2: 773s
3. Try 3: 1875s
4. Try 4: 3600s
5. Try 5: 3600s
6. Try 6: 3600s
It is the same logic used for the [TaskInstance retries](https://github.com/apache/airflow/blob/main/airflow/models/taskinstance.py#L1123-L1124).
--
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 #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019597472
##########
airflow/decorators/__init__.pyi:
##########
@@ -444,6 +445,9 @@ class TaskDecoratorCollection:
prevent too much load on the scheduler.
:param exponential_backoff: allow progressive longer waits between
pokes by using exponential backoff algorithm
+ :param max_wait: maximum wait interval between pokes, can be set as
+ ``timedelta`` or ``float`` seconds, which will be converted into
+ ``timedelta``.
Review Comment:
```suggestion
:param max_wait: maximum wait interval between pokes, can be ``timedelta`` or ``float`` seconds
```
##########
airflow/decorators/__init__.pyi:
##########
@@ -14,12 +14,12 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
Review Comment:
do you have pre-commit enabled? i would not expect a change to this line
--
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] hussein-awala commented on a diff in pull request #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
hussein-awala commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019593995
##########
airflow/sensors/base.py:
##########
@@ -122,6 +135,7 @@ def __init__(
self.timeout = timeout
self.mode = mode
self.exponential_backoff = exponential_backoff
+ self.max_wait = max_wait if max_wait is None else coerce_timedelta(max_wait, key="max_retry_delay")
Review Comment:
yes it would be cleaner
--
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] hussein-awala commented on a diff in pull request #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
hussein-awala commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019573143
##########
airflow/sensors/base.py:
##########
@@ -122,6 +135,7 @@ def __init__(
self.timeout = timeout
self.mode = mode
self.exponential_backoff = exponential_backoff
+ self.max_wait = max_wait if max_wait is None else coerce_timedelta(max_wait, key="max_retry_delay")
Review Comment:
I will try to move `coerce` code to `_validate_input_values`
--
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] hussein-awala commented on a diff in pull request #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
hussein-awala commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019513643
##########
airflow/decorators/__init__.pyi:
##########
@@ -444,6 +445,9 @@ class TaskDecoratorCollection:
prevent too much load on the scheduler.
:param exponential_backoff: allow progressive longer waits between
pokes by using exponential backoff algorithm
+ :param max_wait: maximum wait interval between pokes, can be set as
+ ``timedelta`` or ``float`` seconds, which will be converted into
Review Comment:
When `exponential_backoff` is False, the sensor pokes every `poke_interval` seconds, but when it's True, we [calculate](https://github.com/apache/airflow/pull/27597/files#diff-7486f32e385d7ad0376cccda08d80e54939aa901a24616d11fb1f5cba6af7f83L226) a new exponential interval based `poke_interval` and the try number. After some tries, if we have a big timeout, the new interval will be very big, so this new argument is the max wait interval for the sensor waits.
For example, if we have a sensor with `poke_interval=600` and `timeout=24*60*60`(24h)
- Without max_wait, the senor will poke at the each retry:
1. Try 1: 546s
2. Try 2: 773s
3. Try 3: 1875s
4. Try 4: 3986s
5. Try 5: 5912s
6. Try 6: 18896s
- With max_wait=1 hour, it will poke at the each retry:
1. Try 1: 546s
2. Try 2: 773s
3. Try 3: 1875s
4. Try 4: 3600s
5. Try 5: 3600s
6. Try 6: 3600s
It is the same logic used for the [TaskInstance retries](https://github.com/apache/airflow/blob/main/airflow/models/taskinstance.py#L1123-L1124).
--
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] eladkal commented on a diff in pull request #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
eladkal commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019473571
##########
airflow/decorators/__init__.pyi:
##########
@@ -444,6 +445,9 @@ class TaskDecoratorCollection:
prevent too much load on the scheduler.
:param exponential_backoff: allow progressive longer waits between
pokes by using exponential backoff algorithm
+ :param max_wait: maximum wait interval between pokes, can be set as
+ ``timedelta`` or ``float`` seconds, which will be converted into
Review Comment:
How is that diffrent from poke_interval?
--
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] hussein-awala commented on a diff in pull request #27597: add an argument to limit the wait interval between pokes when exponential_backoff is on
Posted by GitBox <gi...@apache.org>.
hussein-awala commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019573143
##########
airflow/sensors/base.py:
##########
@@ -122,6 +135,7 @@ def __init__(
self.timeout = timeout
self.mode = mode
self.exponential_backoff = exponential_backoff
+ self.max_wait = max_wait if max_wait is None else coerce_timedelta(max_wait, key="max_retry_delay")
Review Comment:
I will try to move `coerce` code to `_validate_input_values`
--
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 #27597: Add max_wait for exponential_backoff in BaseSensor
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019604511
##########
airflow/sensors/base.py:
##########
@@ -122,8 +124,17 @@ def __init__(
self.timeout = timeout
self.mode = mode
self.exponential_backoff = exponential_backoff
+ self.max_wait = self._coerce_max_wait(max_wait)
self._validate_input_values()
+ @staticmethod
+ def _coerce_max_wait(max_wait: float | timedelta | None) -> timedelta | None:
+ if max_wait is None or isinstance(max_wait, timedelta):
+ return max_wait
+ if isinstance(max_wait, (int, float)) and max_wait >= 0:
+ return timedelta(seconds=max_wait)
+ raise AirflowException("The max_wait must be timedelta object or a non-negative number")
Review Comment:
```suggestion
raise AirflowException("Operator arg `max_wait` must be timedelta object or a non-negative number")
```
--
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 #27597: Add max_wait for exponential_backoff in BaseSensor
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019603400
##########
airflow/decorators/__init__.pyi:
##########
@@ -14,12 +14,12 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-
Review Comment:
oh weird... well, lets see what tests say
--
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 #27597: Add max_wait for exponential_backoff in BaseSensor
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27597:
URL: https://github.com/apache/airflow/pull/27597#discussion_r1019607761
##########
airflow/decorators/__init__.pyi:
##########
@@ -444,6 +445,9 @@ class TaskDecoratorCollection:
prevent too much load on the scheduler.
:param exponential_backoff: allow progressive longer waits between
pokes by using exponential backoff algorithm
+ :param max_wait: maximum wait interval between pokes, can be set as
+ ``timedelta`` or ``float`` seconds, which will be converted into
+ ``timedelta``.
Review Comment:
```suggestion
:param max_wait: maximum wait interval between pokes, can be ``timedelta`` or ``float`` seconds
```
--
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