479 lines
17 KiB
Python
479 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
SCRIPT = ROOT / "scripts" / "backup" / "clickhouse-native-backup.sh"
|
|
PLAYBOOK = ROOT / "infra" / "ansible" / "playbooks" / "110-devops.yml"
|
|
|
|
|
|
def _write_executable(path: Path, text: str) -> None:
|
|
path.write_text(text, encoding="utf-8")
|
|
path.chmod(0o755)
|
|
|
|
|
|
def _install_fakes(fake_bin: Path) -> None:
|
|
_write_executable(
|
|
fake_bin / "timeout",
|
|
"""\
|
|
#!/bin/bash
|
|
set -eu
|
|
while [[ "${1:-}" == --kill-after=* ]]; do shift; done
|
|
shift
|
|
exec "$@"
|
|
""",
|
|
)
|
|
_write_executable(fake_bin / "flock", "#!/bin/bash\nexit 0\n")
|
|
_write_executable(
|
|
fake_bin / "sha256sum",
|
|
"""\
|
|
#!/usr/bin/env python3
|
|
import hashlib
|
|
import sys
|
|
|
|
if len(sys.argv) == 1:
|
|
payload = sys.stdin.buffer.read()
|
|
print(f"{hashlib.sha256(payload).hexdigest()} -")
|
|
else:
|
|
with open(sys.argv[1], "rb") as source:
|
|
payload = source.read()
|
|
print(f"{hashlib.sha256(payload).hexdigest()} {sys.argv[1]}")
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "unzip",
|
|
"""\
|
|
#!/bin/bash
|
|
set -eu
|
|
printf 'unzip %s\\n' "$*" >> "${TEST_EVENT_LOG:?}"
|
|
exit "${FAKE_UNZIP_STATUS:-0}"
|
|
""",
|
|
)
|
|
_write_executable(
|
|
fake_bin / "docker",
|
|
"""\
|
|
#!/bin/bash
|
|
set -eu
|
|
printf 'docker %s\\n' "$*" >> "${TEST_EVENT_LOG:?}"
|
|
|
|
find_arg() {
|
|
local wanted="$1"
|
|
shift
|
|
while [ "$#" -gt 0 ]; do
|
|
if [ "$1" = "$wanted" ]; then
|
|
printf '%s\\n' "${2:-}"
|
|
return 0
|
|
fi
|
|
shift
|
|
done
|
|
return 1
|
|
}
|
|
|
|
case "${1:-}" in
|
|
inspect)
|
|
printf 'sha256:clickhouse-test-id\\ttrue\\n'
|
|
;;
|
|
cp)
|
|
[ -e "${TEST_SERVER_ARTIFACT:?}" ] || exit 44
|
|
printf 'clickhouse-native-zip-bytes\\n' > "${3:?}"
|
|
;;
|
|
exec)
|
|
shift
|
|
container="${1:?}"
|
|
shift
|
|
case "${1:-}" in
|
|
test)
|
|
shift
|
|
if [ "${1:-}" = "!" ]; then
|
|
[ ! -e "${TEST_SERVER_ARTIFACT:?}" ]
|
|
elif [ "${1:-}" = "-d" ] || [ "${1:-}" = "-w" ]; then
|
|
[ "${FAKE_DISK_READY:-1}" = "1" ]
|
|
[ "${2:-}" = "${FAKE_DISK_PATH:-/backups/}" ]
|
|
else
|
|
exit 45
|
|
fi
|
|
;;
|
|
rm)
|
|
rm -f "${TEST_SERVER_ARTIFACT:?}"
|
|
;;
|
|
clickhouse-client)
|
|
query="$(find_arg --query "$@")"
|
|
operation_id="$(find_arg --param_operation_id "$@" || true)"
|
|
artifact="$(find_arg --param_artifact "$@" || true)"
|
|
disk="$(find_arg --param_disk "$@" || true)"
|
|
case "$query" in
|
|
'SELECT version()')
|
|
printf '25.5.1.1\\n'
|
|
;;
|
|
*"database='system' AND name='backups'"*)
|
|
printf '%s\\n' "${FAKE_SYSTEM_BACKUPS_COUNT:-1}"
|
|
;;
|
|
'CHECK GRANT BACKUP ON *.*')
|
|
printf '%s\\n' "${FAKE_BACKUP_GRANT:-1}"
|
|
;;
|
|
*"FROM system.disks"*)
|
|
printf '%s\\n' "${FAKE_DISK_PATH:-/backups/}"
|
|
;;
|
|
*"FROM system.databases"*)
|
|
printf '%s\\n' signoz_analytics signoz_logs signoz_metadata \
|
|
signoz_meter signoz_metrics signoz_traces
|
|
if [ "${FAKE_EXTRA_DATABASE:-0}" = "1" ]; then
|
|
printf 'unexpected_db\\n'
|
|
fi
|
|
;;
|
|
*"FROM system.tables WHERE database IN"*)
|
|
schema_hash="$(printf 'A%.0s' {1..64})"
|
|
printf 'signoz_analytics\\tanalytics\\tReplicatedMergeTree\\t10\\t1000\\t%s\\n' "$schema_hash"
|
|
printf 'signoz_logs\\tlogs_v2\\tReplicatedMergeTree\\t20\\t2000\\t%s\\n' "$schema_hash"
|
|
printf 'signoz_metadata\\tmetadata\\tMergeTree\\t3\\t300\\t%s\\n' "$schema_hash"
|
|
printf 'signoz_meter\\tmeter\\tDistributed\\t0\\t0\\t%s\\n' "$schema_hash"
|
|
printf 'signoz_metrics\\tsamples_v4\\tReplicatedMergeTree\\t40\\t4000\\t%s\\n' "$schema_hash"
|
|
printf 'signoz_traces\\tsignoz_index_v3\\tReplicatedMergeTree\\t50\\t5000\\t%s\\n' "$schema_hash"
|
|
;;
|
|
BACKUP*)
|
|
case "$query" in
|
|
*" SETTINGS id='${EXPECTED_FAKE_OPERATION_ID:?}' ASYNC")
|
|
operation_id="${EXPECTED_FAKE_OPERATION_ID}"
|
|
;;
|
|
*) exit 49 ;;
|
|
esac
|
|
printf '%s\\n' "$operation_id" > "${TEST_OPERATION_ID:?}"
|
|
printf '%s\\n' "$artifact" > "${TEST_ARTIFACT_NAME:?}"
|
|
printf '%s\\n' "$disk" > "${TEST_DISK_NAME:?}"
|
|
: > "${TEST_SUBMITTED:?}"
|
|
: > "${TEST_SERVER_ARTIFACT:?}"
|
|
printf '%s\\tCREATING_BACKUP\\n' "$operation_id"
|
|
;;
|
|
*"FROM system.backups WHERE id="*)
|
|
mode="${FAKE_STATUS_MODE:-success}"
|
|
if [ "$mode" = "existing_success" ] || [ -e "${TEST_SUBMITTED:?}" ]; then
|
|
artifact_name="$(cat "${TEST_ARTIFACT_NAME:?}" 2>/dev/null || true)"
|
|
[ -n "$artifact_name" ] || artifact_name="${EXPECTED_FAKE_ARTIFACT:?}"
|
|
backup_name="Disk('backups', '$artifact_name')"
|
|
case "$mode" in
|
|
failed)
|
|
printf '%s\\tBACKUP_FAILED\\t0\\t0\\t0\\t0\\t434F44453A20353938\\t%s\\n' \
|
|
"$backup_name" "$(printf '1%.0s' {1..64})"
|
|
;;
|
|
*)
|
|
: > "${TEST_SERVER_ARTIFACT:?}"
|
|
printf '%s\\tBACKUP_CREATED\\t12\\t1200\\t2400\\t1100\\tnone\\t%s\\n' \
|
|
"$backup_name" "$(printf '0%.0s' {1..64})"
|
|
;;
|
|
esac
|
|
fi
|
|
;;
|
|
*) exit 46 ;;
|
|
esac
|
|
;;
|
|
*) exit 47 ;;
|
|
esac
|
|
;;
|
|
*) exit 48 ;;
|
|
esac
|
|
""",
|
|
)
|
|
|
|
|
|
def _harness(
|
|
tmp_path: Path,
|
|
*,
|
|
status_mode: str = "success",
|
|
extra_database: bool = False,
|
|
disk_ready: bool = True,
|
|
unzip_status: int = 0,
|
|
with_hook: bool = False,
|
|
) -> tuple[dict[str, str], dict[str, Path]]:
|
|
fake_bin = tmp_path / "bin"
|
|
output = tmp_path / "output"
|
|
receipts = tmp_path / "receipts"
|
|
fake_bin.mkdir()
|
|
output.mkdir()
|
|
receipts.mkdir()
|
|
_install_fakes(fake_bin)
|
|
|
|
event_log = tmp_path / "events.log"
|
|
event_log.touch()
|
|
paths = {
|
|
"output": output,
|
|
"receipts": receipts,
|
|
"events": event_log,
|
|
"submitted": tmp_path / "submitted",
|
|
"server_artifact": tmp_path / "server-artifact",
|
|
"operation_id": tmp_path / "operation-id",
|
|
"artifact_name": tmp_path / "artifact-name",
|
|
"disk_name": tmp_path / "disk-name",
|
|
"hook_events": tmp_path / "hook-events.log",
|
|
}
|
|
identity_hash = (
|
|
subprocess.check_output(
|
|
[str(fake_bin / "sha256sum")],
|
|
input=b"trace-native\0run-native\0P0-OBS-002",
|
|
)
|
|
.decode()
|
|
.split()[0]
|
|
)
|
|
expected_artifact = f"signoz-{identity_hash}.zip"
|
|
|
|
env = {
|
|
**os.environ,
|
|
"PATH": f"{fake_bin}:{os.environ['PATH']}",
|
|
"TRACE_ID": "trace-native",
|
|
"RUN_ID": "run-native",
|
|
"WORK_ITEM_ID": "P0-OBS-002",
|
|
"CLICKHOUSE_NATIVE_OUTPUT_DIR": str(output),
|
|
"CLICKHOUSE_NATIVE_RECEIPT_ROOT": str(receipts),
|
|
"CLICKHOUSE_NATIVE_COMMAND_TIMEOUT_SECONDS": "2",
|
|
"CLICKHOUSE_NATIVE_BACKUP_TIMEOUT_SECONDS": "2",
|
|
"CLICKHOUSE_NATIVE_POLL_INTERVAL_SECONDS": "1",
|
|
"CLICKHOUSE_NATIVE_KILL_AFTER_SECONDS": "1",
|
|
"TEST_EVENT_LOG": str(event_log),
|
|
"TEST_SUBMITTED": str(paths["submitted"]),
|
|
"TEST_SERVER_ARTIFACT": str(paths["server_artifact"]),
|
|
"TEST_OPERATION_ID": str(paths["operation_id"]),
|
|
"TEST_ARTIFACT_NAME": str(paths["artifact_name"]),
|
|
"TEST_DISK_NAME": str(paths["disk_name"]),
|
|
"EXPECTED_FAKE_ARTIFACT": expected_artifact,
|
|
"EXPECTED_FAKE_OPERATION_ID": f"awoooi-{identity_hash}",
|
|
"FAKE_STATUS_MODE": status_mode,
|
|
"FAKE_EXTRA_DATABASE": "1" if extra_database else "0",
|
|
"FAKE_DISK_READY": "1" if disk_ready else "0",
|
|
"FAKE_UNZIP_STATUS": str(unzip_status),
|
|
}
|
|
if with_hook:
|
|
hook = tmp_path / "restore-verifier-hook.sh"
|
|
_write_executable(
|
|
hook,
|
|
"""\
|
|
#!/bin/bash
|
|
set -eu
|
|
if [ "$1" = "--apply" ]; then
|
|
test -f "$CLICKHOUSE_NATIVE_ARTIFACT_PATH"
|
|
test -f "$CLICKHOUSE_NATIVE_SOURCE_INVENTORY_PATH"
|
|
test -f "$CLICKHOUSE_NATIVE_MANIFEST_PATH"
|
|
grep -q '"status":"BACKUP_CREATED"' "$CLICKHOUSE_NATIVE_MANIFEST_PATH"
|
|
fi
|
|
printf '%s trace=%s run=%s work=%s dbs=%s inventory=%s manifest=%s\\n' \
|
|
"$1" "$TRACE_ID" "$RUN_ID" "$WORK_ITEM_ID" \
|
|
"$CLICKHOUSE_NATIVE_EXPECTED_DATABASES" \
|
|
"$CLICKHOUSE_NATIVE_SOURCE_INVENTORY_PATH" \
|
|
"${CLICKHOUSE_NATIVE_MANIFEST_PATH:-none}" \
|
|
>> "${TEST_HOOK_EVENTS:?}"
|
|
""",
|
|
)
|
|
env["CLICKHOUSE_NATIVE_POST_VERIFY_HOOK"] = str(hook)
|
|
env["TEST_HOOK_EVENTS"] = str(paths["hook_events"])
|
|
return env, paths
|
|
|
|
|
|
def _run(env: dict[str, str], mode: str) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["bash", str(SCRIPT), mode],
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=15,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def _receipts(path: Path) -> list[dict[str, object]]:
|
|
receipt_file = path / "run-native" / "receipts.jsonl"
|
|
return [
|
|
json.loads(line)
|
|
for line in receipt_file.read_text(encoding="utf-8").splitlines()
|
|
]
|
|
|
|
|
|
def test_check_validates_exact_scope_without_backup_or_artifact_write(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
env, paths = _harness(tmp_path)
|
|
result = _run(env, "--check")
|
|
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
events = paths["events"].read_text(encoding="utf-8")
|
|
assert "BACKUP DATABASE" not in events
|
|
assert "docker cp" not in events
|
|
assert "docker exec signoz-clickhouse rm" not in events
|
|
assert not list(paths["output"].iterdir())
|
|
receipts = _receipts(paths["receipts"])
|
|
assert receipts[-1]["terminal"] == "check_pass_no_write"
|
|
|
|
|
|
def test_apply_requires_same_run_check_receipt(tmp_path: Path) -> None:
|
|
env, paths = _harness(tmp_path)
|
|
result = _run(env, "--apply")
|
|
|
|
assert result.returncode != 0
|
|
assert "same_run_check_pass_receipt_required_before_apply" in result.stderr
|
|
assert paths["events"].read_text(encoding="utf-8") == ""
|
|
|
|
|
|
def test_apply_submits_exact_six_database_backup_and_verifies_zip(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
env, paths = _harness(tmp_path)
|
|
assert _run(env, "--check").returncode == 0
|
|
result = _run(env, "--apply")
|
|
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
events = paths["events"].read_text(encoding="utf-8")
|
|
assert "BACKUP DATABASE signoz_analytics,DATABASE signoz_logs" in events
|
|
assert "DATABASE signoz_metadata,DATABASE signoz_meter" in events
|
|
assert "DATABASE signoz_metrics,DATABASE signoz_traces" in events
|
|
assert "BACKUP ALL" not in events
|
|
backup_line = next(
|
|
line for line in events.splitlines() if "--query BACKUP " in line
|
|
)
|
|
assert " SETTINGS id='awoooi-" in backup_line
|
|
assert "id={operation_id:String}" not in backup_line
|
|
assert "--param_operation_id" not in backup_line
|
|
assert "docker cp signoz-clickhouse:/backups/signoz-" in events
|
|
assert "unzip -tq" in events
|
|
assert "docker exec signoz-clickhouse rm -f -- /backups/signoz-" in events
|
|
assert not paths["server_artifact"].exists()
|
|
outputs = list(paths["output"].glob("clickhouse-native-*.zip"))
|
|
assert len(outputs) == 1
|
|
manifest = json.loads(Path(str(outputs[0]) + ".manifest.json").read_text())
|
|
assert manifest["databases"] == [
|
|
"signoz_analytics",
|
|
"signoz_logs",
|
|
"signoz_metadata",
|
|
"signoz_meter",
|
|
"signoz_metrics",
|
|
"signoz_traces",
|
|
]
|
|
inventory = list(paths["output"].glob("*.source-inventory.tsv"))
|
|
assert len(inventory) == 1
|
|
assert "ReplicatedMergeTree" in inventory[0].read_text(encoding="utf-8")
|
|
assert "Distributed" in inventory[0].read_text(encoding="utf-8")
|
|
|
|
|
|
def test_database_allowlist_drift_fails_check(tmp_path: Path) -> None:
|
|
env, _paths = _harness(tmp_path, extra_database=True)
|
|
result = _run(env, "--check")
|
|
assert result.returncode != 0
|
|
assert "signoz_database_allowlist_drift" in result.stderr
|
|
|
|
|
|
def test_missing_backup_grant_fails_check_before_submission(tmp_path: Path) -> None:
|
|
env, paths = _harness(tmp_path)
|
|
env["FAKE_BACKUP_GRANT"] = "0"
|
|
|
|
result = _run(env, "--check")
|
|
|
|
assert result.returncode != 0
|
|
assert "native_backup_grant_absent" in result.stderr
|
|
assert not paths["submitted"].exists()
|
|
|
|
|
|
def test_missing_runtime_backup_disk_fails_check(tmp_path: Path) -> None:
|
|
env, _paths = _harness(tmp_path, disk_ready=False)
|
|
result = _run(env, "--check")
|
|
assert result.returncode != 0
|
|
assert "backup_disk_directory_absent" in result.stderr
|
|
|
|
|
|
def test_failed_async_terminal_has_exact_cleanup_and_no_local_artifact(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
env, paths = _harness(tmp_path, status_mode="failed")
|
|
assert _run(env, "--check").returncode == 0
|
|
result = _run(env, "--apply")
|
|
|
|
assert result.returncode != 0
|
|
assert "native_backup_terminal_BACKUP_FAILED" in result.stderr
|
|
assert not paths["server_artifact"].exists()
|
|
assert not list(paths["output"].iterdir())
|
|
receipt_dir = paths["receipts"] / "run-native"
|
|
assert list(receipt_dir.glob("*.stderr"))
|
|
assert list(receipt_dir.glob("*.status"))
|
|
bounded_errors = list(receipt_dir.glob("*-backup-error-bounded.hex"))
|
|
assert len(bounded_errors) == 1
|
|
assert bounded_errors[0].read_text(encoding="utf-8") == ("434F44453A20353938\n")
|
|
|
|
|
|
def test_zip_failure_removes_local_partial_and_preserves_server_evidence(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
env, paths = _harness(tmp_path, unzip_status=7)
|
|
assert _run(env, "--check").returncode == 0
|
|
result = _run(env, "--apply")
|
|
|
|
assert result.returncode != 0
|
|
assert "native_backup_archive_integrity_failed" in result.stderr
|
|
assert paths["server_artifact"].exists()
|
|
assert not list(paths["output"].iterdir())
|
|
receipts = _receipts(paths["receipts"])
|
|
assert any(
|
|
item["stage"] == "cleanup" and item["terminal"] == "preserved"
|
|
for item in receipts
|
|
)
|
|
|
|
|
|
def test_restore_verifier_hook_uses_check_and_apply_contract(tmp_path: Path) -> None:
|
|
env, paths = _harness(tmp_path, with_hook=True)
|
|
assert _run(env, "--check").returncode == 0
|
|
result = _run(env, "--apply")
|
|
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
hook_events = paths["hook_events"].read_text(encoding="utf-8").splitlines()
|
|
assert sum(event.startswith("--check ") for event in hook_events) == 1
|
|
assert sum(event.startswith("--apply ") for event in hook_events) == 1
|
|
assert all("signoz_analytics,signoz_logs" in event for event in hook_events)
|
|
assert any(
|
|
"manifest=" in event and "manifest=none" not in event
|
|
for event in hook_events
|
|
if event.startswith("--apply ")
|
|
)
|
|
|
|
|
|
def test_verified_local_artifact_is_idempotently_reused(tmp_path: Path) -> None:
|
|
env, paths = _harness(tmp_path)
|
|
assert _run(env, "--check").returncode == 0
|
|
assert _run(env, "--apply").returncode == 0
|
|
submissions_before = (
|
|
paths["events"].read_text(encoding="utf-8").count("BACKUP DATABASE")
|
|
)
|
|
|
|
result = _run(env, "--apply")
|
|
|
|
assert result.returncode == 0, result.stdout + result.stderr
|
|
submissions_after = (
|
|
paths["events"].read_text(encoding="utf-8").count("BACKUP DATABASE")
|
|
)
|
|
assert submissions_after == submissions_before
|
|
receipts = _receipts(paths["receipts"])
|
|
assert any(
|
|
item["stage"] == "idempotency" and item["terminal"] == "reused"
|
|
for item in receipts
|
|
)
|
|
|
|
|
|
def test_source_contract_has_no_raw_volume_fallback_and_is_deployed() -> None:
|
|
source = SCRIPT.read_text(encoding="utf-8")
|
|
playbook = PLAYBOOK.read_text(encoding="utf-8")
|
|
|
|
assert "BACKUP ${BACKUP_SCOPE_SQL}" in source
|
|
assert "SETTINGS id='${OPERATION_ID}' ASYNC" in source
|
|
assert "SETTINGS id={operation_id:String}" not in source
|
|
assert "BACKUP ALL" not in source
|
|
assert ".zip" in source
|
|
assert "unzip -tq" in source
|
|
assert "--format $'{{.Id}}\\t{{.State.Running}}'" in source
|
|
assert "--format '{{.Id}}\\t{{.State.Running}}'" not in source
|
|
assert "docker run" not in source
|
|
assert "clickhouse-native-backup.sh" in playbook
|
|
result = subprocess.run(
|
|
["bash", "-n", str(SCRIPT)],
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|