titan-iac/services/veles/migration-jobs/cassandra-artifact-export.yaml

# 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