Add Chrysalis E3SM v3 archive ingestion

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

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):

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,
     )

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
 
 

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,

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-host>
+SIMBOARD_API_TOKEN=<service-account-token>
+DRY_RUN=true
+
+# Optional: override the default /lcrc/group/e3sm/PERF_Chrysalis/OLD_PERF root.
+# OLD_PERF_ARCHIVE_ROOT=/path/to/OLD_PERF

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 "$@"