From b2659df829ffe3b2c2f390927cb2cb9f5581befd Mon Sep 17 00:00:00 2001 From: Vo Date: Tue, 21 Jul 2026 11:01:52 -0700 Subject: [PATCH] Add Chrysalis E3SM v3 archive ingestion --- backend/app/scripts/README.md | 39 +++ .../chrysalis_v3_archive_ingestor.py | 238 ++++++++++++++ .../ingestion/hpc_upload_archive_ingestor.py | 11 +- .../ingestion/nersc_archive_ingestor.py | 71 +++- .../test_chrysalis_v3_archive_ingestor.py | 304 ++++++++++++++++++ docs/architecture/metadata-ingestion.md | 20 ++ docs/deploy/hpc-api-token-authentication.md | 39 ++- 7 files changed, 715 insertions(+), 7 deletions(-) create mode 100644 backend/app/scripts/ingestion/chrysalis_v3_archive_ingestor.py create mode 100644 backend/tests/features/ingestion/test_chrysalis_v3_archive_ingestor.py diff --git a/backend/app/scripts/README.md b/backend/app/scripts/README.md index 61849989..d3932eff 100644 --- a/backend/app/scripts/README.md +++ b/backend/app/scripts/README.md @@ -14,6 +14,7 @@ Scripts are organized by domain: scripts/ ├── ingestion/ │ ├── nersc_archive_ingestor.py +│ ├── chrysalis_v3_archive_ingestor.py │ └── sites/ │ └── nersc.sh ├── db/ @@ -44,6 +45,7 @@ python -m app.scripts.db.seed python -m app.scripts.db.rollback_seed python -m app.scripts.users.create_admin_account python -m app.scripts.ingestion.nersc_archive_ingestor +python -m app.scripts.ingestion.chrysalis_v3_archive_ingestor ``` Do not execute scripts directly by file path: @@ -151,6 +153,43 @@ Archive notes: - `ARCHIVE_YEAR_START` / `ARCHIVE_YEAR_END` are intended for scoped backfills so operators can avoid scanning the full historical tree when unnecessary. - `YYYY` values expand to full-year bounds (`START=2020` means `2020-01`; `END=2020` means `2020-12`), while `YYYY-MM` values target exact archive month buckets. +## Chrysalis E3SM v3 Archive Backfill + +`chrysalis_v3_archive_ingestor.py` is a targeted remote-upload backfill for +simulations stored on LCRC Chrysalis and listed in +the [E3SM v3 simulation table](https://docs.e3sm.org/e3sm_data_docs/_build/html/v3/CoupledSystem/simulation_data/simulation_table.html). +It uses a static copy of the table's `Simulation` values, matches archive case +directory leaf names exactly, forces archive scanning from `2024-01`, and +reuses the HPC upload runner's discovery, validation, deduplication, packaging, +and `/api/v1/ingestions/from-hpc-upload` request logic. + +Run a dry run first: + +```bash +DRY_RUN=true \ +uv run python -m app.scripts.ingestion.chrysalis_v3_archive_ingestor +``` + +Review `v3_case_match`, `v3_case_missing`, and `v3_ingestion_summary` events. +The command exits nonzero when an expected simulation is missing, filesystem +traversal is incomplete, an execution has a transient validation error, or a +live ingestion request fails. Set `DRY_RUN=false` only after every expected +simulation maps to the intended archive case directories. + +This targeted runner deliberately ignores database-backed archive snapshot +checkpoints and never writes new ones. A filtered backfill cannot safely mark a +mixed snapshot complete for the general archive runner. Processed execution +state and immutable discovery results still make repeated runs idempotent. + +Run this module on Chrysalis, where source case directories are readable. It +requires explicit `SIMBOARD_API_BASE_URL` and `SIMBOARD_API_TOKEN` values for an +externally reachable SimBoard deployment, defaults `OLD_PERF_ARCHIVE_ROOT` to +the documented Chrysalis archive root, and records uploads under machine +`chrysalis`. Retry, timeout, case-limit, dry-run, and optional +`ARCHIVE_YEAR_END` variables remain supported. `SCAN_MODE`, +`ARCHIVE_YEAR_START`, and `MACHINE_NAME` are ignored because source site and +scan scope are fixed. + ## HPC Upload Archive Ingestor The HPC upload archive ingestor uses the same scan, state, dry-run, retry, and diff --git a/backend/app/scripts/ingestion/chrysalis_v3_archive_ingestor.py b/backend/app/scripts/ingestion/chrysalis_v3_archive_ingestor.py new file mode 100644 index 00000000..faa6f727 --- /dev/null +++ b/backend/app/scripts/ingestion/chrysalis_v3_archive_ingestor.py @@ -0,0 +1,238 @@ +"""Upload documented E3SM v3 cases from Chrysalis archive snapshots. + +This targeted backfill reuses the remote HPC upload runner while filtering case +directories to simulations documented in the E3SM v3 data table. It packages +each selected case and sends it to ``/api/v1/ingestions/from-hpc-upload``. It +intentionally does not read or write whole-snapshot checkpoints because each +snapshot may also contain non-v3 cases needed by the general archive runner. +""" + +from __future__ import annotations + +import os +import time +from collections import defaultdict +from dataclasses import replace +from pathlib import Path, PurePosixPath + +from app.scripts.ingestion.hpc_upload_archive_ingestor import ( + _run_ingestor as _run_upload_ingestor, +) +from app.scripts.ingestion.nersc_archive_ingestor import ( + IngestorConfig, + IngestorRunReport, + _build_config_from_env, + _log_event, +) + +V3_SIMULATION_TABLE_URL = ( + "https://docs.e3sm.org/e3sm_data_docs/_build/html/v3/" + "CoupledSystem/simulation_data/simulation_table.html" +) +V3_ARCHIVE_YEAR_START = "2024-01" +CHRYSALIS_ARCHIVE_ROOT = "/lcrc/group/e3sm/PERF_Chrysalis/OLD_PERF" +CHRYSALIS_MACHINE_NAME = "chrysalis" + +# Values are copied from the source table's Simulation column. Some RFMIP +# entries include a grouping path; archive case directories use the leaf name. +V3_SIMULATIONS = ( + "v3.LR.piControl", + "v3.LR.abrupt-4xCO2_0101_bcdt15m", + "v3.LR.1pctCO2_0101_bcdt15m", + "v3.LR.historical_0051", + "v3.LR.historical_0101", + "v3.LR.historical_0151", + "v3.LR.historical_0201", + "v3.LR.historical_0251", + "v3.LR.hist-GHG_0101", + "v3.LR.hist-GHG_0151", + "v3.LR.hist-GHG_0201", + "v3.LR.hist-aer_0101", + "v3.LR.hist-aer_0151", + "v3.LR.hist-aer_0201", + "v3.LR.hist-xGHG-xaer_0101", + "v3.LR.hist-xGHG-xaer_0151", + "v3.LR.hist-xGHG-xaer_0201", + "v3.LR.amip_0101", + "v3.LR.amip_0151", + "v3.LR.amip_0201", + "v3.LR.piClim-control-iceini", + "v3.LR.piClim-histall/v3.LR.piClim-histall_0101", + "v3.LR.piClim-histall/v3.LR.piClim-histall_0151", + "v3.LR.piClim-histall/v3.LR.piClim-histall_0201", + "v3.LR.piClim-histGHG/v3.LR.piClim-histGHG_0101", + "v3.LR.piClim-histGHG/v3.LR.piClim-histGHG_0151", + "v3.LR.piClim-histGHG/v3.LR.piClim-histGHG_0201", + "v3.LR.piClim-histaer/v3.LR.piClim-histaer_0101", + "v3.LR.piClim-histaer/v3.LR.piClim-histaer_0151", + "v3.LR.piClim-histaer/v3.LR.piClim-histaer_0201", +) + + +def _case_name(simulation: str) -> str: + """Return archive case-directory name for one documented simulation.""" + return PurePosixPath(simulation).name + + +V3_CASE_NAMES_BY_SIMULATION = { + simulation: _case_name(simulation) for simulation in V3_SIMULATIONS +} +V3_CASE_NAMES = frozenset(V3_CASE_NAMES_BY_SIMULATION.values()) + +if len(V3_CASE_NAMES) != len(V3_SIMULATIONS): + raise RuntimeError("Documented v3 simulations must map to unique case names") + + +def _build_v3_config_from_env() -> IngestorConfig: + """Build Chrysalis config with immutable v3 archive scan scope.""" + if not os.getenv("SIMBOARD_API_BASE_URL", "").strip(): + raise ValueError( + "SIMBOARD_API_BASE_URL is required for remote Chrysalis uploads" + ) + + config = _build_config_from_env( + scan_mode_override="archive", + archive_year_start_override=V3_ARCHIVE_YEAR_START, + ) + return replace( + config, + archive_root=Path( + os.getenv("OLD_PERF_ARCHIVE_ROOT", CHRYSALIS_ARCHIVE_ROOT) + ).resolve(), + machine_name=CHRYSALIS_MACHINE_NAME, + ) + + +def _is_v3_case_path(case_path: Path) -> bool: + """Return whether path exactly matches a documented v3 case name.""" + return case_path.name in V3_CASE_NAMES + + +def _matched_paths_by_case_name( + report: IngestorRunReport, +) -> dict[str, list[str]]: + """Group discovered archive paths by documented leaf case name.""" + matched_paths: defaultdict[str, list[str]] = defaultdict(list) + for case_path in report.case_collection_data: + case_name = Path(case_path).name + if case_name in V3_CASE_NAMES: + matched_paths[case_name].append(case_path) + + return { + case_name: sorted(set(case_paths)) + for case_name, case_paths in matched_paths.items() + } + + +def _log_v3_summary(report: IngestorRunReport, *, dry_run: bool) -> list[str]: + """Log target reconciliation and return missing source simulations.""" + matched_paths = _matched_paths_by_case_name(report) + missing_simulations: list[str] = [] + + for simulation in V3_SIMULATIONS: + case_name = V3_CASE_NAMES_BY_SIMULATION[simulation] + case_paths = matched_paths.get(case_name, []) + if case_paths: + _log_event( + "v3_case_match", + { + "simulation": simulation, + "case_name": case_name, + "case_paths": case_paths, + }, + ) + else: + missing_simulations.append(simulation) + _log_event( + "v3_case_missing", + {"simulation": simulation, "case_name": case_name}, + ) + + stats = report.discovery_stats + _log_event( + "v3_ingestion_summary", + { + "mode": "dry-run" if dry_run else "ingest", + "source_url": V3_SIMULATION_TABLE_URL, + "expected_simulations": len(V3_SIMULATIONS), + "matched_simulations": len(V3_SIMULATIONS) - len(missing_simulations), + "missing_simulations": missing_simulations, + "matching_case_directories": sum( + len(case_paths) for case_paths in matched_paths.values() + ), + "execution_dirs_accepted": ( + 0 if stats is None else stats["execution_dirs_accepted"] + ), + "rejected_existing_execution_ids": ( + 0 if stats is None else stats["rejected_existing_execution_ids"] + ), + "rejected_incomplete_execution_ids": ( + 0 if stats is None else stats["rejected_incomplete_execution_ids"] + ), + "rejected_invalid_execution_ids": ( + 0 if stats is None else stats["rejected_invalid_execution_ids"] + ), + "transient_execution_ids": ( + 0 if stats is None else stats["transient_execution_ids"] + ), + "deferred_execution_ids": ( + 0 if stats is None else stats["deferred_execution_ids"] + ), + "submission_qualified_cases": report.submission_qualified_case_count, + "selected_submission_cases": len(report.candidates), + "ingestion_success_count": report.ingestion_success_count, + "ingestion_failure_count": report.ingestion_failure_count, + }, + ) + return missing_simulations + + +def main() -> int: + """Run targeted v3 archive discovery and remote upload.""" + try: + config = _build_v3_config_from_env() + except ValueError as exc: + _log_event("configuration_error", {"error": str(exc)}) + return 1 + + started_at = time.monotonic() + _log_event( + "v3_run_started", + { + "mode": "dry-run" if config.dry_run else "ingest", + "archive_root": str(config.archive_root), + "archive_year_start": config.archive_year_start, + "source_url": V3_SIMULATION_TABLE_URL, + }, + ) + report = IngestorRunReport() + exit_code = _run_upload_ingestor( + config, + case_path_filter=_is_v3_case_path, + archive_checkpointing=False, + run_report=report, + ) + + if report.scan_completed: + missing_simulations = _log_v3_summary(report, dry_run=config.dry_run) + transient_count = ( + 0 + if report.discovery_stats is None + else report.discovery_stats["transient_execution_ids"] + ) + if missing_simulations or transient_count or not report.traversal_complete: + exit_code = 1 + + _log_event( + "v3_run_finished", + { + "mode": "dry-run" if config.dry_run else "ingest", + "exit_code": exit_code, + "duration_seconds": round(time.monotonic() - started_at, 3), + }, + ) + return exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py b/backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py index 62403811..a359cab4 100644 --- a/backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py +++ b/backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py @@ -27,6 +27,7 @@ IngestionRequestError, IngestionRequestResponse, IngestorConfig, + IngestorRunReport, _build_archive_checkpoints_endpoint_url, _build_config_from_env, _build_discovery_results_endpoint_url, @@ -83,6 +84,9 @@ def _run_ingestor( # noqa: C901 post_request_fn: Callable[..., IngestionRequestResponse] | None = None, discovery_post_request_fn: Callable[..., IngestionRequestResponse] | None = None, checkpoint_post_request_fn: Callable[..., IngestionRequestResponse] | None = None, + case_path_filter: Callable[[Path], bool] | None = None, + archive_checkpointing: bool = True, + run_report: IngestorRunReport | None = None, ) -> int: """Execute one complete archive scan-and-upload cycle.""" if post_request_fn is None: @@ -108,7 +112,7 @@ def _run_ingestor( # noqa: C901 return 1 completed_snapshot_keys: set[str] = set() - if config.scan_mode == "archive": + if config.scan_mode == "archive" and archive_checkpointing: try: completed_snapshot_keys = _fetch_archive_checkpoints( _build_archive_checkpoints_endpoint_url(config), @@ -158,6 +162,8 @@ def _run_ingestor( # noqa: C901 metadata_locator=metadata_locator, discovery_results=new_discovery_results, completed_snapshot_keys=completed_snapshot_keys, + case_path_filter=case_path_filter, + run_report=run_report, ) except Exception as exc: _log_event( @@ -225,8 +231,9 @@ def _run_ingestor( # noqa: C901 discovery_stats, sleep_fn=sleep_fn, post_request_fn=post_request_fn, + run_report=run_report, ) - if config.scan_mode != "archive": + if config.scan_mode != "archive" or not archive_checkpointing: return ingest_exit_code settled_snapshot_keys = _settled_archive_snapshot_keys( diff --git a/backend/app/scripts/ingestion/nersc_archive_ingestor.py b/backend/app/scripts/ingestion/nersc_archive_ingestor.py index a7d56300..3ec43691 100644 --- a/backend/app/scripts/ingestion/nersc_archive_ingestor.py +++ b/backend/app/scripts/ingestion/nersc_archive_ingestor.py @@ -394,6 +394,25 @@ class ArchiveSnapshotScan: traversal_complete: bool = True +@dataclass +class IngestorRunReport: + """Mutable result details exposed to specialized archive runners. + + For example, a targeted archive ingestor uses this to log a summary of + discovered cases and ingestion outcomes. + """ + + scan_completed: bool = False + traversal_complete: bool = True + scan_results: list[CaseScanResult] = field(default_factory=list) + candidates: list[IngestionCandidate] = field(default_factory=list) + submission_qualified_case_count: int = 0 + discovery_stats: DiscoveryStats | None = None + case_collection_data: dict[str, CaseCollectionLogData] = field(default_factory=dict) + ingestion_success_count: int = 0 + ingestion_failure_count: int = 0 + + # Entrypoint and Configuration # ---------------------------- @@ -435,7 +454,11 @@ def main() -> int: return exit_code -def _build_config_from_env() -> IngestorConfig: +def _build_config_from_env( + *, + scan_mode_override: Literal["staging", "archive"] | None = None, + archive_year_start_override: str | None = None, +) -> IngestorConfig: """Build and validate runtime config from environment variables. Returns @@ -451,7 +474,11 @@ def _build_config_from_env() -> IngestorConfig: api_base_url = os.getenv("SIMBOARD_API_BASE_URL", DEFAULT_API_BASE_URL) api_token = os.getenv("SIMBOARD_API_TOKEN", "") - scan_mode = os.getenv("SCAN_MODE", "staging").strip().lower() + scan_mode = ( + scan_mode_override + if scan_mode_override is not None + else os.getenv("SCAN_MODE", "staging").strip().lower() + ) if scan_mode not in ARCHIVE_SCAN_MODES: raise ValueError("SCAN_MODE must be either 'staging' or 'archive'") @@ -481,7 +508,11 @@ def _build_config_from_env() -> IngestorConfig: raise ValueError("REQUEST_TIMEOUT_SECONDS must be greater than 0") archive_year_start = _parse_optional_archive_bound( - os.getenv("ARCHIVE_YEAR_START"), + ( + archive_year_start_override + if archive_year_start_override is not None + else os.getenv("ARCHIVE_YEAR_START") + ), env_name="ARCHIVE_YEAR_START", is_end_bound=False, ) @@ -833,6 +864,8 @@ def _scan_archive( metadata_locator: Callable[[str], object], discovery_results: list[ExecutionDiscoveryResult] | None = None, completed_snapshot_keys: set[str] | None = None, + case_path_filter: Callable[[Path], bool] | None = None, + run_report: IngestorRunReport | None = None, ) -> tuple[ list[CaseScanResult], list[IngestionCandidate], @@ -865,7 +898,10 @@ def _scan_archive( staging_root_basename = ( config.archive_root.name or Path(DEFAULT_PERF_ARCHIVE_ROOT).name ) - case_path_filter = _build_case_path_filter(config) + case_path_filter = _combine_case_path_filters( + _build_case_path_filter(config), + case_path_filter, + ) snapshot_scan = ArchiveSnapshotScan(archive_name=config.archive_root.name) if config.scan_mode == "archive": snapshot_scan.eligible_keys = _enumerate_archive_snapshot_keys(config) @@ -944,6 +980,15 @@ def handle_walk_error(exc: OSError) -> None: staging_root_basename=staging_root_basename, ) + if run_report is not None: + run_report.scan_completed = True + run_report.traversal_complete = snapshot_scan.traversal_complete + run_report.scan_results = list(scan_results) + run_report.candidates = list(candidates) + run_report.submission_qualified_case_count = len(all_candidates) + run_report.discovery_stats = discovery_stats.copy() + run_report.case_collection_data = dict(case_collection_data) + return ( scan_results, candidates, @@ -1581,6 +1626,19 @@ def _build_case_path_filter( ) +def _combine_case_path_filters( + *filters: Callable[[Path], bool] | None, +) -> Callable[[Path], bool] | None: + """Combine optional case predicates without changing unfiltered behavior.""" + active_filters = tuple(case_filter for case_filter in filters if case_filter) + if not active_filters: + return None + + return lambda case_path: all( + case_filter(case_path) for case_filter in active_filters + ) + + def _build_walk_dir_filter( config: IngestorConfig, *, @@ -2231,6 +2289,7 @@ def _handle_ingest_run( discovery_stats: DiscoveryStats, sleep_fn: Callable[[float], None], post_request_fn: Callable[..., IngestionRequestResponse], + run_report: IngestorRunReport | None = None, ) -> int: """Execute candidate ingestion loop and emit completion summaries. @@ -2343,6 +2402,10 @@ def _handle_ingest_run( discovery_stats=discovery_stats, ) + if run_report is not None: + run_report.ingestion_success_count = success_count + run_report.ingestion_failure_count = failure_count + return 1 if failure_count else 0 diff --git a/backend/tests/features/ingestion/test_chrysalis_v3_archive_ingestor.py b/backend/tests/features/ingestion/test_chrysalis_v3_archive_ingestor.py new file mode 100644 index 00000000..de12b64f --- /dev/null +++ b/backend/tests/features/ingestion/test_chrysalis_v3_archive_ingestor.py @@ -0,0 +1,304 @@ +"""Tests for targeted Chrysalis E3SM v3 archive uploads.""" + +import json +import urllib.request +from pathlib import Path +from typing import Any + +from app.scripts.ingestion import chrysalis_v3_archive_ingestor as v3_ingestor +from app.scripts.ingestion import hpc_upload_archive_ingestor as upload_ingestor +from app.scripts.ingestion import nersc_archive_ingestor as base_ingestor +from app.scripts.ingestion.nersc_archive_ingestor import ( + CaseCollectionLogData, + IngestionRequestResponse, + IngestorConfig, + IngestorRunReport, + _fresh_state, +) + + +def _config(archive_root: Path, *, dry_run: bool) -> IngestorConfig: + return IngestorConfig( + api_base_url="https://simboard.example", + api_token="token", + archive_root=archive_root, + machine_name="chrysalis", + dry_run=dry_run, + max_cases_per_run=None, + max_attempts=1, + request_timeout_seconds=30, + scan_mode="archive", + archive_year_start="2024-01", + ) + + +def _populate_complete_report(report: IngestorRunReport) -> None: + report.scan_completed = True + report.discovery_stats = base_ingestor._new_discovery_stats() + report.case_collection_data = { + f"/lcrc/OLD_PERF/2024-01/snapshot/COMPLETED/user/{case_name}": ( + CaseCollectionLogData(case_path=case_name, execution_count_total=1) + ) + for case_name in v3_ingestor.V3_CASE_NAMES + } + + +class _FakeHttpResponse: + status = 201 + + def read(self) -> bytes: + return json.dumps( + {"created_count": 1, "duplicate_count": 0, "errors": []} + ).encode() + + def __enter__(self) -> "_FakeHttpResponse": + return self + + def __exit__(self, *args: Any) -> None: + return None + + +def test_documented_simulations_normalize_to_unique_case_names() -> None: + assert len(v3_ingestor.V3_CASE_NAMES) == len(v3_ingestor.V3_SIMULATIONS) + assert ( + v3_ingestor.V3_CASE_NAMES_BY_SIMULATION[ + "v3.LR.piClim-histall/v3.LR.piClim-histall_0101" + ] + == "v3.LR.piClim-histall_0101" + ) + + +def test_v3_case_filter_requires_exact_leaf_name() -> None: + assert v3_ingestor._is_v3_case_path(Path("/archive/v3.LR.piControl")) + assert not v3_ingestor._is_v3_case_path(Path("/archive/prefix-v3.LR.piControl")) + assert not v3_ingestor._is_v3_case_path(Path("/archive/v3.LR.piControl-extra")) + + +def test_v3_config_forces_archive_mode_and_2024_lower_bound( + tmp_path: Path, monkeypatch +) -> None: + monkeypatch.setenv("SCAN_MODE", "staging") + monkeypatch.setenv("ARCHIVE_YEAR_START", "2023-01") + monkeypatch.setenv("SIMBOARD_API_BASE_URL", "https://simboard.example") + monkeypatch.setenv("OLD_PERF_ARCHIVE_ROOT", str(tmp_path / "OLD_PERF")) + + config = v3_ingestor._build_v3_config_from_env() + + assert config.scan_mode == "archive" + assert config.archive_root == (tmp_path / "OLD_PERF").resolve() + assert config.archive_year_start == "2024-01" + assert config.machine_name == "chrysalis" + + +def test_v3_config_requires_remote_api_url(monkeypatch) -> None: + monkeypatch.delenv("SIMBOARD_API_BASE_URL", raising=False) + + try: + v3_ingestor._build_v3_config_from_env() + except ValueError as exc: + assert str(exc) == ( + "SIMBOARD_API_BASE_URL is required for remote Chrysalis uploads" + ) + else: + raise AssertionError("missing remote API URL must fail configuration") + + +def test_v3_summary_reports_paths_missing_and_execution_outcomes( + monkeypatch, +) -> None: + report = IngestorRunReport(scan_completed=True) + stats = base_ingestor._new_discovery_stats() + stats["execution_dirs_accepted"] = 4 + stats["rejected_existing_execution_ids"] = 3 + stats["rejected_incomplete_execution_ids"] = 2 + stats["rejected_invalid_execution_ids"] = 1 + stats["transient_execution_ids"] = 5 + stats["deferred_execution_ids"] = 6 + report.discovery_stats = stats + matched_case_name = "v3.LR.piClim-histall_0101" + first_path = f"/lcrc/OLD_PERF/2024-01/snapshot-a/COMPLETED/user/{matched_case_name}" + second_path = ( + f"/lcrc/OLD_PERF/2024-02/snapshot-b/COMPLETED/user/{matched_case_name}" + ) + report.case_collection_data = { + first_path: CaseCollectionLogData(case_path=first_path), + second_path: CaseCollectionLogData(case_path=second_path), + } + logged_events: list[tuple[str, dict[str, Any]]] = [] + monkeypatch.setattr( + v3_ingestor, + "_log_event", + lambda event, fields=None: logged_events.append((event, fields or {})), + ) + + missing = v3_ingestor._log_v3_summary(report, dry_run=True) + + match_event = next( + fields for event, fields in logged_events if event == "v3_case_match" + ) + summary = next( + fields for event, fields in logged_events if event == "v3_ingestion_summary" + ) + assert match_event["case_paths"] == [first_path, second_path] + assert len(missing) == len(v3_ingestor.V3_SIMULATIONS) - 1 + assert summary["execution_dirs_accepted"] == 4 + assert summary["rejected_existing_execution_ids"] == 3 + assert summary["rejected_incomplete_execution_ids"] == 2 + assert summary["rejected_invalid_execution_ids"] == 1 + assert summary["transient_execution_ids"] == 5 + assert summary["deferred_execution_ids"] == 6 + + +def test_targeted_archive_run_filters_cases_and_skips_all_checkpoints( + tmp_path: Path, monkeypatch +) -> None: + archive_root = tmp_path / "OLD_PERF" + snapshot = ( + archive_root + / "2024-01" + / "performance_archive_2024_01_01_00_00_00" + / "COMPLETED" + / "user" + ) + v3_execution = snapshot / "v3.LR.piControl" / "100.1-1" + unrelated_execution = snapshot / "unrelated-case" / "200.1-1" + old_v3_execution = ( + archive_root + / "2023-12" + / "performance_archive_2023_12_31_00_00_00" + / "COMPLETED" + / "user" + / "v3.LR.piControl" + / "300.1-1" + ) + v3_execution.mkdir(parents=True) + unrelated_execution.mkdir(parents=True) + old_v3_execution.mkdir(parents=True) + validated: list[str] = [] + captured_requests: list[urllib.request.Request] = [] + + monkeypatch.setattr( + upload_ingestor, + "_fetch_ingestion_state", + lambda *args, **kwargs: _fresh_state(), + ) + + def fail_checkpoint(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("targeted runner must not use archive checkpoints") + + monkeypatch.setattr(upload_ingestor, "_fetch_archive_checkpoints", fail_checkpoint) + + def post_discovery(*args: Any, **kwargs: Any) -> IngestionRequestResponse: + return {"status_code": 201, "body": {}} + + def fake_urlopen(request: urllib.request.Request, timeout: int): + captured_requests.append(request) + assert timeout == 30 + return _FakeHttpResponse() + + monkeypatch.setattr(upload_ingestor.urllib.request, "urlopen", fake_urlopen) + + report = IngestorRunReport() + exit_code = upload_ingestor._run_ingestor( + _config(archive_root, dry_run=False), + metadata_locator=lambda path: validated.append(path), + discovery_post_request_fn=post_discovery, + checkpoint_post_request_fn=fail_checkpoint, + case_path_filter=v3_ingestor._is_v3_case_path, + archive_checkpointing=False, + run_report=report, + ) + + assert exit_code == 0 + assert validated == [str(v3_execution)] + assert len(captured_requests) == 1 + request = captured_requests[0] + assert request.full_url.endswith("/api/v1/ingestions/from-hpc-upload") + assert request.headers["Content-type"].startswith("multipart/form-data;") + assert isinstance(request.data, bytes) + assert b'name="machine_name"\r\n\r\nchrysalis' in request.data + assert str(v3_execution.parent).encode() in request.data + assert b'filename="v3.LR.piControl-' in request.data + assert b"unrelated-case" not in request.data + assert b"300.1-1" not in request.data + assert set(report.case_collection_data) == {str(v3_execution.parent)} + + +def test_targeted_dry_run_never_calls_write_functions( + tmp_path: Path, monkeypatch +) -> None: + archive_root = tmp_path / "OLD_PERF" + execution = ( + archive_root + / "2024-01" + / "performance_archive_2024_01_01_00_00_00" + / "COMPLETED" + / "user" + / "v3.LR.piControl" + / "100.1-1" + ) + execution.mkdir(parents=True) + monkeypatch.setattr( + upload_ingestor, + "_fetch_ingestion_state", + lambda *args, **kwargs: _fresh_state(), + ) + + def fail_write(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("dry run must not write") + + exit_code = upload_ingestor._run_ingestor( + _config(archive_root, dry_run=True), + metadata_locator=lambda *_: {}, + post_request_fn=fail_write, + discovery_post_request_fn=fail_write, + checkpoint_post_request_fn=fail_write, + case_path_filter=v3_ingestor._is_v3_case_path, + archive_checkpointing=False, + ) + + assert exit_code == 0 + + +def test_v3_main_disables_checkpoints_and_succeeds_when_all_cases_match( + tmp_path: Path, monkeypatch +) -> None: + config = _config(tmp_path, dry_run=True) + captured_kwargs: dict[str, Any] = {} + logged_events: list[tuple[str, dict[str, Any]]] = [] + monkeypatch.setattr(v3_ingestor, "_build_v3_config_from_env", lambda: config) + monkeypatch.setattr( + v3_ingestor, + "_log_event", + lambda event, fields=None: logged_events.append((event, fields or {})), + ) + + def fake_run(config: IngestorConfig, **kwargs: Any) -> int: + captured_kwargs.update(kwargs) + _populate_complete_report(kwargs["run_report"]) + return 0 + + monkeypatch.setattr(v3_ingestor, "_run_upload_ingestor", fake_run) + + assert v3_ingestor.main() == 0 + assert captured_kwargs["archive_checkpointing"] is False + assert captured_kwargs["case_path_filter"] is v3_ingestor._is_v3_case_path + assert any(event == "v3_ingestion_summary" for event, _ in logged_events) + + +def test_v3_main_fails_reconciliation_when_case_is_missing( + tmp_path: Path, monkeypatch +) -> None: + config = _config(tmp_path, dry_run=True) + monkeypatch.setattr(v3_ingestor, "_build_v3_config_from_env", lambda: config) + monkeypatch.setattr(v3_ingestor, "_log_event", lambda *args, **kwargs: None) + + def fake_run(config: IngestorConfig, **kwargs: Any) -> int: + report = kwargs["run_report"] + _populate_complete_report(report) + report.case_collection_data.pop(next(iter(report.case_collection_data))) + return 0 + + monkeypatch.setattr(v3_ingestor, "_run_upload_ingestor", fake_run) + + assert v3_ingestor.main() == 1 diff --git a/docs/architecture/metadata-ingestion.md b/docs/architecture/metadata-ingestion.md index c3f9bdec..5e74494f 100644 --- a/docs/architecture/metadata-ingestion.md +++ b/docs/architecture/metadata-ingestion.md @@ -174,6 +174,7 @@ Example NERSC path for `COMPLETED` status cases: Automated HPC collection reaches SimBoard ingestion through two site-side submission modes. Both use database-backed stored known execution IDs, but they submit submission-qualified cases through different routes: - `nersc_archive_ingestor.py` for local path submission on NERSC / Perlmutter +- `chrysalis_v3_archive_ingestor.py` for the targeted Chrysalis E3SM v3 remote-upload backfill - `hpc_upload_archive_ingestor.py` for remote automated archive upload from LCRC and other DOE sites | Mode | Script / entry point | Access pattern | Route | Use when | Examples | @@ -182,6 +183,21 @@ Automated HPC collection reaches SimBoard ingestion through two site-side submis | Remote automated archive upload | `hpc_upload_archive_ingestor.py` | Site job uploads one submission-qualified case archive over HTTPS. | `/api/v1/ingestions/from-hpc-upload` | Source archive is not readable from NERSC Spin. | LCRC / Chrysalis; other DOE sites | | Browser/manual upload | N/A | User uploads an archive through the browser. | `/api/v1/ingestions/from-upload` | Manual, test, or ad hoc ingestion is needed. | User workstation | +The v3 backfill is a specialization of remote automated archive upload, not +another API mode. It runs on Chrysalis, statically defines simulations from the +E3SM v3 data table, converts grouped table values to archive leaf case names, +and exact-matches those names while scanning Chrysalis archive snapshots from +`2024-01`. Each submission-qualified case is packaged and sent to +`/api/v1/ingestions/from-hpc-upload`. Reconciliation logs map every expected +simulation to matching case directories and report missing, accepted, already +processed, incomplete, invalid, transient, deferred, and submission outcomes. + +Targeted v3 scans do not read or write archive snapshot checkpoints. Snapshot +checkpoints describe completion of every execution in a snapshot, so a +case-filtered scan must not mark a mixed snapshot complete or let prior general +checkpoints hide targets. The runner still reads processed execution state and +immutable discovery results, preserving submission idempotency. + ### Automated Submission-State Flow Both automated scripts follow the same submission-state sequence. In archive @@ -214,6 +230,10 @@ again. Completed archive snapshots are skipped before their contents are walked. Dry runs compute and log proposed results but never persist discovery, processed state, or archive checkpoints. +The targeted v3 runner is the exception to completed-snapshot pruning: it scans +all eligible snapshots within its fixed lower bound because checkpoint state is +intentionally disabled for filtered reconciliation. + Remote automated uploads must contain exactly one case directory per request. The submitted `case_path` is used as the stable case identifier for that uploaded case. ```mermaid diff --git a/docs/deploy/hpc-api-token-authentication.md b/docs/deploy/hpc-api-token-authentication.md index 5e1665d2..c08cd45c 100644 --- a/docs/deploy/hpc-api-token-authentication.md +++ b/docs/deploy/hpc-api-token-authentication.md @@ -80,13 +80,50 @@ curl -X POST https://api.simboard.org/api/v1/ingestions/from-path \ curl -X POST https://api.simboard.org/api/v1/ingestions/from-hpc-upload \ -H "Authorization: Bearer sbk_xxxxxxxxxxxxxxxxxxxxx" \ -F "file=@case-a.tar.gz" \ - -F "machine_name=perlmutter" \ + -F "machine_name=chrysalis" \ -F "case_path=/lcrc/group/e3sm/PERF_Chrysalis/performance_archive/case_a" \ -F "processed_execution_ids=100.1-1" \ -F "processed_execution_ids=101.1-1" \ -F "hpc_username=johndoe" ``` +#### Chrysalis E3SM v3 archive backfill + +Run the targeted v3 backfill on Chrysalis because source case directories are +not mounted in SimBoard's NERSC backend. The runner scans archive snapshots from +`2024-01`, packages each selected case as a single-case archive, and uploads it +through `/api/v1/ingestions/from-hpc-upload`. + +From `backend/` on Chrysalis, provide an externally reachable SimBoard API URL +and service-account token, then start with dry run: + +```bash +SIMBOARD_API_BASE_URL=https:// \ +SIMBOARD_API_TOKEN= \ +DRY_RUN=true \ +uv run python -m app.scripts.ingestion.chrysalis_v3_archive_ingestor +``` + +`OLD_PERF_ARCHIVE_ROOT` defaults to the documented Chrysalis archive location +and may be overridden when site storage is mounted elsewhere. Machine identity +is fixed to `chrysalis`; archive mode and the `2024-01` lower bound are also +fixed by the targeted runner. + +Review `v3_case_match`, `v3_case_missing`, and `v3_ingestion_summary`. Resolve +missing targets and transient scan errors before enabling uploads. Then run: + +```bash +SIMBOARD_API_BASE_URL=https:// \ +SIMBOARD_API_TOKEN= \ +DRY_RUN=false \ +uv run python -m app.scripts.ingestion.chrysalis_v3_archive_ingestor +``` + +Repeat dry run after upload to confirm processed execution state prevents +duplicate submissions. Targeted scans deliberately neither read nor write +whole-snapshot checkpoints because Chrysalis snapshots may also contain +non-v3 cases. + #### Browser or Manual Upload ```bash