diff --git a/README.rst b/README.rst index d4fd885..b647389 100644 --- a/README.rst +++ b/README.rst @@ -206,6 +206,53 @@ A real life example: applying a patch shell_command_after: - git am "$(git format-patch -1 XXXXXX -o ../patches)" +Network retries +--------------- + +Network git commands (``fetch``, ``pull``, ``push``, ``ls-remote`` and +``clone``) are retried when they fail, with an exponential backoff. This +helps with flaky networks and, above all, with forges that rate limit us: +GitHub answers with an HTTP 401 (which git reports as +``could not read Username for 'https://github.com'``) when too many +unauthenticated git operations come from the same IP address. + +The default is 4 retries, waiting 15 seconds before the first one and +doubling the wait each time (15s, 30s, 60s, 120s), with a random jitter of ++/- 20% so that concurrent aggregations do not retry all at the same time +and get rate limited together again. + +It can be tuned per repository: + +.. code-block:: yaml + + ./product_attribute: + remotes: + oca: https://github.com/OCA/product-attribute.git + merges: + - oca 16.0 + target: oca aggregated_branch_name + retry: + max_retries: 6 + delay: 30 + backoff_factor: 2 + jitter: 0.2 + +Set ``max_retries`` to ``0`` to disable the retries, and ``backoff_factor`` +to ``1`` for a constant delay. Note that the merge check done while loading +the configuration file is never retried, to avoid a very long startup when +the remote is unreachable. + +The same parameters are available on the command line, where they take +precedence over the configuration file, which is handy in a CI environment: + +.. code-block:: bash + + $ gitaggregate -c repos.yaml --retry-max 6 --retry-delay 30 + +.. code-block:: bash + + $ gitaggregate -c repos.yaml --retry-max 0 + Command line Usage ================== diff --git a/git_aggregator/config.py b/git_aggregator/config.py index 266d284..88d19c3 100644 --- a/git_aggregator/config.py +++ b/git_aggregator/config.py @@ -8,18 +8,20 @@ import yaml from ._compat import string_types -from .repo import Repo, ishex +from .repo import RETRY_KEYS, Repo, ishex from .exception import ConfigException log = logging.getLogger(__name__) -def get_repos(config, force=False, skip_merge_check=False): +def get_repos(config, force=False, skip_merge_check=False, retry=None): """Return a :py:obj:`list` list of repos from config file. :param config: the repos config in :py:class:`dict` format. :param bool force: Force aggregate dirty repos or not. :param bool skip_merge_check: True to skip the merge check for non existing refs in remotes. + :param retry: optional dict of retry parameters taking precedence over + the `retry` section of each repo in the config file. :type config: dict :rtype: list """ @@ -35,8 +37,20 @@ def get_repos(config, force=False, skip_merge_check=False): 'apply_patch': repo_data.get('apply_patch', False), 'skip_repo_init': repo_data.get('skip_repo_init', False), } - if 'retry' in repo_data: - repo_dict['retry'] = repo_data['retry'] + repo_retry = dict(repo_data.get('retry') or {}) + unknown_keys = set(repo_retry) - set(RETRY_KEYS) + if unknown_keys: + raise ConfigException( + '%s: Unknown retry parameter(s) %s. Valid ones are %s.' % ( + directory, + ', '.join(sorted(unknown_keys)), + ', '.join(RETRY_KEYS), + ) + ) + # Command line parameters win over the config file ones + repo_retry.update(retry or {}) + if repo_retry: + repo_dict['retry'] = repo_retry remote_names = set() if 'remotes' in repo_data: repo_dict['remotes'] = [] @@ -176,7 +190,8 @@ def get_repos(config, force=False, skip_merge_check=False): def load_config( - config, expand_env=False, env_file=None, force=False, skip_merge_check=False): + config, expand_env=False, env_file=None, force=False, + skip_merge_check=False, retry=None): """Return repos from a directory and fnmatch. Not recursive. :param config: paths to config file @@ -188,6 +203,8 @@ def load_config( :param bool force: True to aggregate even if repo is dirty. :param bool skip_merge_check: True to skip the merge check for non existing refs in remotes. + :param retry: optional dict of retry parameters taking precedence over + the `retry` section of each repo in the config file. :returns: expanded config dict item :rtype: iter(dict) """ @@ -221,4 +238,5 @@ def load_config( conf = yaml.load(config, Loader=yaml.SafeLoader) - return get_repos(conf or {}, force, skip_merge_check=skip_merge_check) + return get_repos( + conf or {}, force, skip_merge_check=skip_merge_check, retry=retry) diff --git a/git_aggregator/main.py b/git_aggregator/main.py index 696bdca..b03eaf6 100644 --- a/git_aggregator/main.py +++ b/git_aggregator/main.py @@ -22,7 +22,7 @@ from .config import load_config from .log import DebugLogFormatter, LogFormatter -from .repo import Repo +from .repo import DEFAULT_RETRY, RETRY_KEYS, Repo from .utils import ThreadNameKeeper logger = logging.getLogger(__name__) @@ -132,6 +132,47 @@ def get_parser(): 'Set `1` or less to disable multiprocessing (default).', ) + main_parser.add_argument( + '--retry-max', + dest='retry_max_retries', + default=None, + type=int, + help='Number of times a failing network git command (fetch, pull, ' + 'push, ls-remote, clone) is retried. Set `0` to disable ' + 'retries. Overrides the `retry` section of the configuration ' + 'file (default: %d).' % DEFAULT_RETRY['max_retries'], + ) + + main_parser.add_argument( + '--retry-delay', + dest='retry_delay', + default=None, + type=float, + help='Seconds to wait before the first retry (default: %d). ' + 'Each subsequent retry waits `--retry-backoff-factor` times ' + 'longer.' % DEFAULT_RETRY['delay'], + ) + + main_parser.add_argument( + '--retry-backoff-factor', + dest='retry_backoff_factor', + default=None, + type=float, + help='Multiplier applied to the delay of each retry (default: %d). ' + 'Set `1` for a constant delay.' % DEFAULT_RETRY['backoff_factor'], + ) + + main_parser.add_argument( + '--retry-jitter', + dest='retry_jitter', + default=None, + type=float, + help='Randomize each retry delay by +/- this ratio, between 0 and 1 ' + '(default: %s). This avoids concurrent aggregations retrying at ' + 'the same time against a rate limiting remote. Set `0` to ' + 'disable.' % DEFAULT_RETRY['jitter'], + ) + main_parser.add_argument( '--no-color', dest='no_color', @@ -182,6 +223,15 @@ def main(): argcomplete.autocomplete(parser, always_complete_options=False) args = parser.parse_args() + if args.retry_max_retries is not None and args.retry_max_retries < 0: + parser.error('--retry-max must be a positive number') + if args.retry_delay is not None and args.retry_delay < 0: + parser.error('--retry-delay must be a positive number') + if args.retry_backoff_factor is not None and args.retry_backoff_factor < 1: + parser.error('--retry-backoff-factor must be greater than or equal 1') + if args.retry_jitter is not None and not 0 <= args.retry_jitter < 1: + parser.error('--retry-jitter must be greater than or equal 0 ' + 'and lower than 1') if args.no_color: colorama.init(strip=True) if not args.command: @@ -197,6 +247,20 @@ def main(): return 1 +def get_retry_overrides(args): + """Return the retry parameters explicitly set on the command line. + + Parameters left out keep the value from the configuration file, or the + :data:`git_aggregator.repo.DEFAULT_RETRY` one. + """ + overrides = {} + for key in RETRY_KEYS: + value = getattr(args, 'retry_%s' % key, None) + if value is not None: + overrides[key] = value + return overrides + + def match_dir(cwd, dirmatch=None): if not dirmatch: return True @@ -208,7 +272,9 @@ def match_dir(cwd, dirmatch=None): def load_aggregate(args): """Load YAML and JSON configs and begin creating / updating , aggregating and pushing the repos (deprecated in favor or run())""" - repos = load_config(args.config, args.expand_env, args.env_file) + repos = load_config( + args.config, args.expand_env, args.env_file, + retry=get_retry_overrides(args)) dirmatch = args.dirmatch for repo_dict in repos: r = Repo(**repo_dict) @@ -253,7 +319,8 @@ def run(args): in args.command""" repos = load_config( - args.config, args.expand_env, args.env_file, args.force) + args.config, args.expand_env, args.env_file, args.force, + retry=get_retry_overrides(args)) jobs = max(args.jobs, 1) threads = [] diff --git a/git_aggregator/repo.py b/git_aggregator/repo.py index 61c3ed9..0e2b445 100644 --- a/git_aggregator/repo.py +++ b/git_aggregator/repo.py @@ -4,6 +4,7 @@ # https://github.com/anybox/anybox.recipe.odoo import logging import os +import random import re import shutil import subprocess @@ -16,7 +17,13 @@ FETCH_DEFAULTS = ("depth", "shallow-since", "shallow-exclude") NETWORK_GIT_COMMANDS = ("fetch", "pull", "push", "ls-remote", "clone") -DEFAULT_RETRY = {"max_retries": 3, "delay": 5, "backoff_factor": 2} +DEFAULT_RETRY = { + "max_retries": 4, + "delay": 15, + "backoff_factor": 2, + "jitter": 0.2, +} +RETRY_KEYS = tuple(DEFAULT_RETRY) logger = logging.getLogger(__name__) @@ -28,6 +35,25 @@ def _is_network_command(cmd): return False +def compute_retry_wait(attempt, delay, backoff_factor, jitter=0): + """Return how many seconds to wait before the given retry attempt. + + Exponential backoff, with an optional random jitter of +/- ``jitter`` + (a ratio of the computed wait). The jitter keeps concurrent + aggregations from retrying in lockstep against a remote that is rate + limiting us, which would only get them limited again together. + + >>> compute_retry_wait(0, 15, 2) + 15 + >>> compute_retry_wait(2, 15, 2) + 60 + """ + wait = delay * backoff_factor ** attempt + if jitter: + wait += wait * random.uniform(-jitter, jitter) + return max(wait, 0) + + def ishex(s): """True iff given string is a valid hexadecimal number. >>> ishex('deadbeef') @@ -75,6 +101,11 @@ def __init__(self, cwd, remotes, merges, target, :param bool skip_repo_init: When ``True``, it will not clone the repository if it does not exist. + :param retry: + Optional dict overriding :data:`DEFAULT_RETRY` for the network + git commands, with the keys ``max_retries``, ``delay``, + ``backoff_factor`` and ``jitter``. Missing keys keep their + default value. """ self.cwd = cwd self.remotes = remotes @@ -90,10 +121,9 @@ def __init__(self, cwd, remotes, merges, target, self.skip_dry_run = skip_dry_run self.apply_patch = apply_patch self.skip_repo_init = skip_repo_init - if retry is None: - self.retry = dict(DEFAULT_RETRY) - else: - self.retry = retry + self.retry = dict(DEFAULT_RETRY) + if retry: + self.retry.update(retry) @property def git_version(self): @@ -189,6 +219,7 @@ def log_call(self, cmd, callwith=subprocess.check_call, backoff_factor = self.retry.get( "backoff_factor", DEFAULT_RETRY["backoff_factor"] ) + jitter = self.retry.get("jitter", DEFAULT_RETRY["jitter"]) last_exception = None for attempt in range(max_retries + 1): try: @@ -199,9 +230,11 @@ def log_call(self, cmd, callwith=subprocess.check_call, except Exception as exc: last_exception = exc if attempt < max_retries: - wait = delay * backoff_factor ** attempt + wait = compute_retry_wait( + attempt, delay, backoff_factor, jitter + ) logger.warning( - "%s> Retry %d/%d for %r in %ds...", + "%s> Retry %d/%d for %r in %.1fs...", self.cwd, attempt + 1, max_retries, cmd, wait, ) time.sleep(wait) diff --git a/tests/test_config.py b/tests/test_config.py index ab2eba2..ec35773 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -435,3 +435,62 @@ def test_fetch_all_true(self): config_yaml = dedent(config_yaml) repos = config.get_repos(self._parse_config(config_yaml)) self.assertIs(repos[0]["fetch_all"], True) + + def test_load_retry(self): + config_yaml = """ +/product_attribute: + remotes: + oca: https://github.com/OCA/product-attribute.git + merges: + - oca 8.0 + target: oca aggregated_branch + retry: + max_retries: 2 + delay: 30 +""" + repos = config.get_repos( + self._parse_config(config_yaml), skip_merge_check=True) + self.assertEqual( + repos[0]['retry'], {'max_retries': 2, 'delay': 30}) + # command line parameters win over the ones of the config file + repos = config.get_repos( + self._parse_config(config_yaml), skip_merge_check=True, + retry={'delay': 60}) + self.assertEqual( + repos[0]['retry'], {'max_retries': 2, 'delay': 60}) + + def test_load_retry_without_config(self): + config_yaml = """ +/product_attribute: + remotes: + oca: https://github.com/OCA/product-attribute.git + merges: + - oca 8.0 + target: oca aggregated_branch +""" + repos = config.get_repos( + self._parse_config(config_yaml), skip_merge_check=True) + self.assertNotIn('retry', repos[0]) + repos = config.get_repos( + self._parse_config(config_yaml), skip_merge_check=True, + retry={'max_retries': 0}) + self.assertEqual(repos[0]['retry'], {'max_retries': 0}) + + def test_load_retry_exception(self): + config_yaml = """ +/product_attribute: + remotes: + oca: https://github.com/OCA/product-attribute.git + merges: + - oca 8.0 + target: oca aggregated_branch + retry: + max_retry: 2 +""" + with self.assertRaises(ConfigException) as ex: + config.get_repos( + self._parse_config(config_yaml), skip_merge_check=True) + self.assertEqual( + ex.exception.args[0], + '/product_attribute: Unknown retry parameter(s) max_retry. ' + 'Valid ones are max_retries, delay, backoff_factor, jitter.') diff --git a/tests/test_repo.py b/tests/test_repo.py index 91c303e..2e67774 100644 --- a/tests/test_repo.py +++ b/tests/test_repo.py @@ -20,9 +20,10 @@ import logging from tempfile import mkdtemp from textwrap import dedent +from unittest import mock from git_aggregator import exception, main -from git_aggregator.repo import Repo +from git_aggregator.repo import DEFAULT_RETRY, Repo, compute_retry_wait from git_aggregator.utils import ( WorkingDirectoryKeeper, working_directory_keeper, @@ -444,3 +445,96 @@ def test_multithreading(self): self.assertTrue(os.path.isfile(os.path.join(repo3_dir, 'tracked'))) self.assertTrue(os.path.isfile(os.path.join(repo3_dir, 'tracked2'))) + + +class TestRetry(unittest.TestCase): + """Test the retry mechanism of the network git commands.""" + + def _failing_call(self, calls, failures=None): + """Return a callwith that appends to `calls` and fails `failures` + times (always if `failures` is None).""" + def callwith(cmd, **kw): + calls.append(cmd) + if failures is None or len(calls) <= failures: + raise subprocess.CalledProcessError(128, cmd) + return 0 + return callwith + + def test_compute_retry_wait(self): + self.assertEqual(compute_retry_wait(0, 15, 2), 15) + self.assertEqual(compute_retry_wait(1, 15, 2), 30) + self.assertEqual(compute_retry_wait(2, 15, 2), 60) + # a factor of 1 gives a constant delay + self.assertEqual(compute_retry_wait(3, 15, 1), 15) + + def test_compute_retry_wait_jitter(self): + for attempt in range(4): + expected = 15 * 2 ** attempt + for _i in range(50): + wait = compute_retry_wait(attempt, 15, 2, jitter=0.2) + self.assertGreaterEqual(wait, expected * 0.8) + self.assertLessEqual(wait, expected * 1.2) + # the jitter is random, so the waits must not all be the same + waits = {compute_retry_wait(0, 15, 2, jitter=0.2) for _i in range(50)} + self.assertGreater(len(waits), 1) + + def test_defaults(self): + repo = Repo('/tmp/notexisting', [], [], None) + self.assertEqual(repo.retry, DEFAULT_RETRY) + # missing keys keep their default value + repo = Repo('/tmp/notexisting', [], [], None, + retry={'max_retries': 0}) + self.assertEqual(repo.retry['max_retries'], 0) + self.assertEqual(repo.retry['delay'], DEFAULT_RETRY['delay']) + + def test_retry_network_command(self): + repo = Repo('/tmp/notexisting', [], [], None, retry={ + 'max_retries': 3, 'delay': 10, 'backoff_factor': 3, 'jitter': 0, + }) + calls = [] + with mock.patch('git_aggregator.repo.time.sleep') as sleep: + with self.assertRaises(subprocess.CalledProcessError): + repo.log_call( + ['git', 'fetch', 'origin', 'refs/pull/42/head'], + callwith=self._failing_call(calls)) + # the initial call plus one per retry + self.assertEqual(len(calls), 4) + self.assertEqual( + [c.args[0] for c in sleep.call_args_list], [10, 30, 90]) + + def test_retry_network_command_success(self): + repo = Repo('/tmp/notexisting', [], [], None, retry={ + 'max_retries': 3, 'delay': 10, + }) + calls = [] + with mock.patch('git_aggregator.repo.time.sleep'): + ret = repo.log_call( + ['git', 'fetch', 'origin', 'refs/pull/42/head'], + callwith=self._failing_call(calls, failures=2)) + self.assertEqual(ret, 0) + self.assertEqual(len(calls), 3) + + def test_no_retry(self): + repo = Repo('/tmp/notexisting', [], [], None, + retry={'max_retries': 0}) + calls = [] + with mock.patch('git_aggregator.repo.time.sleep') as sleep: + with self.assertRaises(subprocess.CalledProcessError): + repo.log_call( + ['git', 'fetch', 'origin', 'refs/pull/42/head'], + callwith=self._failing_call(calls)) + self.assertEqual(len(calls), 1) + sleep.assert_not_called() + + def test_no_retry_local_command(self): + """Only the network commands are retried.""" + repo = Repo('/tmp/notexisting', [], [], None, + retry={'max_retries': 3, 'delay': 10}) + calls = [] + with mock.patch('git_aggregator.repo.time.sleep') as sleep: + with self.assertRaises(subprocess.CalledProcessError): + repo.log_call( + ['git', 'checkout', '-B', 'aggregated'], + callwith=self._failing_call(calls)) + self.assertEqual(len(calls), 1) + sleep.assert_not_called()