# services/veles/migration-jobs/cassandra-artifact-export.yaml # Scale to one only during the Veles -> Cassandra artifact copy, then return to # zero after the Cassandra import job verifies the copied content. apiVersion: v1 kind: Service metadata: name: veles-artifact-export-for-cassandra namespace: veles spec: selector: app: veles-artifact-export-for-cassandra ports: - name: http port: 8765 targetPort: http --- apiVersion: apps/v1 kind: Deployment metadata: name: veles-artifact-export-for-cassandra namespace: veles spec: replicas: 0 selector: matchLabels: app: veles-artifact-export-for-cassandra template: metadata: labels: app: veles-artifact-export-for-cassandra spec: automountServiceAccountToken: false nodeSelector: cassandra.bstein.dev/node-pool: oceanus tolerations: - key: veles.bstein.dev/simulation operator: Equal value: "true" effect: NoSchedule containers: - name: exporter image: python:3.12-alpine imagePullPolicy: IfNotPresent ports: - name: http containerPort: 8765 env: - name: SOURCE_ROOT value: /source command: ["python", "-c"] args: - | from __future__ import annotations from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import hashlib import json import os import tarfile ROOT = Path(os.environ.get("SOURCE_ROOT", "/source")).resolve() EXCLUDE_TOP = {"lost+found", ".cassandra-artifact-migration"} def iter_files(): for path in sorted(ROOT.rglob("*")): if not path.is_file(): continue rel = path.relative_to(ROOT).as_posix() if rel.split("/", 1)[0] in EXCLUDE_TOP: continue yield path, rel 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 build_manifest() -> dict[str, object]: files = [] total_bytes = 0 for path, rel in iter_files(): 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": "veles-artifacts", "file_count": len(files), "total_bytes": total_bytes, "manifest_sha256": hashlib.sha256(manifest_text.encode()).hexdigest(), "files": files, } MANIFEST = build_manifest() MANIFEST_PATHS = [str(item["path"]) for item in MANIFEST["files"]] class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: if self.path == "/healthz": self.send_response(200) self.end_headers() self.wfile.write(b"ok\n") return if self.path == "/manifest.json": payload = json.dumps(MANIFEST, sort_keys=True).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) return if self.path == "/archive.tar": self.send_response(200) self.send_header("Content-Type", "application/x-tar") self.end_headers() with tarfile.open(fileobj=self.wfile, mode="w|") as archive: for rel in MANIFEST_PATHS: path = ROOT / rel if not path.is_file(): raise FileNotFoundError(rel) with path.open("rb") as handle: info = archive.gettarinfo(str(path), arcname=rel) archive.addfile(info, handle) return self.send_response(404) self.end_headers() def log_message(self, fmt: str, *args: object) -> None: print(fmt % args, flush=True) print( "ready file_count={file_count} total_bytes={total_bytes} manifest={manifest_sha256}".format( **MANIFEST ), flush=True, ) ThreadingHTTPServer(("0.0.0.0", 8765), Handler).serve_forever() resources: requests: cpu: 250m memory: 512Mi limits: cpu: "2" memory: 2Gi volumeMounts: - name: artifacts mountPath: /source readOnly: true volumes: - name: artifacts persistentVolumeClaim: claimName: veles-artifacts readOnly: true