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

# services/cassandra/migration-jobs/db-copy-veles-to-cassandra-job.yaml
# Suspended by default. Unsuspend after cassandra-postgres is ready and before
# enabling generator/retention workers. This copies from Veles to Cassandra and
# verifies table sets, row counts, and per-table content hashes.
apiVersion: batch/v1
kind: Job
metadata:
  name: cassandra-db-copy-from-veles-3
  namespace: cassandra
spec:
  suspend: true
  backoffLimit: 0
  ttlSecondsAfterFinished: 86400
  template:
    metadata:
      annotations:
        vault.hashicorp.com/agent-inject: "true"
        vault.hashicorp.com/agent-init-first: "true"
        vault.hashicorp.com/agent-pre-populate-only: "true"
        vault.hashicorp.com/role: "cassandra-migration"
        vault.hashicorp.com/agent-inject-secret-source-db.sh: "kv/data/atlas/veles/veles-db"
        vault.hashicorp.com/agent-inject-template-source-db.sh: |
          {{- with secret "kv/data/atlas/veles/veles-db" }}
          export SOURCE_DATABASE_URL="{{ .Data.data.DATABASE_URL }}"
          {{- end }}
        vault.hashicorp.com/agent-inject-secret-target-db.sh: "kv/data/atlas/cassandra/cassandra-db"
        vault.hashicorp.com/agent-inject-template-target-db.sh: |
          {{- with secret "kv/data/atlas/cassandra/cassandra-db" }}
          export TARGET_DATABASE_URL="{{ .Data.data.DATABASE_URL }}"
          {{- end }}
    spec:
      serviceAccountName: cassandra-db-migration
      restartPolicy: Never
      nodeSelector:
        cassandra.bstein.dev/node-pool: oceanus
        kubernetes.io/arch: amd64
      tolerations:
        - key: veles.bstein.dev/simulation
          operator: Equal
          value: "true"
          effect: NoSchedule
      containers:
        - name: copy-and-verify
          image: postgres:15
          command: ["/bin/bash", "-c"]
          args:
            - |
              set -euo pipefail
              . /vault/secrets/source-db.sh
              . /vault/secrets/target-db.sh
              : "${SOURCE_DATABASE_URL:?missing source database url}"
              : "${TARGET_DATABASE_URL:?missing target database url}"

              export PGCONNECT_TIMEOUT=10
              for attempt in $(seq 1 60); do
                if pg_isready -d "${SOURCE_DATABASE_URL}" >/dev/null 2>&1 \
                  && pg_isready -d "${TARGET_DATABASE_URL}" >/dev/null 2>&1; then
                  break
                fi
                if [ "${attempt}" = "60" ]; then
                  echo "source or target database did not become ready" >&2
                  exit 1
                fi
                sleep 5
              done

              dump_path=/tmp/veles-to-cassandra.dump
              pg_dump --format=custom --no-owner --no-acl \
                --dbname="${SOURCE_DATABASE_URL}" \
                --file="${dump_path}"
              psql -X --set=ON_ERROR_STOP=1 --dbname="${TARGET_DATABASE_URL}" \
                --command="DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public; GRANT ALL ON SCHEMA public TO cassandra; GRANT ALL ON SCHEMA public TO public;"
              pg_restore --no-owner --no-acl \
                --dbname="${TARGET_DATABASE_URL}" \
                "${dump_path}"

              table_sql="select schemaname || E'\t' || tablename from pg_tables where schemaname not in ('pg_catalog','information_schema') order by 1"
              psql -XAt --dbname="${SOURCE_DATABASE_URL}" --command="${table_sql}" > /tmp/source.tables
              psql -XAt --dbname="${TARGET_DATABASE_URL}" --command="${table_sql}" > /tmp/target.tables
              diff -u /tmp/source.tables /tmp/target.tables

              checksum_sql() {
                schema="$1"
                table="$2"
                printf '%s\n' \
                  "SELECT format(" \
                  "  'select count(*)::text || E''\t'' || coalesce(md5(string_agg(md5(row_to_json(t)::text), '''' order by md5(row_to_json(t)::text))), md5('''')) from %I.%I t'," \
                  "  :'schema'," \
                  "  :'table'" \
                  ")\\gexec" > /tmp/checksum.sql
                psql -XAt --set=ON_ERROR_STOP=1 \
                  --set=schema="${schema}" \
                  --set=table="${table}" \
                  --dbname="$3" \
                  --file=/tmp/checksum.sql
              }

              failures=0
              while IFS="$(printf '\t')" read -r schema table; do
                [ -n "${schema}" ] || continue
                source_sum="$(checksum_sql "${schema}" "${table}" "${SOURCE_DATABASE_URL}")"
                target_sum="$(checksum_sql "${schema}" "${table}" "${TARGET_DATABASE_URL}")"
                if [ "${source_sum}" != "${target_sum}" ]; then
                  echo "checksum mismatch for ${schema}.${table}: source=${source_sum} target=${target_sum}" >&2
                  failures=$((failures + 1))
                else
                  echo "verified ${schema}.${table}: ${source_sum}"
                fi
              done < /tmp/source.tables

              if [ "${failures}" -ne 0 ]; then
                echo "database copy integrity verification failed for ${failures} table(s)" >&2
                exit 1
              fi
              echo "database copy integrity verification passed"
          resources:
            requests:
              cpu: 500m
              memory: 1Gi
            limits:
              cpu: "2"
              memory: 4Gi