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: