123 lines
5.3 KiB
YAML
123 lines
5.3 KiB
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
|