FQ Runner FrameWork

4 files changed, 633 insertions, 0 deletions

diff --git a/library/python/retry/__init__.py b/library/python/retry/__init__.py
new file mode 100644
index 0000000000..5520a70332
--- /dev/null
+++ b/library/python/retry/__init__.py
@@ -0,0 +1,250 @@
+import copy
+import datetime
+import functools
+import itertools
+import logging
+import random
+import time
+
+
+"""
+Retry library provides an ability to retry function calls in a configurable way.
+
+To retry a certain function call use `retry_call` function. To make function auto-retriable use `retry`
+or `retry_intrusive` decorators. Both `retry_call` and `retry` optionally accept retry configuration object
+or its fields as kwargs. The `retry_intrusive` is designed for methods and uses intrusive configuration object.
+
+>>> retry_call(foo)
+>>> retry_call(foo, foo_args, foo_kwargs)
+>>> retry_call(foo, foo_args, foo_kwargs, conf=conf)
+
+>>> @retry()
+>>> def foo(...):
+>>>     ...
+
+>>> @retry(conf)
+>>> def foo(...):
+>>>     ...
+
+>>> class Obj(object):
+>>>     def __init__(self):
+>>>         self.retry_conf = conf
+>>>
+>>>     @retry_intrusive
+>>>     def foo(self, ...):
+>>>         ...
+
+This library differs from its alternatives:
+    * `retry` contrib library lacks time-based limits, reusable configuration objects and is generally less flexible
+    * `retrying` contrib library is somewhat more complex, but also lacks reusable configuration objects
+"""
+
+
+DEFAULT_SLEEP_FUNC = time.sleep
+LOGGER = logging.getLogger(__name__)
+
+
+class RetryConf(object):
+    """
+    Configuration object defines retry behaviour and is composed of these fields:
+        * `retriable` - function that decides if an exception should trigger retrying
+        * `get_delay` - function that returns a number of seconds retrier must wait before doing the next attempt
+        * `max_time` - maximum `datetime.timedelta` that can pass after the first call for any retry attempt to be done
+        * `max_times` - maximum number of retry attempts (note retries, not tries/calls)
+        * `handle_error` - function that is called for each failed call attempt
+        * `logger` - logger object to record retry warnings with
+        * `sleep` - custom sleep function to use for waiting
+
+    >>> RetryConf(max_time=datetime.timedelta(seconds=30), max_times=10)
+
+    Empty configuration retries indefinitely on any exceptions raised.
+
+    By default `DEFAULT_CONF` if used, which retries indefinitely, waiting 1 sec with 1.2 backoff between attempts, and
+    also logging with built-in logger object.
+
+    Configuration must be cloned before modification to create separate configuration:
+
+    >>> DEFAULT_CONF.clone()
+
+    There are various methods that provide convenient clone-and-modify shortcuts and "retry recipes".
+    """
+
+    _PROPS = {
+        "retriable": lambda e: True,
+        "get_delay": lambda n, raised_after, last: 0,
+        "max_time": None,
+        "max_times": None,
+        "handle_error": None,
+        "logger": None,
+        "sleep": DEFAULT_SLEEP_FUNC,
+    }
+
+    def __init__(self, **kwargs):
+        for prop, default_value in self._PROPS.items():
+            setattr(self, prop, default_value)
+        self._set(**kwargs)
+
+    def __repr__(self):
+        return repr(self.__dict__)
+
+    def clone(self, **kwargs):
+        """
+        Clone configuration.
+        """
+
+        obj = copy.copy(self)
+        obj._set(**kwargs)
+        return obj
+
+    def on(self, *errors):
+        """
+        Clone and retry on specific exception types (retriable shortcut):
+
+        >>> conf = conf.on(MyException, MyOtherException)
+        """
+
+        obj = self.clone()
+        obj.retriable = lambda e: isinstance(e, errors)
+        return obj
+
+    def waiting(self, delay=0, backoff=1.0, jitter=0, limit=None):
+        """
+        Clone and wait between attempts with backoff, jitter and limit (get_delay shortcut):
+
+        >>> conf = conf.waiting(delay)
+        >>> conf = conf.waiting(delay, backoff=2.0)  # initial delay with backoff x2 on each attempt
+        >>> conf = conf.waiting(delay, jitter=3)  # jitter from 0 to 3 seconds
+        >>> conf = conf.waiting(delay, backoff=2.0, limit=60)  # delay with backoff, but not greater than a minute
+
+        All these options can be combined together, of course.
+        """
+
+        def get_delay(n, raised_after, last):
+            if n == 1:
+                return delay
+
+            s = last * backoff
+            s += random.uniform(0, jitter)
+            if limit is not None:
+                s = min(s, limit)
+            return s
+
+        obj = self.clone()
+        obj.get_delay = get_delay
+        return obj
+
+    def upto(self, seconds=0, **other_timedelta_kwargs):
+        """
+        Clone and do retry attempts only for some time (max_time shortcut):
+
+        >>> conf = conf.upto(30)  # retrying for 30 seconds
+        >>> conf = conf.upto(hours=1, minutes=20)  # retrying for 1:20
+
+        Any `datetime.timedelta` kwargs can be used here.
+        """
+
+        obj = self.clone()
+        obj.max_time = datetime.timedelta(seconds=seconds, **other_timedelta_kwargs)
+        return obj
+
+    def upto_retries(self, retries=0):
+        """
+        Set limit for retry attempts number (max_times shortcut):
+
+        >>> conf = conf.upto_retries(10)
+        """
+
+        obj = self.clone()
+        obj.max_times = retries
+        return obj
+
+    def _set(self, **kwargs):
+        for prop, value in kwargs.items():
+            if prop not in self._PROPS:
+                continue
+            setattr(self, prop, value)
+
+
+DEFAULT_CONF = RetryConf(logger=LOGGER).waiting(1, backoff=1.2)
+
+
+def retry_call(f, f_args=(), f_kwargs={}, conf=DEFAULT_CONF, **kwargs):
+    """
+    Retry function call.
+
+    :param f:           function to be retried
+    :param f_args:      target function args
+    :param f_kwargs:    target function kwargs
+    :param conf:        configuration
+    """
+
+    if kwargs:
+        conf = conf.clone(**kwargs)
+    return _retry(conf, functools.partial(f, *f_args, **f_kwargs))
+
+
+def retry(conf=DEFAULT_CONF, **kwargs):
+    """
+    Retrying decorator.
+
+    :param conf:        configuration
+    """
+
+    if kwargs:
+        conf = conf.clone(**kwargs)
+
+    def decorator(f):
+        @functools.wraps(f)
+        def wrapped(*f_args, **f_kwargs):
+            return _retry(conf, functools.partial(f, *f_args, **f_kwargs))
+
+        return wrapped
+
+    return decorator
+
+
+def retry_intrusive(f):
+    """
+    Retrying method decorator that uses an intrusive conf (obj.retry_conf).
+    """
+
+    @functools.wraps(f)
+    def wrapped(obj, *f_args, **f_kwargs):
+        assert hasattr(obj, "retry_conf"), "Object must have retry_conf attribute for decorator to run"
+        return _retry(obj.retry_conf, functools.partial(f, obj, *f_args, **f_kwargs))
+
+    return wrapped
+
+
+def _retry(conf, f):
+    start = datetime.datetime.now()
+    delay = 0
+    for n in itertools.count(1):
+        try:
+            return f()
+        except Exception as error:
+            raised_after = datetime.datetime.now() - start
+            if conf.handle_error:
+                conf.handle_error(error, n, raised_after)
+            delay = conf.get_delay(n, raised_after, delay)
+            retry_after = raised_after + datetime.timedelta(seconds=delay)
+            retrying = (
+                conf.retriable(error)
+                and (conf.max_times is None or n <= conf.max_times)
+                and (conf.max_time is None or retry_after <= conf.max_time)
+            )
+            if not retrying:
+                raise
+            if delay:
+                conf.sleep(delay)
+            if conf.logger:
+                conf.logger.warning(
+                    "Retrying (try %d) after %s (%s + %s sec) on %s: %s",
+                    n,
+                    retry_after,
+                    raised_after,
+                    delay,
+                    error.__class__.__name__,
+                    error,
+                    exc_info=True,
+                )
diff --git a/library/python/retry/tests/test_retry.py b/library/python/retry/tests/test_retry.py
new file mode 100644
index 0000000000..2e02fa5960
--- /dev/null
+++ b/library/python/retry/tests/test_retry.py
@@ -0,0 +1,361 @@
+import contextlib
+import datetime
+
+import pytest
+
+import library.python.retry as retry
+
+
+def test_default():
+    @retry.retry()
+    def foo():
+        pass
+
+    foo()
+
+
+def test_exec():
+    ctx = {"run": False}
+
+    @retry.retry()
+    def foo():
+        ctx["run"] = True
+
+    foo()
+
+    assert ctx["run"]
+
+
+class RetriableError(Exception):
+    pass
+
+
+def test_conf():
+    conf = retry.RetryConf()
+
+    conf2 = conf.clone()
+    assert conf2 is not conf
+
+    conf_on = conf.on(RetriableError)
+    assert conf_on.retriable is not conf.retriable
+    assert conf_on.retriable(RetriableError("error"))
+    t = datetime.timedelta(seconds=3)
+
+    conf_waiting = conf.waiting(42, backoff=1.5)
+    assert conf_waiting.get_delay is not conf.get_delay
+    assert conf_waiting.get_delay(3, t, 63) == 94.5
+
+
+class Counter(object):
+    def __init__(self):
+        self.value = 0
+
+    def checkin(self):
+        self.value += 1
+
+
+def DUMMY_RUN(*args, **kwargs):
+    return None
+
+
+@contextlib.contextmanager
+def erroneous_runner(run, n=1, error=Exception):
+    counter = Counter()
+
+    def wrapped_run(*args, **kwargs):
+        counter.checkin()
+        if counter.value <= n:
+            raise error("Error")
+        return run(*args, **kwargs)
+
+    yield wrapped_run
+
+
+@contextlib.contextmanager
+def counting_runner(run, counter):
+    def wrapped_run(*args, **kwargs):
+        counter.checkin()
+        return run(*args, **kwargs)
+
+    yield wrapped_run
+
+
+param_runs = pytest.mark.parametrize("runs", (1, 2, 3))
+
+
+@param_runs
+def test_retries_call(runs):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, runs) as run:
+        with counting_runner(run, counter) as run:
+
+            def foo():
+                run()
+
+    retry.retry_call(foo)
+    assert counter.value == runs + 1
+
+
+@param_runs
+def test_retries_call_args(runs):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, runs) as run:
+        with counting_runner(run, counter) as run:
+
+            def foo(arg, kwarg=None):
+                import logging
+
+                logging.info("!!! %s %s", arg, kwarg)
+                run()
+
+    retry.retry_call(foo, (1,), {"kwarg": 2})
+    assert counter.value == runs + 1
+
+
+@param_runs
+def test_retries_decorator(runs):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, runs) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(retry.RetryConf())
+            def foo():
+                run()
+
+    foo()
+    assert counter.value == runs + 1
+
+
+@param_runs
+def test_retries_decorator_args(runs):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, runs) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(retry.RetryConf())
+            def foo(arg, kwarg=None):
+                run()
+
+    foo(1, kwarg=2)
+    assert counter.value == runs + 1
+
+
+@param_runs
+def test_retries_decorator_method(runs):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, runs) as run:
+        with counting_runner(run, counter) as run:
+
+            class Bar(object):
+                @retry.retry(retry.RetryConf())
+                def foo(self):
+                    run()
+
+    Bar().foo()
+    assert counter.value == runs + 1
+
+
+@param_runs
+def test_retries_decorator_method_args(runs):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, runs) as run:
+        with counting_runner(run, counter) as run:
+
+            class Bar(object):
+                @retry.retry(retry.RetryConf())
+                def foo(self, arg, kwarg=None):
+                    run()
+
+    Bar().foo(1, kwarg=2)
+    assert counter.value == runs + 1
+
+
+@param_runs
+def test_retries_decorator_intrusive(runs):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, runs) as run:
+        with counting_runner(run, counter) as run:
+
+            class Bar(object):
+                def __init__(self):
+                    self.retry_conf = retry.RetryConf()
+
+                @retry.retry_intrusive
+                def foo(self, arg, kwarg=None):
+                    run()
+
+    Bar().foo(1, kwarg=2)
+    assert counter.value == runs + 1
+
+
+def test_retries_decorator_intrusive_fail():
+    class Bar(object):
+        @retry.retry_intrusive
+        def foo(self, arg, kwarg=None):
+            pass
+
+    with pytest.raises(AssertionError):
+        Bar().foo(1, kwarg=2)
+
+
+@pytest.mark.parametrize(
+    "conf",
+    (
+        retry.RetryConf(),
+        retry.DEFAULT_CONF,
+    ),
+)
+def test_confs(conf):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN) as run:
+        with counting_runner(run, counter) as run:
+
+            def foo():
+                run()
+
+    retry.retry_call(foo, conf=conf)
+    assert counter.value == 2
+
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo_retried():
+                run()
+
+    foo_retried()
+    assert counter.value == 2
+
+
+@pytest.mark.parametrize(
+    "conf",
+    (
+        retry.RetryConf().on(RetriableError),
+        retry.RetryConf(retriable=lambda e: isinstance(e, RetriableError)),
+    ),
+)
+def test_retriable(conf):
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, error=RetriableError) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo():
+                run()
+
+    foo()
+    assert counter.value == 2
+
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo():
+                run()
+
+    with pytest.raises(Exception):
+        foo()
+    assert counter.value == 1
+
+
+def test_waiting():
+    conf = retry.RetryConf().waiting(1)
+    with erroneous_runner(DUMMY_RUN) as run:
+
+        @retry.retry(conf)
+        def foo():
+            run()
+
+    foo()
+
+
+def test_waiting_backoff():
+    conf = retry.RetryConf().waiting(1, backoff=2)
+    with erroneous_runner(DUMMY_RUN) as run:
+
+        @retry.retry(conf)
+        def foo():
+            run()
+
+    foo()
+
+
+def test_waiting_jitter():
+    conf = retry.RetryConf().waiting(0, jitter=1)
+    with erroneous_runner(DUMMY_RUN) as run:
+
+        @retry.retry(conf)
+        def foo():
+            run()
+
+    foo()
+
+
+def test_upto():
+    conf = retry.RetryConf().upto(0)
+
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo():
+                run()
+
+    with pytest.raises(Exception):
+        foo()
+    assert counter.value == 1
+
+
+def test_upto_retries():
+    conf = retry.RetryConf().upto_retries(0)
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, 2) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo():
+                run()
+
+    with pytest.raises(Exception):
+        foo()
+    assert counter.value == 1
+
+    conf = retry.RetryConf().upto_retries(1)
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, 2) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo():
+                run()
+
+    with pytest.raises(Exception):
+        foo()
+    assert counter.value == 2
+
+    conf = retry.RetryConf().upto_retries(2)
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, 2) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo():
+                run()
+
+    foo()
+    assert counter.value == 3
+
+    conf = retry.RetryConf().upto_retries(4)
+    counter = Counter()
+    with erroneous_runner(DUMMY_RUN, 2) as run:
+        with counting_runner(run, counter) as run:
+
+            @retry.retry(conf)
+            def foo():
+                run()
+
+    foo()
+    assert counter.value == 3
diff --git a/library/python/retry/tests/ya.make b/library/python/retry/tests/ya.make
new file mode 100644
index 0000000000..5e14cd2093
--- /dev/null
+++ b/library/python/retry/tests/ya.make
@@ -0,0 +1,11 @@
+PY23_TEST()
+
+STYLE_PYTHON()
+
+TEST_SRCS(test_retry.py)
+
+PEERDIR(
+    library/python/retry
+)
+
+END()
diff --git a/library/python/retry/ya.make b/library/python/retry/ya.make
new file mode 100644
index 0000000000..dcbf5567cd
--- /dev/null
+++ b/library/python/retry/ya.make
@@ -0,0 +1,11 @@
+PY23_LIBRARY()
+
+STYLE_PYTHON()
+
+PY_SRCS(__init__.py)
+
+END()
+
+RECURSE_FOR_TESTS(
+    tests
+)