diff --git a/backend/app/scripts/README.md b/backend/app/scripts/README.md index 58839f68..082ad006 100644 --- a/backend/app/scripts/README.md +++ b/backend/app/scripts/README.md @@ -20,8 +20,13 @@ scripts/ │ ├── archive_workflow.py │ ├── hpc_upload_archive_ingestor.py │ ├── nersc_archive_ingestor.py -│ └── sites/ +│ ├── sites/ │ └── nersc.sh +│ └── v3_data/ +│ ├── __init__.py +│ ├── lcrc-v3.env.example +│ ├── lcrc_v3.sh +│ └── lcrc_v3_archive_ingestor.py ├── db/ │ ├── seed.py │ ├── rollback_seed.py @@ -50,6 +55,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.v3_data.lcrc_v3_archive_ingestor ``` Do not execute scripts directly by file path: @@ -157,6 +163,53 @@ 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. +## One-Time Chrysalis E3SM v3 Archive Backfill + +`v3_data/lcrc_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. + +For this one-time backfill, copy the committed template outside the repository, +secure it, replace its placeholders, then run a dry run: + +```bash +mkdir -p ~/.config/simboard +cp app/scripts/ingestion/v3_data/lcrc-v3.env.example ~/.config/simboard/lcrc-v3.env +chmod 600 ~/.config/simboard/lcrc-v3.env +# Edit ~/.config/simboard/lcrc-v3.env to replace placeholders. +LCRC_V3_ENV_FILE=~/.config/simboard/lcrc-v3.env \ + ./app/scripts/ingestion/v3_data/lcrc_v3.sh +``` + +`backend/app/scripts/ingestion/v3_data/lcrc_v3.sh` sources the selected +environment file, requires `SIMBOARD_API_BASE_URL` and `SIMBOARD_API_TOKEN`, +and defaults the LCRC archive root and dry-run mode. Set the optional +`OLD_PERF_ARCHIVE_ROOT` in that file only when storage is mounted elsewhere. + +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/archive_discovery.py b/backend/app/scripts/ingestion/archive_discovery.py index 44395696..aee447d0 100644 --- a/backend/app/scripts/ingestion/archive_discovery.py +++ b/backend/app/scripts/ingestion/archive_discovery.py @@ -27,6 +27,7 @@ ExecutionDiscoveryResult, IngestionCandidate, IngestorConfig, + IngestorRunReport, MetadataLocator, _build_discovery_results_by_key, _case_log_label, @@ -61,12 +62,42 @@ class _CaseCollectionOutcome: decisions_by_execution_id: dict[str, ExecutionCollectionDecision] +def _combine_case_path_filters( + configured_filter: Callable[[Path], bool] | None, + supplied_filter: Callable[[Path], bool] | None, +) -> Callable[[Path], bool] | None: + """Return the intersection of configured and caller-supplied path filters.""" + if supplied_filter is None: + return configured_filter + if configured_filter is None: + return supplied_filter + + return partial( + _matches_combined_case_path_filters, + configured_filter=configured_filter, + supplied_filter=supplied_filter, + ) + + +def _matches_combined_case_path_filters( + path: Path, + *, + configured_filter: Callable[[Path], bool], + supplied_filter: Callable[[Path], bool], +) -> bool: + """Return whether a path passes both case path filters.""" + return configured_filter(path) and supplied_filter(path) + + def _scan_archive( config: IngestorConfig, state: dict[str, Any], metadata_locator: MetadataLocator, discovery_results: list[ExecutionDiscoveryResult] | None = None, completed_snapshot_keys: set[str] | None = None, + case_path_filter: Callable[[Path], bool] | None = None, + additional_dir_pruner: Callable[[str, list[str]], None] | None = None, + run_report: IngestorRunReport | None = None, ) -> tuple[ list[CaseScanResult], list[IngestionCandidate], @@ -99,7 +130,11 @@ 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 = _initialize_snapshot_scan(config, completed_snapshot_keys) selected_snapshot_keys = _selected_snapshot_keys(snapshot_scan) walk_dir_filter = _build_walk_dir_filter( @@ -123,6 +158,7 @@ def _scan_archive( case_collection_data, case_path_filter=case_path_filter, walk_dir_filter=walk_dir_filter, + additional_dir_pruner=additional_dir_pruner, scan_mode=config.scan_mode, processed_ids_by_key=processed_ids_by_key, discovery_results=discovery_results, @@ -154,6 +190,15 @@ def _scan_archive( 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 = scan_results + run_report.candidates = candidates + run_report.submission_qualified_case_count = len(all_candidates) + run_report.discovery_stats = discovery_stats + run_report.case_collection_data = case_collection_data + return ( scan_results, candidates, @@ -238,6 +283,7 @@ def _discover_case_executions( *, case_path_filter: Callable[[Path], bool] | None = None, walk_dir_filter: Callable[[str, list[str]], None] | None = None, + additional_dir_pruner: Callable[[str, list[str]], None] | None = None, scan_mode: str = "staging", processed_ids_by_key: defaultdict[str, set[str]] | None = None, staging_root_basename: str = Path(DEFAULT_PERF_ARCHIVE_ROOT).name, @@ -287,6 +333,10 @@ def _discover_case_executions( if walk_dir_filter is not None: walk_dir_filter(dirpath, dirnames) + # Apply specialized pruning only after generic archive layout pruning. + if additional_dir_pruner is not None: + additional_dir_pruner(dirpath, dirnames) + case_dir = Path(dirpath) for dirname in list(dirnames): diff --git a/backend/app/scripts/ingestion/archive_ingestor_core.py b/backend/app/scripts/ingestion/archive_ingestor_core.py index d076bf3a..7fa53416 100644 --- a/backend/app/scripts/ingestion/archive_ingestor_core.py +++ b/backend/app/scripts/ingestion/archive_ingestor_core.py @@ -366,6 +366,21 @@ class CaseCollectionLogData: rejected_decisions: list[ExecutionCollectionDecision] = field(default_factory=list) +@dataclass +class IngestorRunReport: + """Mutable details from a specialized archive-ingestion run.""" + + 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 + + @dataclass class ArchiveSnapshotScan: """Filesystem snapshot units and execution identities scanned in this run.""" @@ -450,7 +465,11 @@ def __call__( # ------------- -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 @@ -466,7 +485,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", DEFAULT_SCAN_MODE).strip().lower() + scan_mode = ( + scan_mode_override + if scan_mode_override is not None + else os.getenv("SCAN_MODE", DEFAULT_SCAN_MODE).strip().lower() + ) if scan_mode not in ARCHIVE_SCAN_MODES: raise ValueError("SCAN_MODE must be either 'staging' or 'archive'") @@ -499,7 +522,9 @@ 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, ) diff --git a/backend/app/scripts/ingestion/archive_workflow.py b/backend/app/scripts/ingestion/archive_workflow.py index d26c6e61..cdaf54e5 100644 --- a/backend/app/scripts/ingestion/archive_workflow.py +++ b/backend/app/scripts/ingestion/archive_workflow.py @@ -25,6 +25,7 @@ IngestionCandidate, IngestionRequestError, IngestorConfig, + IngestorRunReport, SleepCallback, StructuredLogCallback, _case_log_label, @@ -250,6 +251,7 @@ def _handle_ingest_run( Callable[[IngestionCandidate], AbstractContextManager[CaseSubmissionCallback]] | None ) = None, + run_report: IngestorRunReport | None = None, ) -> int: """Execute candidate ingestion loop and emit completion summaries.""" log_event_fn = log_event_fn or _log_event @@ -344,6 +346,10 @@ def _handle_ingest_run( log_event_fn=log_event_fn, ) + 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/app/scripts/ingestion/hpc_upload_archive_ingestor.py b/backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py index 9e87be65..73c9a475 100644 --- a/backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py +++ b/backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py @@ -33,6 +33,7 @@ from collections.abc import Iterator from contextlib import contextmanager from pathlib import Path +from typing import Callable from app.features.ingestion.parsers.parser import _locate_metadata_files from app.scripts.ingestion.archive_client import ( @@ -58,6 +59,7 @@ IngestionRequestError, IngestionRequestResponse, IngestorConfig, + IngestorRunReport, MetadataLocator, SleepCallback, _build_config_from_env, @@ -122,6 +124,10 @@ def _run_ingestor( post_request_fn: CaseSubmissionCallback | None = None, discovery_post_request_fn: DiscoveryResultsPersistenceCallback | None = None, checkpoint_post_request_fn: ArchiveCheckpointPersistenceCallback | None = None, + case_path_filter: Callable[[Path], bool] | None = None, + additional_dir_pruner: Callable[[str, list[str]], None] | None = None, + archive_checkpointing: bool = True, + run_report: IngestorRunReport | None = None, ) -> int: """Execute one complete archive scan-and-upload cycle.""" use_prepared_archives = post_request_fn is None @@ -140,7 +146,7 @@ def _run_ingestor( 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), @@ -190,6 +196,9 @@ def _run_ingestor( metadata_locator=metadata_locator, discovery_results=new_discovery_results, completed_snapshot_keys=completed_snapshot_keys, + case_path_filter=case_path_filter, + additional_dir_pruner=additional_dir_pruner, + run_report=run_report, ) except Exception as exc: _log_event( @@ -246,7 +255,12 @@ def _run_ingestor( if use_prepared_archives else None ), + run_report=run_report, ) + + if not archive_checkpointing: + return ingest_exit_code + if not _finalize_archive_checkpoints( snapshot_scan, state, diff --git a/backend/app/scripts/ingestion/v3_data/__init__.py b/backend/app/scripts/ingestion/v3_data/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/app/scripts/ingestion/v3_data/lcrc-v3.env.example b/backend/app/scripts/ingestion/v3_data/lcrc-v3.env.example new file mode 100644 index 00000000..e530eddf --- /dev/null +++ b/backend/app/scripts/ingestion/v3_data/lcrc-v3.env.example @@ -0,0 +1,7 @@ +# Copy this file outside the repository, replace placeholders, and chmod it 600. +SIMBOARD_API_BASE_URL=https:// +SIMBOARD_API_TOKEN= +DRY_RUN=true + +# Optional: override the default /lcrc/group/e3sm/PERF_Chrysalis/OLD_PERF root. +# OLD_PERF_ARCHIVE_ROOT=/path/to/OLD_PERF diff --git a/backend/app/scripts/ingestion/v3_data/lcrc_v3.sh b/backend/app/scripts/ingestion/v3_data/lcrc_v3.sh new file mode 100755 index 00000000..a0ce217e --- /dev/null +++ b/backend/app/scripts/ingestion/v3_data/lcrc_v3.sh @@ -0,0 +1,34 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +INGESTION_DIR="$(cd -- "${SCRIPT_DIR}/.." && pwd)" +BACKEND_DIR="$(cd -- "${INGESTION_DIR}/../../.." && pwd)" +PYTHON_BIN="${PYTHON_BIN:-${BACKEND_DIR}/.venv/bin/python}" + +if [[ ! -x "${PYTHON_BIN}" ]]; then + echo "Expected Python interpreter at ${PYTHON_BIN}" >&2 + echo "Run 'make install' from the repository root to create it." >&2 + exit 1 +fi + +: "${LCRC_V3_ENV_FILE:?LCRC_V3_ENV_FILE must identify a readable environment file.}" +if [[ ! -f "${LCRC_V3_ENV_FILE}" || ! -r "${LCRC_V3_ENV_FILE}" ]]; then + echo "LCRC_V3_ENV_FILE must identify a readable file: ${LCRC_V3_ENV_FILE}" >&2 + exit 1 +fi + +set -a +# shellcheck source=/dev/null +source "${LCRC_V3_ENV_FILE}" +set +a + +: "${SIMBOARD_API_TOKEN:?SIMBOARD_API_TOKEN must be set in LCRC_V3_ENV_FILE.}" +: "${SIMBOARD_API_BASE_URL:?SIMBOARD_API_BASE_URL must be set in LCRC_V3_ENV_FILE.}" + +export DRY_RUN="${DRY_RUN:-true}" +export OLD_PERF_ARCHIVE_ROOT="${OLD_PERF_ARCHIVE_ROOT:-/lcrc/group/e3sm/PERF_Chrysalis/OLD_PERF}" + +cd "${BACKEND_DIR}" +exec "${PYTHON_BIN}" -m app.scripts.ingestion.v3_data.lcrc_v3_archive_ingestor "$@" diff --git a/backend/app/scripts/ingestion/v3_data/lcrc_v3_archive_ingestor.py b/backend/app/scripts/ingestion/v3_data/lcrc_v3_archive_ingestor.py new file mode 100644 index 00000000..c32e6d2d --- /dev/null +++ b/backend/app/scripts/ingestion/v3_data/lcrc_v3_archive_ingestor.py @@ -0,0 +1,255 @@ +"""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 functools import partial +from pathlib import Path + +from app.scripts.ingestion.archive_ingestor_core import ( + IngestorConfig, + IngestorRunReport, + _build_config_from_env, + _log_event, +) +from app.scripts.ingestion.archive_layout import ( + ARCHIVE_COMPLETED_STATUS_DIR_NAME, + _archive_dir_bucket, + _is_archive_snapshot_dir, +) +from app.scripts.ingestion.hpc_upload_archive_ingestor import ( + _run_ingestor as _run_upload_ingestor, +) + +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" + +# Normalized archive case names from the source table's Simulation column. +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_0101", + "v3.LR.piClim-histall_0151", + "v3.LR.piClim-histall_0201", + "v3.LR.piClim-histGHG_0101", + "v3.LR.piClim-histGHG_0151", + "v3.LR.piClim-histGHG_0201", + "v3.LR.piClim-histaer_0101", + "v3.LR.piClim-histaer_0151", + "v3.LR.piClim-histaer_0201", +) + + +V3_CASE_NAMES_BY_SIMULATION = {simulation: 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 _prune_v3_case_directories( + dirpath: str, + dirnames: list[str], + *, + archive_root: Path, +) -> None: + """Prune non-v3 case names only beneath proven archive user directories.""" + try: + relative_parts = Path(dirpath).relative_to(archive_root).parts + except ValueError: + return + + if ( + len(relative_parts) != 4 + or _archive_dir_bucket(relative_parts[0]) is None + or not _is_archive_snapshot_dir(relative_parts[1]) + or relative_parts[2] != ARCHIVE_COMPLETED_STATUS_DIR_NAME + ): + return + + dirnames[:] = [dirname for dirname in dirnames if dirname 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 or {} + _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": stats.get("execution_dirs_accepted", 0), + "rejected_existing_execution_ids": stats.get( + "rejected_existing_execution_ids", 0 + ), + "rejected_incomplete_execution_ids": stats.get( + "rejected_incomplete_execution_ids", 0 + ), + "rejected_invalid_execution_ids": stats.get( + "rejected_invalid_execution_ids", 0 + ), + "transient_execution_ids": stats.get("transient_execution_ids", 0), + "deferred_execution_ids": stats.get("deferred_execution_ids", 0), + "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, + additional_dir_pruner=partial( + _prune_v3_case_directories, + archive_root=config.archive_root.resolve(), + ), + archive_checkpointing=False, + run_report=report, + ) + + if report.scan_completed: + missing_simulations = _log_v3_summary(report, dry_run=config.dry_run) + transient_count = (report.discovery_stats or {}).get( + "transient_execution_ids", 0 + ) + 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/tests/features/ingestion/test_lcrc_v3_archive_ingestor.py b/backend/tests/features/ingestion/test_lcrc_v3_archive_ingestor.py new file mode 100644 index 00000000..9c3203f4 --- /dev/null +++ b/backend/tests/features/ingestion/test_lcrc_v3_archive_ingestor.py @@ -0,0 +1,381 @@ +"""Tests for targeted Chrysalis E3SM v3 archive uploads.""" + +import json +import urllib.request +from functools import partial +from pathlib import Path +from typing import Any + +from app.scripts.ingestion import hpc_upload_archive_ingestor as upload_ingestor +from app.scripts.ingestion.archive_discovery import _new_discovery_stats +from app.scripts.ingestion.archive_ingestor_core import ( + CaseCollectionLogData, + IngestionRequestResponse, + IngestorConfig, + IngestorRunReport, + _fresh_state, +) +from app.scripts.ingestion.v3_data import lcrc_v3_archive_ingestor as v3_ingestor + + +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 = _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 _fail_checkpoint(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("targeted runner must not use archive checkpoints") + + +def _fail_write(*args: Any, **kwargs: Any) -> Any: + raise AssertionError("dry run must not write") + + +def _post_discovery(*args: Any, **kwargs: Any) -> IngestionRequestResponse: + return {"status_code": 201, "body": {}} + + +class _FakeUrlopen: + def __init__(self, captured_requests: list[urllib.request.Request]) -> None: + self.captured_requests = captured_requests + + def __call__( + self, request: urllib.request.Request, timeout: int + ) -> _FakeHttpResponse: + self.captured_requests.append(request) + assert timeout == 30 + return _FakeHttpResponse() + + +class _CompleteReportRunner: + def __init__( + self, captured_kwargs: dict[str, Any], *, remove_one_case: bool = False + ) -> None: + self.captured_kwargs = captured_kwargs + self.remove_one_case = remove_one_case + + def __call__(self, config: IngestorConfig, **kwargs: Any) -> int: + self.captured_kwargs.update(kwargs) + report = kwargs["run_report"] + _populate_complete_report(report) + if self.remove_one_case: + report.case_collection_data.pop(next(iter(report.case_collection_data))) + return 0 + + +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_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_case_directory_pruner_only_prunes_proven_user_directory( + tmp_path: Path, +) -> None: + archive_root = tmp_path / "OLD_PERF" + user_dir = ( + archive_root + / "2024-01" + / "performance_archive_2024_01_01_00_00_00" + / "COMPLETED" + / "user" + ) + unchanged_paths = ( + archive_root, + archive_root / "2024-01", + archive_root / "2024-01" / "performance_archive_2024_01_01_00_00_00", + user_dir.parent, + archive_root + / "2024-13" + / "performance_archive_2024_01_01_00_00_00" + / "COMPLETED" + / "user", + archive_root + / "2024-01" + / "not_performance_archive_2024_01_01_00_00_00" + / "COMPLETED" + / "user", + user_dir / "v3.LR.piControl", + user_dir / "v3.LR.piControl" / "100.1-1", + archive_root / "2024-01" / "performance_archive_2024_01_01_00_00_00" / "user", + archive_root + / "2024-01" + / "performance_archive_2024_01_01_00_00_00" + / "FAILED" + / "user", + ) + child_names = ["v3.LR.piControl", "unrelated-case"] + + v3_ingestor._prune_v3_case_directories( + str(user_dir), child_names, archive_root=archive_root + ) + + assert child_names == ["v3.LR.piControl"] + for path in unchanged_paths: + child_names = ["v3.LR.piControl", "unrelated-case"] + v3_ingestor._prune_v3_case_directories( + str(path), child_names, archive_root=archive_root + ) + assert child_names == ["v3.LR.piControl", "unrelated-case"] + + +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 = _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(), + ) + + monkeypatch.setattr(upload_ingestor, "_fetch_archive_checkpoints", _fail_checkpoint) + monkeypatch.setattr( + upload_ingestor.urllib.request, "urlopen", _FakeUrlopen(captured_requests) + ) + + 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(), + ) + + 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 {})), + ) + + monkeypatch.setattr( + v3_ingestor, "_run_upload_ingestor", _CompleteReportRunner(captured_kwargs) + ) + + 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 + additional_dir_pruner = captured_kwargs["additional_dir_pruner"] + assert isinstance(additional_dir_pruner, partial) + assert additional_dir_pruner.func is v3_ingestor._prune_v3_case_directories + assert additional_dir_pruner.keywords == { + "archive_root": config.archive_root.resolve() + } + 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) + + monkeypatch.setattr( + v3_ingestor, + "_run_upload_ingestor", + _CompleteReportRunner({}, remove_one_case=True), + ) + + assert v3_ingestor.main() == 1 diff --git a/docs/architecture/metadata-ingestion.md b/docs/architecture/metadata-ingestion.md index f53fea52..b1ea8d5c 100644 --- a/docs/architecture/metadata-ingestion.md +++ b/docs/architecture/metadata-ingestion.md @@ -192,6 +192,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 +- `v3_data/lcrc_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 | @@ -200,6 +201,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 @@ -232,6 +248,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 2bccc42c..850fb049 100644 --- a/docs/deploy/hpc-api-token-authentication.md +++ b/docs/deploy/hpc-api-token-authentication.md @@ -80,13 +80,56 @@ 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" ``` +#### One-time 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, copy the committed template outside the repository, +secure it, replace its placeholders, then start this one-time backfill with a +dry run: + +```bash +mkdir -p ~/.config/simboard +cp app/scripts/ingestion/v3_data/lcrc-v3.env.example ~/.config/simboard/lcrc-v3.env +chmod 600 ~/.config/simboard/lcrc-v3.env +# Edit ~/.config/simboard/lcrc-v3.env to replace placeholders. +LCRC_V3_ENV_FILE=~/.config/simboard/lcrc-v3.env \ + ./app/scripts/ingestion/v3_data/lcrc_v3.sh +``` + +Use `backend/app/scripts/ingestion/v3_data/lcrc_v3.sh` from `backend/`; it +requires `LCRC_V3_ENV_FILE` and validates both API variables after sourcing it. +The template defaults to dry-run mode and the LCRC archive root. + +`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, then set `DRY_RUN=false` in the +external environment file before enabling uploads: + +```bash +LCRC_V3_ENV_FILE=~/.config/simboard/lcrc-v3.env \ + ./app/scripts/ingestion/v3_data/lcrc_v3.sh +``` + +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 diff --git a/docs/deploy/nersc-spin-runbook.md b/docs/deploy/nersc-spin-runbook.md index 59f4a6c5..8bd16da3 100644 --- a/docs/deploy/nersc-spin-runbook.md +++ b/docs/deploy/nersc-spin-runbook.md @@ -637,6 +637,20 @@ Service Discovery -> Ingresses -> Create | Name | `lb` | | Ingress class | `nginx` | +#### Labels & Annotations + +Add the following annotation to allow remote archive-ingestion uploads to reach +the backend. Without it, the NGINX Ingress default request-body limit can reject +uploads before the API processes them. + +| Annotation key | Value | +| -------------- | ----- | +| `nginx.ingress.kubernetes.io/proxy-body-size` | `52m` | + +Keep this value slightly above the backend upload limit in +`backend/app/features/ingestion/api.py` to allow multipart overhead. Update the +Ingress annotation if that backend limit changes. + #### TLS tab | Rancher field | Value |