diff --git a/scripts/migrate/README.md b/scripts/migrate/README.md index 5ff1a5f26..f0d09b9f0 100644 --- a/scripts/migrate/README.md +++ b/scripts/migrate/README.md @@ -1,26 +1,25 @@ -# Science-file S3/DB migration +# Science-file S3 migration One-off tooling to re-version IMAP science files: it rewrites each file's -name (and the embedded CDF metadata) to the new versioning scheme and updates -the matching `science_files` rows in the RDS database. +name (and the embedded CDF metadata) to the new versioning scheme. The heavy lifting runs on a short-lived EC2 instance so it has in-region, egress-free -access to the S3 bucket and the RDS instance. The `run` script provisions that -instance, ships the migration code to it, executes it, and tears everything down. +access to the S3 buckets. The `run` script provisions that instance, ships the +migration code to it, executes it, and tears everything down. ``` run (your laptop) ├─ ensures IAM role + instance profile + key pair exist - ├─ launches (or reuses) a t3.large EC2 instance - ├─ opens the RDS security group to that instance's IP - ├─ scp run_remote.sh + migrate.py to the instance - └─ ssh: run_remote.sh - ├─ installs uv + Python deps - └─ python migrate.py <- the actual S3 / DB migration + ├─ ensures an SSH security group (port 22) exists + ├─ launches (or reuses) an m9g.48xlarge EC2 instance + ├─ scp run_remote_rename.sh + rename.py to the instance + └─ ssh: run_remote_rename.sh + ├─ builds the NASA CDF C library + installs uv + Python deps + └─ python rename.py <- the actual S3 migration ``` -On exit (success, failure, or Ctrl-C) the instance is terminated and the -security-group rule is revoked automatically. +On exit (success, failure, or Ctrl-C) the instance is terminated automatically. +The SSH security group is created once and left in place between runs. --- @@ -31,8 +30,8 @@ On the **local** machine that runs `run`: - AWS CLI v2, authenticated for the target account. `run` uses `AWS_PROFILE=imap-dev` — set up that profile (`aws configure --profile imap-dev`) or edit the variable at the top of `run`. -- Permission to manage IAM roles/instance profiles, EC2 key pairs and - instances, and to modify the RDS security group. +- Permission to manage IAM roles/instance profiles, EC2 key pairs, + security groups, and instances. - `ssh`, `scp`, `openssl`, `envsubst` on `PATH`. Run `bash run` to test out the setup. This won't actually run the migration, simply @@ -46,40 +45,35 @@ Configure the run by editing the variables near the top of `run`, then execute i | Variable | Default | Meaning | |----------|---------|------------------------------------------------------------| -| `COPY_FILES` | `0` | `1` = copy/rewrite files in S3 (stage 1). | -| `MODIFY_ROWS` | `0` | `1` = update `file_path` in the database (stage 2). | +| `DRY_RUN` | `1` | `1` = only print the old -> new name mapping, write nothing. Set to `0` to actually migrate. | | `OVERWRITE` | `0` | `1` = re-copy destination files even if they already exist. | | `MAX_FILES` | `1000` | Max CDF/PKTS files to process this run (`0` = all). | -> **`COPY_FILES` and `MODIFY_ROWS` are mutually exclusive.** `migrate.py` -> asserts you never enable both in the same run — do it in stages (below). +`SRC_PREFIX`, `DRY_RUN`, and `MAX_FILES` can also be overridden from the +environment without editing `run`, e.g. `DRY_RUN=0 MAX_FILES=500 bash run`. +This is handy for running several prefixes in parallel (see below). --- ## Workflow -### Incremental file copying (stage 1) +### Incremental file copying -```bash -COPY_FILES=1 -MODIFY_ROWS=0 -``` - -Reads each source object under `imap/`, rewrites the CDF metadata +Reads each source object under `$SRC_PREFIX` in `$SRC_BUCKET`, rewrites the CDF metadata (`Data_version`, `Logical_file_id`, `Parents`) and writes it to the new name -under the `renamed/` prefix. PKTS files are copied as-is to the new name. -Existing objects under `renamed/` are skipped, so this stage is resumable and -can be run in batches (see below). +under the same `$SRC_PREFIX`, but now in `$DST_BUCKET`. PKTS files are copied as-is to +the new name. Existing objects at destination are skipped, so this stage is resumable +and can be run in batches (see below). `MAX_FILES` limits how many CDF/PKTS files a single run processes. Because -stage 1 **skips destinations that already exist** under `renamed/`, repeated +it **skips destinations that already exist**, repeated runs with the same `MAX_FILES` walk through the whole set a batch at a time: ```bash -# in run: COPY_FILES=1, MODIFY_ROWS=0, OVERWRITE=0, MAX_FILES=1000 +# in run: DRY_RUN=0, OVERWRITE=0, MAX_FILES=1000 bash run # copies the first 1000 not-yet-copied files bash run # copies the next 1000 -bash run # ... repeat until everything is under renamed/ +bash run # ... repeat until everything is copied to DST_BUCKET ``` Set `MAX_FILES=0` to process everything in one run. *Not recommended.* except for @@ -89,45 +83,36 @@ To **regenerate** files you have already copied (e.g. after fixing the metadata logic), set `OVERWRITE=1`. This disables the skip-if-exists behavior and re-copies the selected files in place. -Copying runs across multiple CPU cores automatically (one worker per core). -`run` uses `t3.large` (2 vCPUs), so two files are rewritten in parallel; -use a larger instance type if you want more throughput. `t3.xlarge` has 4 vCPUs. -`t3.2xlarge` has 8 vCPUs. - - -### Manual step - promote the renamed files - -After verifying `renamed/`, back up the existing `imap/` tree and then bulk-move -objects from `renamed/..` to their final `imap/..` paths. `run` does not do this move -for you. - -You will want to keep the `ancillary/`, `dependency/` and `spice/` trees in `imap/` -intact, since these do not have rows in the science_files table and are not affected by -the renaming. - -### Update the database (stage 2) - -```bash -COPY_FILES=0 -MODIFY_ROWS=1 -``` +This is single threaded since I'm having trouble getting `spacepy` to behave in a +multi-processing environment. However, a bigger instance will still make it go faster. -Updates each `science_files.file_path` from the old name to the new one. +### Running several prefixes in parallel -Then run it: +To speed things up you can shard the work by `SRC_PREFIX` and run several copies +of `run` at once, one per prefix. `SRC_PREFIX`, `DRY_RUN`, and `MAX_FILES` can +be overridden from the environment, so **do not edit `run` in place** for this — +pass them on the command line instead: ```bash -bash run # or ./run +DRY_RUN=0 SRC_PREFIX=imap/lo/ bash run # terminal 1 +DRY_RUN=0 SRC_PREFIX=imap/mag/ bash run # terminal 2 +DRY_RUN=0 SRC_PREFIX=imap/swe/ bash run # terminal 3 ``` -#### Tips +Note the default is `DRY_RUN=1`, which only prints the `old -> new` mapping and +writes **nothing** — pass `DRY_RUN=0` to actually migrate. -- Inspect s3 bucket in a separate terminal to monitor file copy progress. +Each copy derives its own EC2 instance name from the prefix +(`INSTANCE_TAG=s3-transition-`), so every terminal gets its **own** +instance, its own `$HOME` on that instance, and a teardown that only kills its +own box — the runs don't interfere. The shared IAM role, key pair, and security +group are created-if-missing, so a quick warm-up `bash run` before fanning out +avoids a first-run creation race on those. - ```bash - aws s3 ls s3://sds-data-593025701104/renamed/ --recursive --summarize --human-readable - ``` +Caveats: -- Set `INTERACTIVE=1` to provision the instance and drop into an SSH shell -instead of running the migration. The instance is still torn down when you exit -the shell. \ No newline at end of file +- `SRC_PREFIX` **must end with `/`. +- Give every terminal a **different** `SRC_PREFIX`. Two runs with the same + prefix resolve to the same instance name and will collide. +- Each shard launches its own `m9g.48xlarge`, so N terminals = N instances + running at once — watch the cost and any vCPU quota. diff --git a/scripts/migrate/ec2-perms.json b/scripts/migrate/ec2-perms.json index cab9d83a3..d40c671d8 100644 --- a/scripts/migrate/ec2-perms.json +++ b/scripts/migrate/ec2-perms.json @@ -6,22 +6,24 @@ "Action": [ "s3:ListBucket" ], - "Resource": "arn:aws:s3:::${AWS_BUCKET}" + "Resource": [ + "arn:aws:s3:::${SRC_BUCKET}", + "arn:aws:s3:::${DST_BUCKET}" + ] }, { "Effect": "Allow", "Action": [ - "s3:GetObject", - "s3:PutObject" + "s3:GetObject" ], - "Resource": "arn:aws:s3:::${AWS_BUCKET}/*" + "Resource": "arn:aws:s3:::${SRC_BUCKET}/*" }, { "Effect": "Allow", "Action": [ - "secretsmanager:GetSecretValue" + "s3:PutObject" ], - "Resource": "arn:aws:secretsmanager:${AWS_DEFAULT_REGION}:${AWS_ACCOUNT}:secret:sdp-database-cred*" + "Resource": "arn:aws:s3:::${DST_BUCKET}/*" } ] } diff --git a/scripts/migrate/migrate.py b/scripts/migrate/migrate.py deleted file mode 100644 index ddad2d1d9..000000000 --- a/scripts/migrate/migrate.py +++ /dev/null @@ -1,351 +0,0 @@ -"""Migration script for renaming science files in S3/DB.""" - -import logging -import multiprocessing as mp -import os -import tempfile -from concurrent.futures import ProcessPoolExecutor -from pathlib import Path - -import boto3 -import imap_data_access -from imap_data_access.file_validation import ScienceFilePath, Version -from imap_processing.cdf.utils import load_cdf -from imap_processing.cdf.utils import write_cdf as _write_cdf - -from sds_data_manager.lambda_code.SDSCode.database import database as db -from sds_data_manager.lambda_code.SDSCode.database import models - -# Destination prefix for copied files (e.g. "renamed/") -DEST_PREFIX: str = os.getenv("DEST_PREFIX", "renamed/") -# Reverse the sense of `old` vs `new` paths? (for testing on dev) -REVERSE: bool = False -# Write a dummy CDF instead of a real one to make the script go fast (for testing) -DUMMY_CDF: bool = False - - -def write_cdf(dataset, **kwargs): - """Write a CDF, or a dummy placeholder file if ``DUMMY_CDF`` is set.""" - if DUMMY_CDF: - with tempfile.NamedTemporaryFile(suffix=".cdf", delete=False) as tmp: - tmp.write(b"dummy cdf") - return tmp.name - return _write_cdf(dataset, **kwargs) - - -logger = logging.getLogger(__name__) -logger.setLevel(logging.DEBUG) -logging.basicConfig(level=logging.INFO) - - -def remap_parents(dataset, basename_map: dict[str, str]): - """Update the ``Parents`` attribute to reflect the CDF renaming. - - ``Parents`` is a list of dependency file *basenames* (see imap_processing - ``cli.py``: ``[p.name for p in dependencies.get_file_paths()]``). Many of - those parents are themselves science files being renamed by this migration, - so each basename is remapped via ``basename_map``. Parents not in the map - (e.g. SPICE/ancillary files) are left unchanged. ``load_cdf`` collapses a - single-element ``Parents`` to a scalar string. - """ - parents = dataset.attrs.get("Parents") - logger.info(f"Parents: {parents}") - if parents is None: - return - if isinstance(parents, str): - parents = [parents] - - # verbose, but we need the logging - new_parents = [] - for p in parents: - if p in basename_map: - logger.info(f"Remapping parent {p} to {basename_map[p]}") - new_parents.append(basename_map[p]) - else: - new_parents.append(p) - dataset.attrs["Parents"] = new_parents - - -def upload_cdf( - client, - bucket: str, - src_key: str, - dst_key: str, - dst_version: str, - basename_map: dict[str, str], - overwrite: bool = False, -): - """Download/modify/upload a pkts/cdf file on S3.""" - if not overwrite: - try: - client.head_object(Bucket=bucket, Key=dst_key) - except client.exceptions.ClientError as e: - code = e.response.get("Error", {}).get("Code") - if code not in ("404", "NoSuchKey", "NotFound"): - raise - else: - logger.info(f"Target exists, leaving untouched: s3://{bucket}/{dst_key}") - return - - if src_key.endswith("pkts"): - client.copy_object( - Bucket=bucket, - CopySource={"Bucket": bucket, "Key": src_key}, - Key=dst_key, - ) - logger.info(f"Copied PKTS {src_key} -> s3://{bucket}/{dst_key}") - return - - with tempfile.NamedTemporaryFile(suffix=".cdf") as tmp: - client.download_fileobj(bucket, src_key, tmp) - tmp.flush() - dataset = load_cdf(tmp.name) - - # From @tech3371 - `Data_version` is sans `v` - dataset.attrs["Data_version"] = dst_version.lstrip("v") - - # `Logical_file_id` must match the renamed filename, sans extension. - dataset.attrs["Logical_file_id"] = Path(dst_key).stem - logger.info(f"Logical_file_id = {dataset.attrs['Logical_file_id']}") - - # Parent filenames embed the old version format and may themselves be - # renamed science files, so remap them to match the new CDF names. - remap_parents(dataset, basename_map) - - # Making guarantees about spdf conformance on existing files is out of scope - written = Path(write_cdf(dataset, istp=True, terminate_on_warning=False)) - try: - client.upload_file(str(written), bucket, dst_key) - finally: - written.unlink(missing_ok=True) - logger.info(f"Copied CDF {src_key} -> s3://{bucket}/{dst_key}") - - -# --- Parallel copy workers ------------------------------------------------- -# Each pool worker holds its own boto3 client (clients must not be shared -# across processes) plus the read-only state every copy needs. Populated once -# per process by the pool initializer, then reused for every task. -_WORKER: dict = {} - - -def _init_worker(bucket: str, basename_map: dict[str, str], overwrite: bool): - """Set up per-process state for the copy pool.""" - _WORKER["client"] = boto3.client("s3") - _WORKER["bucket"] = bucket - _WORKER["basename_map"] = basename_map - _WORKER["overwrite"] = overwrite - - -def _copy_one(task: tuple[str, str, str]) -> tuple[str, str, str | None]: - """Copy one file in a worker; return ``(src, dst, error_or_None)``.""" - src_path, dst_key, dst_version = task - try: - upload_cdf( - _WORKER["client"], - _WORKER["bucket"], - src_path, - dst_key, - dst_version, - _WORKER["basename_map"], - overwrite=_WORKER["overwrite"], - ) - return src_path, dst_key, None - except Exception as e: - return src_path, dst_key, str(e) - - -def get_s3_keys(bucket, prefix="imap/"): - """Return the set of all object keys in ``bucket`` under ``prefix``.""" - client = boto3.client("s3") - paginator = client.get_paginator("list_objects_v2") - keys = set() - for page in paginator.paginate(Bucket=bucket, Prefix=prefix): - keys.update(obj["Key"] for obj in page.get("Contents", [])) - return keys - - -def get_existing_new_files(bucket, prefix="imap/"): - """Return the set of all object keys in ``bucket`` under ``prefix``.""" - client = boto3.client("s3") - paginator = client.get_paginator("list_objects_v2") - keys = set() - for page in paginator.paginate(Bucket=bucket, Prefix=prefix): - keys.update( - obj["Key"] for obj in page.get("Contents", []) if "v001.0" in obj["Key"] - ) - return keys - - -def compute_paths(row, data_dir): - """Return ``(old_path, old_version, new_path, new_version)`` for a DB row. - - Paths are relative to ``data_dir``. The new filename is constructed from - the table columns; the old path differs only in the version suffix (the - old names lacked the major version). - """ - old_version = str(Version(None, row.minor_version)) - new_version = str(Version(row.major_version, row.minor_version)) - - old_suffix = f"_{old_version}.{row.extension}" - new_suffix = f"_{new_version}.{row.extension}" - - # Construct the new filename from scratch using the table columns. - new_file = ScienceFilePath.generate_from_inputs( - instrument=row.instrument, - data_level=row.data_level, - descriptor=row.descriptor, - start_time=row.start_date.strftime("%Y%m%d"), - major_version=row.major_version, - minor_version=row.minor_version, - extension=row.extension, - repointing=row.repointing, - cr=row.cr, - ) - - # construct_path() prepends DATA_DIR; strip it - new_file_path = str(new_file.construct_path().relative_to(data_dir)) - old_file_path = new_file_path[: -len(new_suffix)] + old_suffix - return old_file_path, old_version, new_file_path, new_version - - -def migrate( # noqa: PLR0912, PLR0915 - copy_files: bool = False, - modify_rows: bool = False, - overwrite: bool = False, - max_files: int = 0, - max_workers: int = 0, -): - """Migrate science files in S3 or update the database.""" - assert not all([copy_files, modify_rows]), "Please do this in stages!" - - data_dir = imap_data_access.config["DATA_DIR"] - - bucket = os.getenv("S3_BUCKET") - if not bucket: - raise ValueError("S3_BUCKET environment variable is not set") - s3_keys: set[str] = set() - if copy_files: - logger.info(f"Listing objects in s3://{bucket}/imap/ ...") - s3_keys = get_s3_keys(bucket) - logger.info(f"Found {len(s3_keys)} objects in the bucket") - - with db.Session() as session: - count = session.query(models.ScienceFiles).count() - logger.info(f"Verifying file_path mapping for {count} records") - - # old_basename => new_basename, covering ALL rows (not just the ones - # copied this run) so the `Parents` attribute can be fully remapped. - basename_map: dict[str, str] = {} - for row in session.query(models.ScienceFiles): - old_file_path, _, new_file_path, _ = compute_paths(row, data_dir) - basename_map[os.path.basename(old_file_path)] = os.path.basename( - new_file_path - ) - - # Destinations already copied under DEST_PREFIX. Skipping these lets - # repeated copy runs advance through the full set, max_files at a time, - # instead of re-copying the same first N. Only meaningful when copying - # and not force-overwriting (overwrite deliberately re-copies existing). - existing_dsts: set[str] = set() - if copy_files and not overwrite: - existing_dsts = get_existing_new_files(bucket, prefix=DEST_PREFIX) - logger.info(f"Found {len(existing_dsts)} objects under {DEST_PREFIX}") - - # Candidate CDF/PKTS rows, deterministically ordered so the "next N not - # yet copied" is well-defined and reproducible across runs. - candidates = ( - session.query(models.ScienceFiles) - .filter(models.ScienceFiles.extension.in_(["cdf", "pkts"])) - .order_by(models.ScienceFiles.file_path) - ) - - # (current_path, current_version_str) => (new_path, new_version_str), - # skipping files already present under DEST_PREFIX and capping the - # selection at max_files rows (0 => no cap). - path_mapping: dict[tuple[str, str], tuple[str, str]] = {} - for row in candidates: - old_file_path, old_version, new_file_path, new_version = compute_paths( - row, data_dir - ) - # Path actually written under DEST_PREFIX (old vs new per REVERSE). - dst_path = old_file_path if REVERSE else new_file_path - if f"{DEST_PREFIX}{dst_path}" in existing_dsts: - continue - path_mapping[(old_file_path, old_version)] = (new_file_path, new_version) - if not modify_rows and max_files != 0 and len(path_mapping) >= max_files: - break - - if REVERSE: - rename_map = {v: k for k, v in path_mapping.items()} - basename_map = {v: k for k, v in basename_map.items()} - else: - rename_map = dict(path_mapping) - - # for (src_path, _), (dst_path, _) in rename_map.items(): - # logger.info(f"Mapping {src_path} -> {dst_path}") - - dst_paths = list(rename_map.values()) - assert len(set(dst_paths)) == len(dst_paths), "Duplicates in dst_paths!" - - if copy_files: - # Build the task list first (cheap, serial), skipping no-ops so the - # workers only ever do real copies. - tasks: list[tuple[str, str, str]] = [] - for (src_path, _), (dst_path, dst_version) in rename_map.items(): - if src_path == dst_path: - logger.info(f"Identical src/dst: {src_path}") - continue - if src_path not in s3_keys: - logger.info(f"Cannot read missing object: {src_path}") - continue - tasks.append((src_path, f"{DEST_PREFIX}{dst_path}", dst_version)) - - # Each copy downloads, rewrites and re-uploads a CDF (CPU + I/O - # heavy), so fan the work across processes to use all cores. A - # `spawn` context gives each worker a clean interpreter with no - # inherited boto3/DB sockets; workers never touch the DB. - if tasks: - workers = max_workers if max_workers > 0 else (os.cpu_count() or 1) - workers = max(1, min(workers, len(tasks))) - logger.info(f"Copying {len(tasks)} files across {workers} workers") - with ProcessPoolExecutor( - max_workers=workers, - mp_context=mp.get_context("spawn"), - initializer=_init_worker, - initargs=(bucket, basename_map, overwrite), - ) as executor: - for src_path, dst_key, err in executor.map(_copy_one, tasks): - if err: - logger.info( - f"Failed to copy {src_path} -> {dst_key} - {err}" - ) - logger.info("All destination files written") - - # Updating rows does not use `dst_key` at all. It is assumed that after making - # a backup of the `imap/` path in the S3 bucket, files will be moved from - # DEST_PREFIX to the original path in bulk, and then this block will be run. - if modify_rows: - for (src_path, _), (dst_path, _) in rename_map.items(): - session.query(models.ScienceFiles).filter( - models.ScienceFiles.file_path == src_path - ).update( - {models.ScienceFiles.file_path: dst_path}, - synchronize_session=False, - ) - session.commit() - logger.info(f"Updated file_path for {len(rename_map)} records") - - -if __name__ == "__main__": - copy_files = os.getenv("COPY_FILES", "0") == "1" - modify_rows = os.getenv("MODIFY_ROWS", "0") == "1" - overwrite = os.getenv("OVERWRITE", "0") == "1" - max_files = int(os.getenv("MAX_FILES", "0")) - max_workers = int(os.getenv("MAX_WORKERS", "0")) - migrate( - copy_files=copy_files, - modify_rows=modify_rows, - overwrite=overwrite, - max_files=max_files, - max_workers=max_workers, - ) diff --git a/scripts/migrate/rename.py b/scripts/migrate/rename.py new file mode 100644 index 000000000..689e963e3 --- /dev/null +++ b/scripts/migrate/rename.py @@ -0,0 +1,281 @@ +# ruff: noqa + +import logging +import multiprocessing as mp +import os +import re +import tempfile +from concurrent.futures import ProcessPoolExecutor +from pathlib import Path + +import boto3 +from botocore.exceptions import ClientError + +# NB: spacepy is imported lazily (inside `upgrade_file`), never at module top. +# `import spacepy` bootstraps a per-user config file ($SPACEPY/.spacepy/spacepy.rc) +# via a non-atomic read-modify-write. Under the `spawn` pool every worker imports +# the module fresh, and concurrent bootstraps race on that shared file -- one +# worker's partial write is another's corrupt read, crashing with +# `UnboundLocalError: nextsec` -> BrokenProcessPool. Deferring the import lets +# `_init_worker` point each worker at its own $SPACEPY dir first, so there is no +# shared config file to race on. + +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.INFO, format="%(message)s") + +MAJOR_VERSION = 1 + +_OLD_SUFFIX_RE = re.compile(r"_v(\d{3})\.(cdf|pkts)$") + +# Ancillary parents (e.g. imap_mag_l1b-calibration_20240229_v001.cdf) keep +# vNNN naming and must pass through unchanged (see #1538). +_OLD_SCIENCE_RE = re.compile( + r"^imap_[a-z0-9]+_l\d[a-z]?_[a-zA-Z0-9-]+_\d{8}(?:-repoint\d+)?_v\d{3}\.(cdf|pkts)$" +) + + +def make_new_file_name(name: str, major: int = MAJOR_VERSION) -> str: + replacement = r"_v%03d.0\1.\2" % major + return _OLD_SUFFIX_RE.sub(replacement, name) + + +def upgrade_file( + original_file: Path, working_dir: Path, major: int = MAJOR_VERSION +) -> Path: + original_file = Path(original_file) + match = _OLD_SUFFIX_RE.search(original_file.name) + if match is None: + raise ValueError(f"Not a legacy-versioned science file: {original_file.name}") + old_version = match[1] # e.g. "005" + new_version = f"{major:03d}.{int(old_version):04d}" # e.g. "001.0005" + + new_name = make_new_file_name(original_file.name, major) + out_path = Path(working_dir) / new_name + + # Lazy import: see the note at the top of this module. In the pool this runs + # after `_init_worker` has isolated $SPACEPY; standalone/sequential callers + # just import it here in-process (no concurrency, no race). + from spacepy.pycdf import CDF + + with CDF(str(out_path), masterpath=str(original_file)) as cdf: + # `Data_version` is stored without the leading `v`. + cdf.attrs["Data_version"] = new_version + + # `Logical_file_id` must match the renamed filename, sans extension. + if "Logical_file_id" in cdf.attrs and len(cdf.attrs["Logical_file_id"]): + logical_file_id = str(cdf.attrs["Logical_file_id"][0]) + cdf.attrs["Logical_file_id"] = logical_file_id.replace( + f"_v{old_version}", f"_v{new_version}" + ) + + if "File_naming_convention" in cdf.attrs: + cdf.attrs["File_naming_convention"] = ( + "source_descriptor_datatype_yyyyMMdd_vMMM.mmmm" + ) + + if "Parents" in cdf.attrs: + parents = [str(p) for p in cdf.attrs["Parents"]] + cdf.attrs["Parents"] = [ + make_new_file_name(p, major) if _OLD_SCIENCE_RE.match(p) else p + for p in parents + ] + + return out_path + + +# --- S3 listing ------------------------------------------------------------ +def list_keys(bucket: str, prefix: str) -> list[str]: + """Return all object keys in ``bucket`` under ``prefix``.""" + client = boto3.client("s3") + paginator = client.get_paginator("list_objects_v2") + keys: list[str] = [] + for page in paginator.paginate(Bucket=bucket, Prefix=prefix): + keys.extend(obj["Key"] for obj in page.get("Contents", [])) + return keys + + +# --- worker pool ----------------------------------------------------------- +# Per-process state. We deliberately do NOT pickle a boto3 client (or anything +# holding native/OpenSSL handles) across the process boundary: each worker +# builds its own in `_init_worker`. Combined with the `spawn` start method (see +# `rename`), this guarantees no half-initialized C-library state -- libcdf or +# boto3's SSL sockets -- is inherited across a fork, which is the actual cause +# of the intermittent "spacepy crashing under multiprocessing" segfaults. +_WORKER: dict = {} + + +def _init_worker(src_bucket: str, dst_bucket: str, major: int) -> None: + """Set up per-process state for the rename pool.""" + # Give this worker its own spacepy config dir *before* spacepy is ever + # imported, so the workers never share (and race on) a single + # ~/.spacepy/spacepy.rc. spacepy reads $SPACEPY/.spacepy/spacepy.rc; a unique + # per-process dir makes the config bootstrap contention-free. Importing + # spacepy here (single-threaded within this worker) bootstraps it once. + spacepy_home = tempfile.mkdtemp(prefix=f"spacepy-{os.getpid()}-") + os.environ["SPACEPY"] = spacepy_home + import spacepy.pycdf # noqa: F401 (force the isolated config bootstrap now) + + _WORKER["client"] = boto3.client("s3") + _WORKER["src_bucket"] = src_bucket + _WORKER["dst_bucket"] = dst_bucket + _WORKER["major"] = major + + +def _process_one_worker(task: tuple[str, str]) -> tuple[str, str, str | None]: + """Rename one object using this worker's own client (pool entry point).""" + return _process_one( + _WORKER["client"], + _WORKER["src_bucket"], + _WORKER["dst_bucket"], + task, + _WORKER["major"], + ) + + +def _process_one( + client, src_bucket: str, dst_bucket: str, task: tuple[str, str], major: int +) -> tuple[str, str, str | None]: + """Rename one object; return ``(src, dst, error_or_None)``.""" + src_key, dst_key = task + try: + if src_key.endswith(".pkts"): + # No metadata to rewrite - a server-side cross-bucket copy suffices. + client.copy_object( + Bucket=dst_bucket, + CopySource={"Bucket": src_bucket, "Key": src_key}, + Key=dst_key, + ) + return src_key, dst_key, None + + # CDF: download, rewrite attributes in place, upload the renamed copy. + with tempfile.TemporaryDirectory() as workdir: + local_src = Path(workdir) / Path(src_key).name + client.download_file(src_bucket, src_key, str(local_src)) + out_path = upgrade_file(local_src, workdir, major) + client.upload_file(str(out_path), dst_bucket, dst_key) + return src_key, dst_key, None + except Exception as e: + return src_key, dst_key, f"ERROR: {e}" + + +def rename( + src_bucket: str, + dst_bucket: str, + prefix: str = "imap/", + overwrite: bool = False, + max_files: int = 0, + max_workers: int = 0, + dry_run: bool = False, + major: int = MAJOR_VERSION, +): + """Rename science files from ``src_bucket`` to ``dst_bucket``.""" + if dst_bucket == src_bucket: + raise ValueError("dst_bucket must differ from src_bucket") + + # S3 prefixes match by raw string + # Let the caller explicitly choose "imap/hi/" vs "imap/hit/" to avoid confusion + if not prefix.endswith("/"): + raise ValueError( + f"prefix must end with '/' to keep shards disjoint (got {prefix!r}); " + f"did you mean {prefix + '/'!r}?" + ) + + logger.info(f"Listing s3://{src_bucket}/{prefix} ...") + src_keys = sorted(list_keys(src_bucket, prefix)) + logger.info(f"Found {len(src_keys)} source objects") + + # Destinations already written. Skipping these (unless overwriting) lets + # repeated runs advance through the whole set, max_files at a time, instead + # of re-processing the same first N. + existing_dst: set[str] = set() + if not overwrite: + try: + existing_dst = set(list_keys(dst_bucket, prefix)) + logger.info(f"Found {len(existing_dst)} objects already in destination") + except ClientError as e: + if not dry_run: + raise + code = e.response.get("Error", {}).get("Code") + logger.info( + f"Destination s3://{dst_bucket}/{prefix} not listable ({code}); " + f"assuming empty (dry run)" + ) + + tasks: list[tuple[str, str]] = [] + for key in src_keys: + base = os.path.basename(key) + new_base = make_new_file_name(base, major) + if new_base == base: + # Not a legacy-versioned science file (already-new, ancillary, + # spice, dependency, ...): leave it alone. + continue + dst_key = key[: len(key) - len(base)] + new_base + if not overwrite and dst_key in existing_dst: + continue + tasks.append((key, dst_key)) + if max_files and len(tasks) >= max_files: + break + + logger.info(f"{len(tasks)} files to process") + if not tasks: + return + + if dry_run: + for src_key, dst_key in tasks: + logger.info(f"{src_key} -> {dst_key}") + logger.info(f"Dry run: {len(tasks)} files would be renamed (nothing written)") + return + + # Each file downloads, rewrites and re-uploads a CDF (I/O heavy), so fan the + # work across processes. A `spawn` context gives every worker a clean + # interpreter that imports spacepy/boto3 fresh -- nothing native is inherited + # across a fork, which is what made spacepy segfault under multiprocessing. + workers = max_workers if max_workers > 0 else (os.cpu_count() or 1) + workers = max(1, min(workers, len(tasks))) + ok = fail = 0 + + if workers == 1: + # Single worker: stay in-process, no pool overhead. + logger.info(f"Processing {len(tasks)} files sequentially") + client = boto3.client("s3") + results = ( + _process_one(client, src_bucket, dst_bucket, task, major) for task in tasks + ) + else: + logger.info(f"Processing {len(tasks)} files across {workers} workers") + executor = ProcessPoolExecutor( + max_workers=workers, + mp_context=mp.get_context("spawn"), + initializer=_init_worker, + initargs=(src_bucket, dst_bucket, major), + ) + results = executor.map(_process_one_worker, tasks) + + try: + for src_key, dst_key, err in results: + if err: + fail += 1 + logger.info(f"FAIL {src_key} -> {dst_key}: {err}") + else: + ok += 1 + logger.info(f"OK {src_key} -> {dst_key}") + finally: + if workers != 1: + executor.shutdown() + logger.info(f"Done: {ok} succeeded, {fail} failed") + + +if __name__ == "__main__": + src_bucket = os.getenv("SRC_BUCKET") + dst_bucket = os.getenv("DST_BUCKET") + if not src_bucket or not dst_bucket: + raise SystemExit("SRC_BUCKET and DST_BUCKET must both be set") + rename( + src_bucket=src_bucket, + dst_bucket=dst_bucket, + prefix=os.getenv("SRC_PREFIX", "imap/"), + overwrite=os.getenv("OVERWRITE", "0") == "1", + max_files=int(os.getenv("MAX_FILES", "0")), + max_workers=int(os.getenv("MAX_WORKERS", "0")), + dry_run=os.getenv("DRY_RUN", "0") == "1", + ) diff --git a/scripts/migrate/run b/scripts/migrate/run index 71b1199a6..80887a13e 100755 --- a/scripts/migrate/run +++ b/scripts/migrate/run @@ -1,37 +1,49 @@ #!/bin/bash set -euo pipefail -export AWS_PROFILE=default +export AWS_PROFILE=imap-dev export AWS_DEFAULT_REGION=us-west-2 -export AWS_ACCOUNT=593025701104 -export AWS_BUCKET=sds-data-$AWS_ACCOUNT -export SECRET_NAME=sdp-database-cred + +# Source (read) and destination (write) S3 buckets for the rename. +export SRC_BUCKET=sds-data-449431850278 +export DST_BUCKET=deprecated-data-archive-dev +# Key prefix (relative to the bucket root) to read source files from. +# Override from the environment to run several copies concurrently, one per +# prefix, e.g. SRC_PREFIX=imap/lo/ bash run +# The prefix MUST end with "/" +export SRC_PREFIX="${SRC_PREFIX:-imap/mag/}" # Set to 1 to drop into an interactive SSH shell on the instance; set to 0 -# for fully-automated mode that runs the migration script remotely. +# for fully-automated mode that runs the rename script remotely. INTERACTIVE=0 -# Passed through to migrate.py. Set to 1 to copy files on S3, and/or to 1 to -# modify the DB rows. Run in stages - do not enable both at once. -COPY_FILES=1 -MODIFY_ROWS=0 -# Force overwriting of target cdf/pkts file(s) if they exist? +# Passed through to rename.py. MAX_FILES and DRY_RUN can be overridden from the +# environment (like SRC_PREFIX), e.g. DRY_RUN=0 MAX_FILES=500 bash run +# Force overwriting of target file(s) if they already exist in DST_BUCKET? OVERWRITE=0 -# Max cdf/pkts files to process in this batch (0 for all) -MAX_FILES=0 +# Max files to process in this batch (0 for all). +MAX_FILES="${MAX_FILES:-1000}" +# List the planned renames without writing anything? +DRY_RUN="${DRY_RUN:-1}" + +# Max workers, 0 for all available cores +MAX_WORKERS=0 ROLE_NAME=s3-transition-runner PROFILE_NAME=s3-transition-runner KEY_NAME=s3-transition-key KEY_FILE=s3-transition-key.pem -INSTANCE_TAG=s3-transition -# Function to tear down the instance (and the SG rule) on exit +# Derive a unique, tag-safe instance name from SRC_PREFIX so that multiple +# copies of this script (one per prefix) each get their own instance, their +# own $HOME on that instance, and a teardown that only kills their own box. +# The IAM role/profile, key pair, and security group above are intentionally +# NOT parametrized - they are shared and only ever created-if-missing. +PREFIX_SLUG=$(echo "$SRC_PREFIX" | tr -c 'A-Za-z0-9' '-' | sed 's/^-*//; s/-*$//') +INSTANCE_TAG=s3-transition-$PREFIX_SLUG + +# Function to tear down the instance on exit cleanup() { - if [ -n "${RDS_SG:-}" ] && [ -n "${EC2_IP:-}" ]; then - aws ec2 revoke-security-group-ingress --group-id "$RDS_SG" \ - --protocol tcp --port 5432 --cidr "$EC2_IP/32" 2>/dev/null || true - fi if [ -n "${IID:-}" ]; then aws ec2 terminate-instances --instance-ids "$IID" fi @@ -42,10 +54,10 @@ if ! aws iam get-role --role-name "$ROLE_NAME" >/dev/null 2>&1; then aws iam create-role --role-name "$ROLE_NAME" --assume-role-policy-document file://ec2-trust.json fi -# Render the permissions policy template, substituting the account/bucket/region -# from the variables above so they are not hardcoded in ec2-perms.json. +# Render the permissions policy template, substituting the source/destination +# buckets from the variables above so they are not hardcoded in ec2-perms.json. PERMS_FILE=$(mktemp) -envsubst '$AWS_BUCKET $AWS_ACCOUNT $AWS_DEFAULT_REGION' < ec2-perms.json > "$PERMS_FILE" +envsubst '$SRC_BUCKET $DST_BUCKET' < ec2-perms.json > "$PERMS_FILE" aws iam put-role-policy --role-name "$ROLE_NAME" --policy-name s3-transition-perms --policy-document "file://$PERMS_FILE" rm -f "$PERMS_FILE" @@ -154,15 +166,6 @@ EC2_IP=$(aws ec2 describe-instances --instance-ids "$IID" \ --query 'Reservations[0].Instances[0].PublicIpAddress' --output text) echo "$EC2_IP" -# Find the RDS security group id -RDS_SG=$(aws ec2 describe-security-groups \ - --filters Name=group-name,Values='SDCStack-RdsSecurityGroup*' \ - --query 'SecurityGroups[0].GroupId' --output text) - -# Authorize ingress (ignore error if the rule already exists). -aws ec2 authorize-security-group-ingress --group-id "$RDS_SG" \ - --protocol tcp --port 5432 --cidr "$EC2_IP/32" 2>/dev/null || true - # Ephemeral instances reuse public IPs, so skip host-key checking to avoid # "REMOTE HOST IDENTIFICATION HAS CHANGED" failures on re-runs. SSH_OPTS=(-i "$KEY_FILE" @@ -189,18 +192,18 @@ if [ "$ssh_ready" != 1 ]; then exit 1 fi -# Copy the migration scripts up to the instance. -scp "${SSH_OPTS[@]}" run_remote.sh migrate.py ec2-user@"$EC2_IP":~/ +# Copy the rename scripts up to the instance. +scp "${SSH_OPTS[@]}" run_remote_rename.sh rename.py ec2-user@"$EC2_IP":~/ if [ "${INTERACTIVE:-0}" = 1 ]; then echo "Interactive mode: opening a shell on $EC2_IP (instance is torn down on exit)." - echo " Run the migration manually with: bash ~/run_remote.sh" + echo " Run the rename manually with: bash ~/run_remote_rename.sh" ssh "${SSH_OPTS[@]}" ec2-user@"$EC2_IP" exit 0 fi -# Run the transition script on the instance. +# Run the rename script on the instance. ssh "${SSH_OPTS[@]}" ec2-user@"$EC2_IP" \ - "AWS_DEFAULT_REGION=$AWS_DEFAULT_REGION S3_BUCKET=$AWS_BUCKET SECRET_NAME=$SECRET_NAME COPY_FILES=$COPY_FILES MODIFY_ROWS=$MODIFY_ROWS OVERWRITE=$OVERWRITE MAX_FILES=$MAX_FILES bash ~/run_remote.sh" + "AWS_DEFAULT_REGION=$AWS_DEFAULT_REGION SRC_BUCKET=$SRC_BUCKET DST_BUCKET=$DST_BUCKET SRC_PREFIX=$SRC_PREFIX OVERWRITE=$OVERWRITE MAX_FILES=$MAX_FILES DRY_RUN=$DRY_RUN bash ~/run_remote_rename.sh" # Teardown happens in cleanup() via the EXIT trap. diff --git a/scripts/migrate/run_remote.sh b/scripts/migrate/run_remote.sh deleted file mode 100644 index c41d15240..000000000 --- a/scripts/migrate/run_remote.sh +++ /dev/null @@ -1,47 +0,0 @@ -#!/usr/bin/env bash - -set -euo pipefail - -# --------------------------------------------------------------------------- -# Config (dev account 593025701104) -# --------------------------------------------------------------------------- -export AWS_DEFAULT_REGION="${AWS_DEFAULT_REGION:-us-west-2}" -export S3_BUCKET="${S3_BUCKET:-sds-data-593025701104}" -export SECRET_NAME="${SECRET_NAME:-sdp-database-cred}" - -# Passed through to migrate.py. -export COPY_FILES="${COPY_FILES:-0}" -export MODIFY_ROWS="${MODIFY_ROWS:-0}" -export OVERWRITE="${OVERWRITE:-0}" -export MAX_FILES="${MAX_FILES:-0}" - -# t3 instances uses tmpdir as half of available RAM - not nearly enough for CDF files -export TMPDIR="$HOME/tmp" -mkdir -p "$TMPDIR" - - -# --------------------------------------------------------------------------- -# uv -# --------------------------------------------------------------------------- -sudo dnf install -y git >/dev/null -if ! command -v uv >/dev/null 2>&1; then - curl -LsSf https://astral.sh/uv/install.sh | sh -fi -export PATH="$HOME/.local/bin:$PATH" - -uv venv --python 3.12 .venv -source .venv/bin/activate - -# --------------------------------------------------------------------------- -# deps -# --------------------------------------------------------------------------- -uv pip install \ - "git+https://github.com/IMAP-Science-Operations-Center/sds-data-manager.git@release_version_work" \ - "git+https://github.com/IMAP-Science-Operations-Center/imap_processing.git@new_version_work" \ - "git+https://github.com/IMAP-Science-Operations-Center/imap-data-access.git" \ - "SQLAlchemy<=3.0.0" \ - "pandas>=3.0.3,<4.0.0" \ - psycopg2-binary \ - boto3 - -uv run python migrate.py diff --git a/scripts/migrate/run_remote_rename.sh b/scripts/migrate/run_remote_rename.sh new file mode 100755 index 000000000..62dd16170 --- /dev/null +++ b/scripts/migrate/run_remote_rename.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash + +set -euo pipefail + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- +export AWS_DEFAULT_REGION="${AWS_DEFAULT_REGION:-us-west-2}" + +# Source and destination S3 buckets. DST_BUCKET must differ from SRC_BUCKET. +export SRC_BUCKET="${SRC_BUCKET:-deprecated-data-archive}" +export DST_BUCKET="${DST_BUCKET:-todo}" + +# Passed through to rename.py. +export SRC_PREFIX="${SRC_PREFIX:-imap/}" +export MAJOR_VERSION="${MAJOR_VERSION:-1}" +export OVERWRITE="${OVERWRITE:-0}" +export MAX_FILES="${MAX_FILES:-0}" +export MAX_WORKERS="${MAX_WORKERS:-0}" +export DRY_RUN="${DRY_RUN:-1}" + +# The default instance temp dir is too small for CDF files; use the large EBS +# root volume instead. +export TMPDIR="$HOME/tmp" +mkdir -p "$TMPDIR" + +# --------------------------------------------------------------------------- +# uv + python deps +# --------------------------------------------------------------------------- +if ! command -v uv >/dev/null 2>&1; then + curl -LsSf https://astral.sh/uv/install.sh | sh +fi +export PATH="$HOME/.local/bin:$PATH" + +uv venv --python 3.12 .venv +source .venv/bin/activate + +# rename.py needs only spacepy (CDF rewrite) and boto3 (S3) - none of the +# heavier imap_processing / sds-data-manager stack that migrate.py pulls in. +uv pip install spacepy numpy boto3 + +uv run python rename.py