diff --git a/pointCollection/io_utils.py b/pointCollection/io_utils.py index 61eea30..ea38621 100644 --- a/pointCollection/io_utils.py +++ b/pointCollection/io_utils.py @@ -30,11 +30,16 @@ 'NSIDC': 'https://data.nsidc.earthdatacloud.nasa.gov/s3credentials', } -# The broker call is tried this many times, this far apart. The MAAP API -# stops answering when several hundred jobs start together: on 2026-10-01, 61 -# of 556 jobs submitted at once lost their one attempt, each after ~135 s. -MAAP_BROKER_ATTEMPTS = 5 -MAAP_BROKER_PAUSE_S = 10 +# The broker call is retried after pauses that grow, each jittered by +# +-MAAP_BROKER_JITTER so that jobs which failed together do not retry +# together, and no pause is taken that would end past MAAP_BROKER_BUDGET_S +# from the first try -- a node waiting costs money. The MAAP API stops +# answering when several hundred jobs start together: on 2026-10-01, 61 of 556 +# jobs submitted at once lost their one attempt, each after ~135 s. The same +# schedule as ATL1415's workspace_credentials.py (plan_pack_tiles.sh K2-K3). +MAAP_BROKER_PAUSES_S = (10, 20, 40, 60, 60) # between tries: up to 6 tries +MAAP_BROKER_JITTER = 0.5 +MAAP_BROKER_BUDGET_S = 240 # daac -> why the last broker call failed, for the error get_s3fs() raises # when the earthaccess fallback has nothing either. @@ -176,9 +181,13 @@ def _s3fs_from_maap(daac, **kwargs): environment or the broker will not answer, so the caller falls back to earthaccess. Off MAAP this costs one dict lookup and returns (None, None). - The broker call is tried MAAP_BROKER_ATTEMPTS times, MAAP_BROKER_PAUSE_S - apart, before giving up; the reason it gave up is kept in _BROKER_FAILURES - so that get_s3fs() can report it if earthaccess cannot help either. + The broker call is tried up to len(MAAP_BROKER_PAUSES_S) + 1 times, after + the pauses above, on ONE MAAP client (building it is itself an API call), + before giving up; the reason it gave up is kept in _BROKER_FAILURES so + that get_s3fs() can report it if earthaccess cannot help either. HTTP 401 + gives up at once: MAAP_PGT was rejected, and retrying the same token + cannot help. There is no per-try timeout: this may run off the main + thread, where SIGALRM is not available. This exists because a MAAP DPS worker has NO Earthdata credentials: it runs as root with no ~/.netrc, and earthaccess's netrc and environment @@ -199,6 +208,7 @@ def _s3fs_from_maap(daac, **kwargs): cause. """ import os + import random import time import warnings @@ -218,18 +228,32 @@ def _s3fs_from_maap(daac, **kwargs): f'falling back to earthaccess for {daac}.') return None, None - for attempt in range(1, MAAP_BROKER_ATTEMPTS + 1): + client = None + attempts = len(MAAP_BROKER_PAUSES_S) + 1 + t0 = time.monotonic() + why = '' + for attempt in range(1, attempts + 1): try: - creds = MAAP( - maap_host=os.environ.get('MAAP_API_HOST', 'api.maap-project.org') - ).aws.earthdata_s3_credentials(endpoint) + if client is None: + client = MAAP(maap_host=os.environ.get('MAAP_API_HOST', 'api.maap-project.org')) + creds = client.aws.earthdata_s3_credentials(endpoint) return _s3fs_with_credentials(creds, **kwargs), _expiry_timestamp(creds) except Exception as exc: last = f'{type(exc).__name__}: {exc}' - if attempt < MAAP_BROKER_ATTEMPTS: - time.sleep(MAAP_BROKER_PAUSE_S) + if getattr(getattr(exc, 'response', None), 'status_code', None) == 401: + why = ('; HTTP 401: MAAP_PGT was rejected (likely the runner\'s token fetch ' + 'failed at job start), and a retry cannot help') + break + if attempt == attempts: + break + nap = MAAP_BROKER_PAUSES_S[attempt - 1] * random.uniform(1 - MAAP_BROKER_JITTER, + 1 + MAAP_BROKER_JITTER) + if time.monotonic() - t0 + nap > MAAP_BROKER_BUDGET_S: + why = f'; a {nap:.0f} s pause would pass the {MAAP_BROKER_BUDGET_S} s budget' + break + time.sleep(nap) _BROKER_FAILURES[daac] = (f'MAAP could not broker {daac} credentials from {endpoint} ' - f'in {MAAP_BROKER_ATTEMPTS} attempts (last error: {last})') + f'in {attempt} attempts (last error: {last}){why}') warnings.warn(f'{_BROKER_FAILURES[daac]}; falling back to earthaccess.') return None, None diff --git a/tests/test_maap_broker.py b/tests/test_maap_broker.py index 6c1a760..3772db7 100644 --- a/tests/test_maap_broker.py +++ b/tests/test_maap_broker.py @@ -26,26 +26,38 @@ @pytest.fixture def broker(monkeypatch): - """A stand-in maap-py whose broker fails `failures` times, then answers.""" - state = {'calls': 0, 'failures': 0, 'naps': []} + """A stand-in maap-py whose broker fails `failures` times (with `error`), + then answers; whose client fails to build `build_failures` times; and a + clock that only sleep() moves, with the jitter fixed at `jitter`.""" + state = {'calls': 0, 'failures': 0, 'naps': [], 'builds': 0, 'build_failures': 0, + 'error': ConnectionError('timed out'), 'now': 0.0, 'jitter': 1.0} class AWS: def earthdata_s3_credentials(self, endpoint): state['calls'] += 1 if state['calls'] <= state['failures']: - raise ConnectionError('timed out') + raise state['error'] return dict(CREDS) class MAAP: def __init__(self, maap_host=None): + state['builds'] += 1 + if state['builds'] <= state['build_failures']: + raise ConnectionError('config timed out') self.aws = AWS() + def sleep(s): + state['naps'].append(s) + state['now'] += s + package, module = types.ModuleType('maap'), types.ModuleType('maap.maap') module.MAAP = MAAP monkeypatch.setitem(sys.modules, 'maap', package) monkeypatch.setitem(sys.modules, 'maap.maap', module) monkeypatch.setenv('MAAP_PGT', 'set') - monkeypatch.setattr('time.sleep', state['naps'].append) + monkeypatch.setattr('time.sleep', sleep) + monkeypatch.setattr('time.monotonic', lambda: state['now']) + monkeypatch.setattr('random.uniform', lambda a, b: state['jitter'] * (a + b) / 2) monkeypatch.setattr(io_utils, '_s3fs_with_credentials', lambda creds, **kw: ('fs', creds)) monkeypatch.setattr(io_utils, '_S3FS_CACHE', {}) monkeypatch.setattr(io_utils, '_BROKER_FAILURES', {}) @@ -68,15 +80,47 @@ def test_the_broker_call_is_retried(broker): fs, expires = io_utils._s3fs_from_maap('NSIDC') assert fs == ('fs', CREDS) and expires is not None assert broker['calls'] == 3 - assert broker['naps'] == [io_utils.MAAP_BROKER_PAUSE_S] * 2 + assert broker['naps'] == list(io_utils.MAAP_BROKER_PAUSES_S[:2]) + assert broker['builds'] == 1 # one client for every try def test_giving_up_warns_with_the_reason(broker): broker['failures'] = 99 - with pytest.warns(UserWarning, match=r'in 5 attempts \(last error: ConnectionError: timed out\)'): + with pytest.warns(UserWarning, match=r'in 6 attempts \(last error: ConnectionError: timed out\)'): + assert io_utils._s3fs_from_maap('NSIDC') == (None, None) + assert broker['calls'] == len(io_utils.MAAP_BROKER_PAUSES_S) + 1 + assert broker['naps'] == list(io_utils.MAAP_BROKER_PAUSES_S) # none after the last + assert sum(broker['naps']) <= io_utils.MAAP_BROKER_BUDGET_S + + +def test_pauses_are_jittered(broker): + broker['failures'], broker['jitter'] = 1, 0.5 + io_utils._s3fs_from_maap('NSIDC') + assert broker['naps'] == [io_utils.MAAP_BROKER_PAUSES_S[0] * 0.5] + + +def test_no_pause_past_the_budget(broker): + # jitter at its top: 15 + 30 + 60 + 90 = 195 s, and the next 90 s would end at 285 + broker['failures'], broker['jitter'] = 99, 1.5 + with pytest.warns(UserWarning, match=r'in 5 attempts .*a 90 s pause would pass the 240 s budget'): + assert io_utils._s3fs_from_maap('NSIDC') == (None, None) + assert broker['naps'] == [15, 30, 60, 90] + + +def test_401_stops_at_once(broker): + class HTTPError(Exception): + response = types.SimpleNamespace(status_code=401) + broker['failures'], broker['error'] = 99, HTTPError('401 Unauthorized') + with pytest.warns(UserWarning, match=r'in 1 attempts .*HTTP 401: MAAP_PGT was rejected'): assert io_utils._s3fs_from_maap('NSIDC') == (None, None) - assert broker['calls'] == io_utils.MAAP_BROKER_ATTEMPTS - assert len(broker['naps']) == io_utils.MAAP_BROKER_ATTEMPTS - 1 # none after the last + assert broker['calls'] == 1 and broker['naps'] == [] + + +def test_a_client_that_failed_to_build_is_built_again(broker): + broker['build_failures'] = 2 + fs, _ = io_utils._s3fs_from_maap('NSIDC') + assert fs == ('fs', CREDS) + assert broker['builds'] == 3 and broker['calls'] == 1 def test_no_broker_and_no_earthaccess_login_names_both(broker, monkeypatch):