From 04c2064e5901ce7dfe1e1ca1dac22096db0a343e Mon Sep 17 00:00:00 2001 From: Ben Smith Date: Tue, 8 Sep 2026 21:28:39 +0000 Subject: [PATCH] Expire brokered credentials, filter by release, and fix the granule block size Four fixes from a review of the MAAP/S3 path against the platform docs. P1, credential expiry. get_s3fs() cached a session for the life of the process, but MAAP-brokered DAAC credentials expire -- measured against the live broker, the NSIDC response carries expiration exactly one hour out. A read phase that outlives it starts 403-ing partway through, which at a per-tile fan-out reads as a random data error rather than as an expiry. The cache now holds (filesystem, expires_at) and re-derives inside a 600 s margin. An unparseable or absent expiration falls back to a SHORT lifetime rather than an unlimited one: guessing "no expiry" from a value we failed to read is how a session outlives its credentials. Sessions with nothing to expire -- daac=None, which uses the self-refreshing default chain -- cache as before. Exercised against the live broker: expiry parsed, repeat call cached, a session forced inside the margin re-derived, daac=None recorded no expiry, a malformed expiration warned and fell back to 40 minutes, and a real granule still read. C6, release filtering. find_ATL11_granules() searched with no filter at all, so it returned every generation CMR holds for the short name. That is benign only while exactly one per-granule index subtree is staged -- a second would make every tile read both and double-count its data, silently. granule_release= now restricts results to one (cycles, release, version) triple, matched on granule names because CMR's own version filter selects the COLLECTION version (007) and cannot separate cycle ranges or granule versions within it. release_spec() parses '007_cycle_03_31_v04' into ('0331','007','04'). IT IS DELIBERATELY PER-PRODUCT, WITH NO MODULE-LEVEL DEFAULT: ATL11 and ATL11XO are normally different generations -- rel_006_0331 pairs --ATL11_release= 007_cycle_03_31_v04 with --ATL11xo_version=007_cycle_03_30_v03, differing in both cycle range and version under one release -- so a single global "current release" would silently filter a crossover search by ATL11's cycles. Verified both parse distinctly and that a granule of either generation matches only its own spec. C7, missing indices. A granule whose index was absent warned and was skipped, which silently produced a partial fit: a half-uploaded index tree (~8100 files) gave a tile fewer granules than it should have had, with nothing recorded. Now that the search is filtered to the generation we indexed, every granule that reaches the read is one we expect an index for, so its absence is a fault and raises. missing_index='skip' restores best-effort reads, and the option is threaded through all three entry points. C5, block size. geoIndex.get_data() opened the granule with no block_size, so DEFAULT_REMOTE_BLOCK_SIZE never reached the read it was measured for -- scattered index_ranges out of a chunked, compressed granule, where Q27 measured 35 MiB against 6.6 MiB for the same window. The index-file read is left on the larger default deliberately: an index averages ~250 KiB and is read end to end. Also: the hard-coded earthaccess.login(strategy='netrc') had a THIRD site, tilingSchema.resolve_files_for_box() (the crossover path), which the review missed and a grep found. The helper now lives in io_utils as try_earthaccess_login(), beside get_s3fs(), so the login policy and the credential policy stay together; query_ATL11_cloud keeps the old private name as a delegating alias for outside callers. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013MdMEXw2s6eeGtaGcaDxpF --- pointCollection/geoIndex.py | 12 +- pointCollection/io_utils.py | 135 ++++++++++++--- pointCollection/scripts/query_ATL11_cloud.py | 163 +++++++++++++++---- pointCollection/tilingSchema.py | 6 +- 4 files changed, 257 insertions(+), 59 deletions(-) diff --git a/pointCollection/geoIndex.py b/pointCollection/geoIndex.py index 00b64e6..f735cd4 100644 --- a/pointCollection/geoIndex.py +++ b/pointCollection/geoIndex.py @@ -773,7 +773,17 @@ def get_data(self, query_results, fields=None, data=None, dir_root='', # this keeps the shared handle's access pattern close to # monotonic rather than jumping around arbitrarily. tasks.sort(key=lambda t: (t['pair_num'], int(t['index_range'][0]))) - with pc.io_utils.open_h5(physical_file, fs=fs) as h5f: + # THIS is the read DEFAULT_REMOTE_BLOCK_SIZE was measured for: + # scattered index_ranges out of a chunked, compressed granule, + # where fsspec's 5 MiB default reads mostly waste (Q27 measured + # 35 MiB against 6.6 MiB for the same window). It was not being + # passed, so the most windowed read in the job ran on the + # default. The index-file read in from_file() above is left + # alone deliberately -- an index averages ~250 KiB and is read + # end to end, which is what the larger default suits. + with pc.io_utils.open_h5( + physical_file, fs=fs, + block_size=pc.io_utils.DEFAULT_REMOTE_BLOCK_SIZE) as h5f: for task in tasks: try: if task['type'] == 'h5': diff --git a/pointCollection/io_utils.py b/pointCollection/io_utils.py index ccfaa3c..adc84d4 100644 --- a/pointCollection/io_utils.py +++ b/pointCollection/io_utils.py @@ -8,8 +8,18 @@ """ import re +# key -> (filesystem, expires_at or None). Brokered DAAC credentials expire +# (MAAP/NSIDC issue roughly four hours), so the session cannot simply be cached +# for the life of the process: a long read phase would start 403-ing partway +# through, which at a per-tile fan-out looks like a random data error rather +# than an expiry. A session with no expiry (daac=None, the worker's own +# identity) caches as before. _S3FS_CACHE = {} +# Re-derive this many seconds BEFORE the stated expiry, so a read that starts +# just under the wire does not expire mid-request. +_CREDENTIAL_SAFETY_MARGIN = 600 + # DAAC -> the DAAC's own S3-credentials endpoint, for the MAAP path in # get_s3fs(). MAAP brokers these: maap.aws.earthdata_s3_credentials() # returns short-lived accessKeyId/secretAccessKey/sessionToken using MAAP's own @@ -94,30 +104,46 @@ def get_s3fs(daac='NSIDC', **kwargs): masks and tiling schemas on s3://maap-ops-workspace/... -- since earthaccess credentials do not grant access to them. - Sessions are cached by (daac, kwargs) so repeated calls don't re-derive - credentials. + Sessions are cached by (daac, kwargs), so repeated calls are a dict lookup + -- but a cached session whose brokered credentials are within + _CREDENTIAL_SAFETY_MARGIN of expiring is discarded and re-derived. Callers + doing a long sequence of reads should therefore call this per item rather + than hoisting one filesystem out of the loop, which costs a dict lookup and + is what makes the refresh actually take effect. """ + import time + key = (daac, tuple(sorted(kwargs.items()))) - if key not in _S3FS_CACHE: - if daac is None: - import s3fs - _S3FS_CACHE[key] = s3fs.S3FileSystem(**kwargs) - else: - fs = _s3fs_from_maap(daac, **kwargs) - if fs is None: - import earthaccess - fs = earthaccess.get_s3fs_session(daac=daac, **kwargs) - _S3FS_CACHE[key] = fs - return _S3FS_CACHE[key] + entry = _S3FS_CACHE.get(key) + if entry is not None: + fs, expires_at = entry + if expires_at is None or time.time() < expires_at - _CREDENTIAL_SAFETY_MARGIN: + return fs + # Otherwise fall through and re-derive: the credentials this session + # carries are about to stop working. + + if daac is None: + import s3fs + # The default credential chain refreshes itself; nothing to expire here. + _S3FS_CACHE[key] = (s3fs.S3FileSystem(**kwargs), None) + else: + fs, expires_at = _s3fs_from_maap(daac, **kwargs) + if fs is None: + import earthaccess + # earthaccess manages its own session; we do not know its expiry. + fs, expires_at = earthaccess.get_s3fs_session(daac=daac, **kwargs), None + _S3FS_CACHE[key] = (fs, expires_at) + return _S3FS_CACHE[key][0] def _s3fs_from_maap(daac, **kwargs): """ Build an s3fs session from MAAP-brokered DAAC credentials. - Returns None -- with a warning saying why -- whenever this is not a MAAP + Returns (filesystem, expires_at) where expires_at is a POSIX timestamp, or + (None, None) -- with a warning saying why -- whenever this is not a MAAP environment or the broker will not answer, so the caller falls back to - earthaccess. Off MAAP this costs one dict lookup and returns None. + earthaccess. Off MAAP this costs one dict lookup and returns (None, None). This exists because a MAAP DPS worker has NO Earthdata credentials: it runs as root with no ~/.netrc, and earthaccess's netrc and environment @@ -126,10 +152,11 @@ def _s3fs_from_maap(daac, **kwargs): maap.aws.earthdata_s3_credentials() exchanges for the DAAC's temporary read credentials. See docs.maap-project.org, science/NISAR/NISAR_access.html. - The credentials are short-lived and get_s3fs() caches the session for the - life of the process. That is fine for a per-tile job of minutes; a process - that runs longer than the token's `expiration` would need to re-derive, and - nothing here does that yet. + The credentials are short-lived -- MAAP/NSIDC issue roughly four hours -- + so the response's `expiration` is parsed and handed back for get_s3fs() to + cache against. An Antarctic tile whose ATL11 read phase runs long would + otherwise start 403-ing partway through, and at a per-tile fan-out that + reads as a random data error rather than as an expiry. Every failure warns rather than passing silently: a session that quietly came from somewhere other than where you think is exactly the kind of @@ -141,28 +168,88 @@ def _s3fs_from_maap(daac, **kwargs): endpoint = MAAP_S3_CREDENTIALS_ENDPOINTS.get(str(daac).upper()) if endpoint is None: - return None + return None, None if not os.environ.get('MAAP_PGT'): # The ADE and DPS workers both set it; its absence means this is not a # MAAP environment, which is not worth warning about. - return None + return None, None try: from maap.maap import MAAP except ImportError as exc: warnings.warn(f'MAAP_PGT is set but maap-py is not importable ({exc}); ' f'falling back to earthaccess for {daac}.') - return None + return None, None try: creds = MAAP( maap_host=os.environ.get('MAAP_API_HOST', 'api.maap-project.org') ).aws.earthdata_s3_credentials(endpoint) - return _s3fs_with_credentials(creds, **kwargs) + return _s3fs_with_credentials(creds, **kwargs), _expiry_timestamp(creds) except Exception as exc: warnings.warn(f'MAAP could not broker {daac} credentials from {endpoint} ' f'({type(exc).__name__}: {exc}); falling back to earthaccess.') - return None + return None, None + + +def try_earthaccess_login(): + """ + Log in to Earthdata if we can, and carry on if we cannot. + + A CMR metadata search needs no authentication -- only granule READS do, and + those get their credentials from get_s3fs(), which on MAAP uses MAAP's + broker rather than earthaccess. Callers used to write + earthaccess.login(strategy='netrc'), which hard-codes the ONE strategy a + MAAP DPS worker cannot satisfy: it runs as root with no ~/.netrc, so the + search raised LoginStrategyUnavailable before ever reaching CMR. + + Bare login() tries environment, then netrc, then interactive, so a local + user's existing setup still works. Failure warns rather than raising, + because the caller very likely does not need it. + """ + import warnings + import earthaccess + try: + earthaccess.login() + except Exception as exc: + warnings.warn(f'earthaccess.login() failed ({type(exc).__name__}: {exc}); ' + 'continuing, since a CMR search needs no credentials. ' + 'Granule reads get their credentials separately, via ' + 'pointCollection.io_utils.get_s3fs().') + + +def _expiry_timestamp(creds): + """ + POSIX timestamp for a credential response's `expiration`, or None. + + The field arrives as e.g. '2026-09-08 22:19:41+00:00'. An unparseable or + absent value is treated as a SHORT lifetime rather than an unlimited one: + guessing "no expiry" from a value we failed to read is how a session + outlives its credentials. + """ + import datetime + import warnings + + raw = creds.get('expiration') + if raw is None: + warnings.warn('credential response carried no expiration; ' + 're-deriving conservatively.') + return _conservative_expiry() + try: + when = datetime.datetime.fromisoformat(str(raw)) + except ValueError: + warnings.warn(f'could not parse credential expiration {raw!r}; ' + 're-deriving conservatively.') + return _conservative_expiry() + if when.tzinfo is None: + when = when.replace(tzinfo=datetime.timezone.utc) + return when.timestamp() + + +def _conservative_expiry(): + """A short fallback lifetime for a credential response we could not read.""" + import time + return time.time() + 1800 + _CREDENTIAL_SAFETY_MARGIN def _s3fs_with_credentials(creds, **kwargs): diff --git a/pointCollection/scripts/query_ATL11_cloud.py b/pointCollection/scripts/query_ATL11_cloud.py index 98f04f3..f1c37c6 100644 --- a/pointCollection/scripts/query_ATL11_cloud.py +++ b/pointCollection/scripts/query_ATL11_cloud.py @@ -30,7 +30,65 @@ r'ATL11_(?P\d{6})_(?P\d{4})_(?P\d{3})_(?P\d{2})\.h5$') -def find_ATL11_granules(bounding_box, short_name='ATL11', **search_kwargs): +def release_spec(release_string): + """ + Parse a release string into the (cycles, release, version) triple that + identifies one ATL11-family product generation. + + ATL1415 spells a generation as e.g. '007_cycle_03_31_v04', which is a + DIRECTORY-NAMING convention on the processing side, not a CMR selector: + make_ATL11_index.py takes only split('_')[0] from it. The same three + numbers appear in a granule name (ATL11_000103_0331_007_04.h5) and in the + per-granule index subtree (ATL11_index_0331_007_04), which is what makes + the triple the right thing to filter on. + + THE ATL11 AND ATL11XO GENERATIONS ARE NORMALLY DIFFERENT -- rel_006_0331 + pairs --ATL11_release=007_cycle_03_31_v04 with + --ATL11xo_version=007_cycle_03_30_v03, i.e. different cycle ranges AND + different versions under the same release 007. So a caller must pass the + spec for the product it is actually searching; there is deliberately no + module-level "current release" to fall back on, because a single global + would silently filter an ATL11XO search by ATL11's cycles. + + Parameters + ---------- + release_string : str + '007_cycle_03_31_v04', optionally prefixed 'ATL11_'; or an already + parsed '0331_007_04'. + + Returns + ------- + (cycles, release, version) : tuple of str, e.g. ('0331', '007', '04') + """ + text = str(release_string) + if text.startswith('ATL11_'): + text = text[len('ATL11_'):] + + m = re.match(r'^(?P\d{3})_cycle_(?P\d{2})_(?P\d{2})_v(?P\d{2})$', + text) + if m is not None: + return (m['c0'] + m['c1'], m['release'], m['version']) + + m = re.match(r'^(?P\d{4})_(?P\d{3})_(?P\d{2})$', text) + if m is not None: + return (m['cycles'], m['release'], m['version']) + + raise ValueError( + f'cannot parse release string {release_string!r}; expected ' + "'_cycle___v' (e.g. 007_cycle_03_31_v04) or " + "'__' (e.g. 0331_007_04)") + + +def granule_matches_release(granule_basename, spec): + """True if an ATL11-family granule name carries the given release triple.""" + m = _ATL11_RE.search(granule_basename) + if m is None: + return False + return (m['cycles'], m['release'], m['version']) == tuple(spec) + + +def find_ATL11_granules(bounding_box, short_name='ATL11', granule_release=None, + **search_kwargs): """ Search Earthdata Cloud for ATL11 granules intersecting bounding_box. @@ -40,6 +98,18 @@ def find_ATL11_granules(bounding_box, short_name='ATL11', **search_kwargs): (lon_min, lat_min, lon_max, lat_max), as expected by earthaccess.search_data(). short_name : str, optional + granule_release : str or tuple, optional + Restrict results to one product generation, as a release string + ('007_cycle_03_31_v04') or an already parsed (cycles, release, + version) triple. CMR's own `version` filter selects the COLLECTION + version (007) and cannot separate cycle ranges or granule versions + within it, so the filtering is done on the granule names. + + Left None the search returns every generation CMR holds for + short_name. That is only safe while exactly one generation has a + staged per-granule index -- a second staged index subtree would make + every tile read both and double-count its data. Pass this whenever + the caller knows which generation it indexed. **search_kwargs : additional keyword arguments passed to earthaccess.search_data() (e.g. version, cycle, temporal). @@ -50,34 +120,33 @@ def find_ATL11_granules(bounding_box, short_name='ATL11', **search_kwargs): """ import earthaccess _try_earthaccess_login() - return earthaccess.search_data(short_name=short_name, bounding_box=bounding_box, **search_kwargs) + granules = earthaccess.search_data(short_name=short_name, + bounding_box=bounding_box, **search_kwargs) + if granule_release is None: + return granules + + spec = (release_spec(granule_release) if isinstance(granule_release, str) + else tuple(granule_release)) + kept = [] + for granule in granules: + try: + url = granule.data_links(access='direct')[0] + except (IndexError, KeyError): + continue + if granule_matches_release(os.path.basename(url), spec): + kept.append(granule) + return kept def _try_earthaccess_login(): """ - Log in to Earthdata if we can, and carry on if we cannot. - - A CMR metadata search needs no authentication -- only the granule READS do, - and those go through pc.io_utils.get_s3fs(), which on MAAP gets its - credentials from MAAP's broker rather than from earthaccess. This used to - be earthaccess.login(strategy='netrc'), which hard-codes the ONE strategy a - MAAP DPS worker cannot satisfy: it runs as root with no ~/.netrc, so every - search raised LoginStrategyUnavailable before reaching CMR at all, even - though the search itself needed no credentials. - - Bare login() tries environment, then netrc, then interactive, so it still - picks up whatever a local user has. Failure warns rather than raising: - the caller may well not need it. + Best-effort Earthdata login; see pc.io_utils.try_earthaccess_login(). + + Kept as a name here because callers outside this module import it from + here, but the implementation lives in io_utils with get_s3fs(), so the + login policy and the credential policy stay in one place. """ - import warnings - import earthaccess - try: - earthaccess.login() - except Exception as exc: - warnings.warn(f'earthaccess.login() failed ({type(exc).__name__}: {exc}); ' - 'continuing, since a CMR search needs no credentials. ' - 'Granule reads get their credentials separately, via ' - 'pointCollection.io_utils.get_s3fs().') + return pc.io_utils.try_earthaccess_login() def index_path_for_granule(granule_basename, index_root): @@ -97,7 +166,8 @@ def index_path_for_granule(granule_basename, index_root): def read_ATL11_granule_cloud_items(s3_url, index_file, xr, yr, fields=None, fs=None, - index_fs=None, version_mismatch='error'): + index_fs=None, version_mismatch='error', + missing_index='error'): """ Find the rows of a cloud ATL11 granule falling within [xr, yr], using a pre-built per-granule geoIndex to locate them, and return them as the @@ -143,6 +213,8 @@ def read_ATL11_granule_cloud_items(s3_url, index_file, xr, yr, fields=None, fs=N """ if version_mismatch not in ('error', 'skip'): raise ValueError(f"version_mismatch must be 'error' or 'skip', got {version_mismatch!r}") + if missing_index not in ('error', 'skip'): + raise ValueError(f"missing_index must be 'error' or 'skip', got {missing_index!r}") if pc.io_utils.is_remote_path(index_file) and index_fs is None: index_fs = pc.io_utils.get_s3fs(daac=None) @@ -150,7 +222,23 @@ def read_ATL11_granule_cloud_items(s3_url, index_file, xr, yr, fields=None, fs=N # checked against the bucket -- otherwise every granule would be reported # missing and skipped, and the tile would come back empty rather than failing if not pc.io_utils.path_exists(index_file, fs=index_fs, assume_remote_exists=False): - warnings.warn(f'query_ATL11_cloud: missing geoIndex {index_file} for granule {s3_url}, skipping') + # A MISSING INDEX IS A STAGING ERROR, NOT A DATA CONDITION. This used + # to warn and skip, which silently produced a partial fit: a tile whose + # index tree was half-uploaded (there are ~8100 files) came back with + # fewer granules than it should have, and nothing recorded that. With + # find_ATL11_granules(granule_release=...) filtering the search to the + # generation we actually indexed, every granule that reaches here is + # one we expect to have an index for, so its absence is a real fault. + # + # missing_index='skip' restores the old behaviour for a caller that + # genuinely wants best-effort reads. + msg = (f'query_ATL11_cloud: missing geoIndex {index_file} for granule ' + f'{s3_url}. The granule matches the requested release, so its ' + f'index should be staged; reading on would silently return a ' + f'partial fit for this tile.') + if missing_index == 'error': + raise FileNotFoundError(msg) + warnings.warn(msg + ' Skipping (missing_index="skip").') return None s3_basename = os.path.basename(s3_url) @@ -173,7 +261,8 @@ def read_ATL11_granule_cloud_items(s3_url, index_file, xr, yr, fields=None, fs=N def read_ATL11_granule_cloud(s3_url, index_file, xr, yr, fields=None, fs=None, - index_fs=None, version_mismatch='error'): + index_fs=None, version_mismatch='error', + missing_index='error'): """ Read the rows of a cloud ATL11 granule falling within [xr, yr], using a pre-built per-granule geoIndex to locate them, concatenated into a @@ -187,28 +276,35 @@ def read_ATL11_granule_cloud(s3_url, index_file, xr, yr, fields=None, fs=None, """ D = read_ATL11_granule_cloud_items(s3_url, index_file, xr, yr, fields=fields, fs=fs, index_fs=index_fs, - version_mismatch=version_mismatch) + version_mismatch=version_mismatch, + missing_index=missing_index) if D is None: return None return pc.data().from_list(D) def query_ATL11_cloud(bounding_box, index_root, xr, yr, fields=None, - version_mismatch='error', verbose=False): + version_mismatch='error', missing_index='error', + granule_release=None, verbose=False): """ Search Earthdata Cloud for ATL11 granules in bounding_box, and read the rows within [xr, yr] from each, using pre-built per-granule geoIndex files under index_root (see index_path_for_granule() for the expected layout). - version_mismatch is passed through to read_ATL11_granule_cloud(); see - its docstring. + version_mismatch and missing_index are passed through to + read_ATL11_granule_cloud(); see its docstring. + + granule_release restricts the search to one product generation -- see + find_ATL11_granules(). Pass it whenever more than one generation could be + in CMR, which is normally: index_root holds one subtree per generation, and + an unfiltered search would read every subtree that happens to be staged. Returns ------- pointCollection.data containing the concatenated results """ - granules = find_ATL11_granules(bounding_box) + granules = find_ATL11_granules(bounding_box, granule_release=granule_release) fs = pc.io_utils.get_s3fs() if verbose: print(f'query_ATL11_cloud: found {len(granules)} candidate granules') @@ -221,7 +317,8 @@ def query_ATL11_cloud(bounding_box, index_root, xr, yr, fields=None, if verbose: print(f'query_ATL11_cloud: reading {basename}') D = read_ATL11_granule_cloud(s3_url, index_file, xr, yr, fields=fields, - fs=fs, version_mismatch=version_mismatch) + fs=fs, version_mismatch=version_mismatch, + missing_index=missing_index) if D is not None: results.append(D) diff --git a/pointCollection/tilingSchema.py b/pointCollection/tilingSchema.py index 34a5b61..647ab6c 100644 --- a/pointCollection/tilingSchema.py +++ b/pointCollection/tilingSchema.py @@ -396,7 +396,11 @@ def resolve_files_for_box(self, xyr, fs=None, resolution=1.e4, verbose=False): import earthaccess search_kwargs = {k: v for k, v in self.source.items() if k not in ('type', 'daac')} - earthaccess.login(strategy='netrc') + # Best-effort, and NOT strategy='netrc': this is a CMR search, + # which needs no credentials, and a MAAP DPS worker has no ~/.netrc + # to offer. See pc.io_utils.try_earthaccess_login(). + from pointCollection.io_utils import try_earthaccess_login + try_earthaccess_login() granules = earthaccess.search_data(granule_name=candidates, **search_kwargs) found = {} for g in granules: