titan-iac/services/cassandra/migration-jobs/artifact-copy-veles-to-cassandra-job.yaml

# services/cassandra/migration-jobs/artifact-copy-veles-to-cassandra-job.yaml
# Suspended by default. Run only after scaling the Veles artifact exporter to 1.
apiVersion: batch/v1
kind: Job
metadata:
  name: cassandra-artifact-copy-from-veles-2
  namespace: cassandra
spec:
  suspend: true
  backoffLimit: 0
  activeDeadlineSeconds: 43200
  ttlSecondsAfterFinished: 86400
  template:
    spec:
      serviceAccountName: cassandra-artifact-migration
      automountServiceAccountToken: false
      restartPolicy: Never
      nodeSelector:
        cassandra.bstein.dev/node-pool: oceanus
      tolerations:
        - key: veles.bstein.dev/simulation
          operator: Equal
          value: "true"
          effect: NoSchedule
      containers:
        - name: copy-artifacts
          image: python:3.12-alpine
          imagePullPolicy: IfNotPresent
          env:
            - name: SOURCE_URL
              value: http://veles-artifact-export-for-cassandra.veles.svc.cluster.local:8765
            - name: DEST_ROOT
              value: /dest
            - name: ALLOW_NONEMPTY_DEST
              value: "0"
            - name: RESET_DESTINATION
              value: "1"
          command: ["python", "-c"]
          args:
            - |
              from __future__ import annotations

              from pathlib import Path
              import hashlib
              import json
              import os
              import shutil
              import tarfile
              import time
              import urllib.request

              SOURCE_URL = os.environ["SOURCE_URL"].rstrip("/")
              DEST = Path(os.environ.get("DEST_ROOT", "/dest")).resolve()
              ALLOW_NONEMPTY = os.environ.get("ALLOW_NONEMPTY_DEST", "") in {"1", "true", "yes"}
              RESET_DESTINATION = os.environ.get("RESET_DESTINATION", "") in {"1", "true", "yes"}
              EXCLUDE_TOP = {"lost+found", ".cassandra-artifact-migration"}
              REPORT_DIR = DEST / ".cassandra-artifact-migration"

              def visible_entries() -> list[Path]:
                  return [entry for entry in DEST.iterdir() if entry.name not in EXCLUDE_TOP]

              def assert_safe_destination() -> None:
                  DEST.mkdir(parents=True, exist_ok=True)
                  entries = visible_entries()
                  if entries and RESET_DESTINATION:
                      for entry in entries:
                          if entry.is_dir():
                              shutil.rmtree(entry)
                          else:
                              entry.unlink()
                      entries = visible_entries()
                  if entries and not ALLOW_NONEMPTY:
                      names = ", ".join(entry.name for entry in entries[:10])
                      raise SystemExit(f"destination is not empty; refusing to merge into existing data: {names}")
                  REPORT_DIR.mkdir(parents=True, exist_ok=True)

              def fetch_manifest() -> dict[str, object]:
                  with urllib.request.urlopen(f"{SOURCE_URL}/manifest.json", timeout=120) as response:
                      return json.loads(response.read().decode())

              def wait_for_manifest() -> dict[str, object]:
                  last_error = ""
                  for _attempt in range(90):
                      try:
                          return fetch_manifest()
                      except Exception as exc:
                          last_error = str(exc)
                          time.sleep(5)
                  raise SystemExit(f"source artifact exporter did not become ready: {last_error}")

              def safe_extract() -> None:
                  root_text = str(DEST)
                  with urllib.request.urlopen(f"{SOURCE_URL}/archive.tar", timeout=600) as response:
                      with tarfile.open(fileobj=response, mode="r|*") as archive:
                          for member in archive:
                              target = (DEST / member.name).resolve()
                              if target != DEST and not str(target).startswith(root_text + os.sep):
                                  raise SystemExit(f"unsafe archive path: {member.name}")
                              archive.extract(member, path=DEST)

              def digest_file(path: Path) -> str:
                  digest = hashlib.sha256()
                  with path.open("rb") as handle:
                      for chunk in iter(lambda: handle.read(1024 * 1024), b""):
                          digest.update(chunk)
                  return digest.hexdigest()

              def destination_manifest() -> dict[str, object]:
                  files = []
                  total_bytes = 0
                  for path in sorted(DEST.rglob("*")):
                      if not path.is_file():
                          continue
                      rel = path.relative_to(DEST).as_posix()
                      if rel.split("/", 1)[0] in EXCLUDE_TOP:
                          continue
                      size = path.stat().st_size
                      total_bytes += size
                      files.append({"path": rel, "size": size, "sha256": digest_file(path)})
                  manifest_text = "\n".join(
                      f"{item['sha256']} {item['size']} {item['path']}" for item in files
                  )
                  return {
                      "schema": "cassandra.artifact_manifest.v1",
                      "source": "cassandra-artifacts",
                      "file_count": len(files),
                      "total_bytes": total_bytes,
                      "manifest_sha256": hashlib.sha256(manifest_text.encode()).hexdigest(),
                      "files": files,
                  }

              def comparable(manifest: dict[str, object]) -> dict[str, object]:
                  return {
                      "file_count": manifest.get("file_count"),
                      "total_bytes": manifest.get("total_bytes"),
                      "manifest_sha256": manifest.get("manifest_sha256"),
                      "files": manifest.get("files"),
                  }

              assert_safe_destination()
              source = wait_for_manifest()
              (REPORT_DIR / "source_manifest.json").write_text(json.dumps(source, indent=2, sort_keys=True))
              safe_extract()
              dest = destination_manifest()
              (REPORT_DIR / "destination_manifest.json").write_text(json.dumps(dest, indent=2, sort_keys=True))
              summary = {
                  "source_file_count": source.get("file_count"),
                  "destination_file_count": dest.get("file_count"),
                  "source_total_bytes": source.get("total_bytes"),
                  "destination_total_bytes": dest.get("total_bytes"),
                  "source_manifest_sha256": source.get("manifest_sha256"),
                  "destination_manifest_sha256": dest.get("manifest_sha256"),
                  "verified": comparable(source) == comparable(dest),
              }
              (REPORT_DIR / "copy_summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True))
              if not summary["verified"]:
                  raise SystemExit(f"artifact copy verification failed: {summary}")
              print(json.dumps(summary, sort_keys=True), flush=True)
          resources:
            requests:
              cpu: 500m
              memory: 512Mi
            limits:
              cpu: "2"
              memory: 2Gi
          volumeMounts:
            - name: artifacts
              mountPath: /dest
      volumes:
        - name: artifacts
          persistentVolumeClaim:
            claimName: cassandra-artifacts