wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity
Files
227189cb91aafc82a490e73dc25eb53552c0cf35
awoooi/scripts/backup/tests/test_signoz_backup_canary_wrapper.py

824 lines
30 KiB
Python
Raw Blame History

from __future__ import annotations
import json
import os
import stat
import subprocess
import time
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[3]
WRAPPER = ROOT / "scripts" / "backup" / "run-signoz-backup-canary.sh"
PLAYBOOK = ROOT / "infra" / "ansible" / "playbooks" / "110-devops.yml"
ANSIBLE_VALIDATE = ROOT / "scripts" / "ops" / "ansible-validate.sh"
READINESS_AUDIT = (
ROOT / "scripts" / "reboot-recovery" / "reboot-recovery-readiness-audit.sh"
)
def _write_executable(path: Path, text: str) -> None:
path.write_text(text, encoding="utf-8")
path.chmod(0o755)
def _monitor_source(cooldown_dir: Path, *, omit_contract: bool = False) -> str:
cooldown_log = (
'log "COOLDOWN: ${container} skipped"'
if not omit_contract
else 'log "maintenance skip for ${container}"'
)
return f"""\
#!/bin/bash
: "${{ACTION_COOLDOWN_SECONDS:=${{SEND_COOLDOWN_SECONDS}}}}"
: "${{COOLDOWN_DIR:={cooldown_dir}}}"
: "${{EXCLUDE_CONTAINERS:=signoz-clickhouse}}"
is_in_cooldown() {{
local container="$1"
local cooldown_file="${{COOLDOWN_DIR}}/${{container}}.cooldown"
local last_sent now elapsed
last_sent=$(cat "$cooldown_file")
now=$(date +%s)
elapsed=$(( now - last_sent ))
if (( elapsed < ACTION_COOLDOWN_SECONDS )); then
{cooldown_log}
return 0
fi
return 1
}}
set_cooldown() {{ :; }}
# A second monitor path may use the same elapsed expression. The wrapper must
# validate the expression inside is_in_cooldown, not require global uniqueness.
# elapsed=$(( now - last_sent ))
while read -r container_name; do
is_in_cooldown "$container_name" && continue
set_cooldown "$container_name"
container="$container_name"
log "AUTO_REPAIR: docker restart ${{container}}"
if docker restart "$container"; then
:
fi
done
"""
def _install_fake_commands(fake_bin: Path) -> None:
_write_executable(
fake_bin / "crontab",
"""\
#!/bin/bash
set -eu
[ "${1:-}" = "-l" ]
printf '*/5 * * * * %s >> %s 2>&1\n' \
"${TEST_MONITOR_SCRIPT:?}" "${TEST_MONITOR_LOG:?}"
if [ "${FAKE_DUPLICATE_CRON:-0}" = "1" ]; then
printf '*/5 * * * * %s >> %s 2>&1\n' \
"${TEST_MONITOR_SCRIPT:?}" "${TEST_MONITOR_LOG:?}"
fi
""",
)
_write_executable(
fake_bin / "docker",
"""\
#!/bin/bash
set -eu
[ "${1:-}" = "inspect" ] || exit 70
case " $* " in
*"{{.State.StartedAt}}"*)
printf 'sha256:collector-test-id\\t%s\\t2026-07-15T00:00:00Z\\t0\\thealthy\\n' \
"$(cat "${TEST_COLLECTOR_STATE:?}")"
;;
*"{{.Id}}"*) printf 'sha256:collector-test-id\n' ;;
*"{{.State.Running}}"*) cat "${TEST_COLLECTOR_STATE:?}" ;;
*) exit 71 ;;
esac
""",
)
_write_executable(
fake_bin / "ss",
"""\
#!/bin/bash
set -eu
while IFS= read -r port; do
[ -n "$port" ] || continue
printf 'LISTEN 0 4096 *:%s *:*\n' "$port"
done < "${TEST_LISTENERS:?}"
""",
)
_write_executable(
fake_bin / "timeout",
"""\
#!/bin/bash
set -eu
while [[ "${1:-}" == --kill-after=* ]]; do
shift
done
[ "$#" -gt 1 ]
shift
if [ "${FAKE_DOCKER_TIMEOUT:-0}" = "1" ] && [ "${1:-}" = "docker" ]; then
exit 124
fi
exec "$@"
""",
)
_write_executable(
fake_bin / "flock",
"""\
#!/bin/bash
if [ -n "${TEST_FLOCK_MARKER:-}" ]; then
printf 'held\n' > "${TEST_FLOCK_MARKER}"
fi
exit 0
""",
)
_write_executable(
fake_bin / "pgrep",
"""\
#!/bin/bash
exit 1
""",
)
_write_executable(
fake_bin / "stat",
"""\
#!/usr/bin/env python3
import os
import stat
import sys
if len(sys.argv) != 4 or sys.argv[1] != "-c":
raise SystemExit(64)
value = os.stat(sys.argv[3])
formats = {
"%a": format(stat.S_IMODE(value.st_mode), "o"),
"%u": str(value.st_uid),
"%g": str(value.st_gid),
"%s": str(value.st_size),
"%d:%i": f"{value.st_dev}:{value.st_ino}",
}
if sys.argv[2] not in formats:
raise SystemExit(65)
print(formats[sys.argv[2]])
""",
)
_write_executable(
fake_bin / "sha256sum",
"""\
#!/usr/bin/env python3
import hashlib
import sys
if len(sys.argv) != 2:
raise SystemExit(64)
with open(sys.argv[1], "rb") as source:
digest = hashlib.sha256(source.read()).hexdigest()
print(f"{digest} {sys.argv[1]}")
""",
)
def _restore_hook_source() -> str:
return """\
#!/bin/bash
set -eu
[ "$#" -eq 1 ]
case "$1" in
--check) mode="check" ;;
--apply) mode="apply" ;;
*) exit 64 ;;
esac
receipt_dir="${CLICKHOUSE_RESTORE_RECEIPT_ROOT:?}/${RUN_ID:?}"
mkdir -p "$receipt_dir"
receipt_log="$receipt_dir/receipts.jsonl"
identity_run="$RUN_ID"
if [ "${FAKE_RECEIPT_FAULT:-valid}" = "forged_identity" ]; then
identity_run="stale-$RUN_ID"
fi
invocation_id="${mode}-fake-restore-$$"
emit() {
printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","invocation_id":"%s","stage":"%s","terminal":"%s","detail":"fake"}\n' \
"$TRACE_ID" "$identity_run" "$WORK_ITEM_ID" "$invocation_id" "$1" "$2" \
>> "$receipt_log"
}
if [ "$mode" = "check" ]; then
emit check check_pass_no_runtime_write
emit cleanup pass
emit terminal check_pass_no_runtime_write
exit 0
fi
emit check pass
emit independent_post_verifier pass
emit cleanup pass
emit terminal verified
if [ "${FAKE_RECEIPT_FAULT:-valid}" != "missing_status" ]; then
cat > "$receipt_dir/status.json" <<EOF
{
"schema_version":"clickhouse_native_restore_drill_status_v1",
"trace_id":"$TRACE_ID",
"run_id":"$identity_run",
"work_item_id":"$WORK_ITEM_ID",
"mode":"apply",
"terminal":"verified",
"exit_code":0,
"network_mode":"docker_internal_only",
"staging_network_mode":"none",
"production_network_attached":false,
"production_restore_performed":false,
"allow_non_empty_tables":false,
"isolated_keeper":true,
"artifact_staging_verified":true,
"clickhouse_artifact_readback_verified":true,
"ephemeral_runtime_created":true,
"check_pass":true,
"apply_pass":true,
"restore_terminal":"RESTORED",
"manifest_parity":1,
"source_database_count":6,
"restored_database_count":6,
"source_table_count":100,
"restored_table_count":100,
"check_table_count":44,
"check_table_pass_count":44,
"critical_nonzero_count":4,
"cleanup_verified":true
}
EOF
if [ "${FAKE_RECEIPT_FAULT:-valid}" = "symlink_status" ]; then
mv "$receipt_dir/status.json" "$receipt_dir/status-target.json"
ln -s "status-target.json" "$receipt_dir/status.json"
fi
fi
if [ "${FAKE_RECEIPT_FAULT:-valid}" = "malformed_restore_receipt" ]; then
printf '{malformed-json\n' >> "$receipt_log"
fi
"""
def _backup_source() -> str:
return """\
#!/bin/bash
set -eu
[ "${BACKUP_SKIP_RETENTION_CLEANUP:-}" = "1" ] || exit 80
[ -n "${TRACE_ID:-}" ] && [ -n "${RUN_ID:-}" ] && [ -n "${WORK_ITEM_ID:-}" ]
[ "${CLICKHOUSE_NATIVE_POST_VERIFY_HOOK:-}" = "${TEST_RESTORE_HOOK:?}" ] || exit 82
[ "${CLICKHOUSE_RESTORE_INVENTORY_HELPER:-}" = "${TEST_INVENTORY_HELPER:?}" ] || exit 83
[ "${CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG:-}" = "${TEST_BACKUP_DISK_CONFIG:?}" ] || exit 84
[ "${CLICKHOUSE_RESTORE_CLUSTER_CONFIG:-}" = "${TEST_CLUSTER_CONFIG:?}" ] || exit 85
[ "${CLICKHOUSE_NATIVE_RECEIPT_ROOT:-}" = "${TEST_NATIVE_RECEIPT_ROOT:?}" ] || exit 86
[ "${CLICKHOUSE_RESTORE_RECEIPT_ROOT:-}" = "${TEST_RESTORE_RECEIPT_ROOT:?}" ] || exit 87
printf '%s\t%s\t%s\n' "$TRACE_ID" "$RUN_ID" "$WORK_ITEM_ID" \
> "${TEST_BACKUP_IDENTITIES:?}"
printf '%s\t%s\t%s\t%s\n' \
"$CLICKHOUSE_NATIVE_POST_VERIFY_HOOK" \
"$CLICKHOUSE_RESTORE_INVENTORY_HELPER" \
"$CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG" \
"$CLICKHOUSE_RESTORE_CLUSTER_CONFIG" \
> "${TEST_RESTORE_CONTRACT:?}"
if [ "${FAKE_BACKUP_STOPS_COLLECTOR:-1}" = "1" ]; then
cat "${TEST_COOLDOWN_FILE:?}" > "${TEST_LEASE_OBSERVED:?}"
printf 'false\n' > "${TEST_COLLECTOR_STATE:?}"
case "${FAKE_MONITOR_MODE:-cooldown}" in
cooldown)
printf '[test] DETECTED: signoz-otel-collector state=exited health=none\n' \
>> "${TEST_MONITOR_LOG:?}"
printf '[test] COOLDOWN: signoz-otel-collector skipped\n' \
>> "${TEST_MONITOR_LOG:?}"
;;
auto_repair)
printf '[test] DETECTED: signoz-otel-collector state=exited health=none\n' \
>> "${TEST_MONITOR_LOG:?}"
printf '[test] COOLDOWN: signoz-otel-collector skipped\n' \
>> "${TEST_MONITOR_LOG:?}"
printf '[test] AUTO_REPAIR: docker restart signoz-otel-collector\n' \
>> "${TEST_MONITOR_LOG:?}"
;;
none) : ;;
*) exit 81 ;;
esac
sleep "${FAKE_BACKUP_SLEEP_SECONDS:-0.2}"
printf 'true\n' > "${TEST_COLLECTOR_STATE:?}"
else
sleep "${FAKE_BACKUP_SLEEP_SECONDS:-0.2}"
native_dir="$CLICKHOUSE_NATIVE_RECEIPT_ROOT/$RUN_ID"
mkdir -p "$native_dir"
native_receipts="$native_dir/receipts.jsonl"
printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","invocation_id":"check-fake-native","stage":"terminal","terminal":"check_pass_no_write","detail":"fake"}\n' \
"$TRACE_ID" "$RUN_ID" "$WORK_ITEM_ID" >> "$native_receipts"
if [ "${FAKE_RECEIPT_FAULT:-valid}" != "no_hook" ]; then
"$CLICKHOUSE_NATIVE_POST_VERIFY_HOOK" --check
printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","invocation_id":"apply-fake-native","stage":"command","terminal":"pass","detail":"step_post_verify_hook_check_exit_0_stdout_fake_stderr_fake"}\n' \
"$TRACE_ID" "$RUN_ID" "$WORK_ITEM_ID" >> "$native_receipts"
"$CLICKHOUSE_NATIVE_POST_VERIFY_HOOK" --apply
printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","invocation_id":"apply-fake-native","stage":"command","terminal":"pass","detail":"step_post_verify_hook_exit_0_stdout_fake_stderr_fake"}\n' \
"$TRACE_ID" "$RUN_ID" "$WORK_ITEM_ID" >> "$native_receipts"
fi
printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","invocation_id":"apply-fake-native","stage":"independent_verifier","terminal":"pass","detail":"fake"}\n' \
"$TRACE_ID" "$RUN_ID" "$WORK_ITEM_ID" >> "$native_receipts"
printf '{"trace_id":"%s","run_id":"%s","work_item_id":"%s","invocation_id":"apply-fake-native","stage":"terminal","terminal":"pass","detail":"fake"}\n' \
"$TRACE_ID" "$RUN_ID" "$WORK_ITEM_ID" >> "$native_receipts"
fi
if [ "${FAKE_DROP_LISTENER:-0}" = "1" ]; then
printf '4317\n' > "${TEST_LISTENERS:?}"
fi
printf '略過 SignOz retention cleanup (BACKUP_SKIP_RETENTION_CLEANUP=1)\n'
if [ "${FAKE_BACKUP_STOPS_COLLECTOR:-1}" = "1" ]; then
printf '[SUCCESS] ========== SignOz 備份完成 (1s) ==========\n'
else
printf '[SUCCESS] ========== SigNoz ClickHouse native 備份完成 (1s) ==========\n'
fi
"""
def _run_canary(
tmp_path: Path,
*,
mode: str = "apply",
prior_lease: bool = True,
monitor_mode: str = "cooldown",
backup_sleep_seconds: float = 0.2,
monitor_interval_seconds: int = 300,
omit_monitor_contract: bool = False,
duplicate_cron: bool = False,
drop_listener: bool = False,
expect_collector_stop: bool = True,
native_restore_contract: str = "valid",
native_receipt_state: str = "valid",
docker_timeout: bool = False,
) -> tuple[subprocess.CompletedProcess[str], dict[str, Path], str]:
if native_receipt_state not in {
"valid",
"no_hook",
"forged_identity",
"missing_status",
"symlink_status",
"malformed_restore_receipt",
}:
raise ValueError(f"unsupported native receipt state: {native_receipt_state}")
fake_bin = tmp_path / "bin"
cooldown_dir = tmp_path / "cooldown"
receipt_root = tmp_path / "receipts"
native_receipt_root = tmp_path / "native-receipts"
restore_receipt_root = tmp_path / "restore-receipts"
fake_bin.mkdir()
cooldown_dir.mkdir()
receipt_root.mkdir()
native_receipt_root.mkdir()
restore_receipt_root.mkdir()
monitor_script = tmp_path / "docker-health-monitor.sh"
monitor_log = tmp_path / "monitor.log"
backup_script = tmp_path / "backup-signoz.sh"
collector_state = tmp_path / "collector.state"
listeners = tmp_path / "listeners"
identities = tmp_path / "backup-identities.tsv"
lease_observed = tmp_path / "lease-observed.txt"
restore_contract_observed = tmp_path / "restore-contract.tsv"
flock_marker = tmp_path / "canary-lock-held.txt"
restore_hook = tmp_path / "clickhouse-native-restore-drill.sh"
inventory_helper = tmp_path / "clickhouse-restore-inventory.py"
backup_disk_config = tmp_path / "backup_disk.xml"
cluster_config = tmp_path / "isolated-cluster.xml"
cooldown_file = cooldown_dir / "signoz-otel-collector.cooldown"
run_id = f"canary-{tmp_path.name}"
_write_executable(
monitor_script,
_monitor_source(cooldown_dir, omit_contract=omit_monitor_contract),
)
_write_executable(backup_script, _backup_source())
if native_restore_contract == "symlink":
restore_hook_target = tmp_path / "restore-hook-target.sh"
_write_executable(restore_hook_target, _restore_hook_source())
restore_hook.symlink_to(restore_hook_target)
elif native_restore_contract == "valid":
_write_executable(restore_hook, _restore_hook_source())
elif native_restore_contract != "missing":
raise ValueError(
f"unsupported native restore contract: {native_restore_contract}"
)
_write_executable(inventory_helper, "#!/bin/sh\nexit 0\n")
backup_disk_config.write_text("<clickhouse/>\n", encoding="utf-8")
cluster_config.write_text("<clickhouse/>\n", encoding="utf-8")
_install_fake_commands(fake_bin)
monitor_log.write_text("[test] monitor seed\n", encoding="utf-8")
collector_state.write_text("true\n", encoding="utf-8")
listeners.write_text("4317\n4318\n", encoding="utf-8")
if prior_lease:
cooldown_file.write_text("123\n", encoding="utf-8")
cooldown_file.chmod(0o640)
env = {
**os.environ,
"PATH": f"{fake_bin}:{os.environ['PATH']}",
"TRACE_ID": "trace-P0-OBS-002",
"RUN_ID": run_id,
"WORK_ITEM_ID": "P0-OBS-002",
"SIGNOZ_CANARY_BACKUP_SCRIPT": str(backup_script),
"SIGNOZ_CANARY_MONITOR_SCRIPT": str(monitor_script),
"SIGNOZ_CANARY_MONITOR_LOG": str(monitor_log),
"SIGNOZ_CANARY_COOLDOWN_DIR": str(cooldown_dir),
"SIGNOZ_CANARY_RECEIPT_ROOT": str(receipt_root),
"SIGNOZ_CANARY_LOCK_FILE": str(tmp_path / "canary.lock"),
"SIGNOZ_CANARY_TIMEOUT_SECONDS": "5",
"SIGNOZ_CANARY_KILL_AFTER_SECONDS": "2",
"SIGNOZ_CANARY_DOCKER_TIMEOUT_SECONDS": "2",
"SIGNOZ_CANARY_LEASE_MARGIN_SECONDS": "2",
"SIGNOZ_CANARY_MONITOR_CRON_INTERVAL_SECONDS": str(monitor_interval_seconds),
"SIGNOZ_CANARY_STATE_POLL_SECONDS": "0.05",
"SIGNOZ_CANARY_EXPECT_COLLECTOR_STOP": ("1" if expect_collector_stop else "0"),
"CLICKHOUSE_NATIVE_POST_VERIFY_HOOK": str(restore_hook),
"CLICKHOUSE_RESTORE_INVENTORY_HELPER": str(inventory_helper),
"CLICKHOUSE_RESTORE_BACKUP_DISK_CONFIG": str(backup_disk_config),
"CLICKHOUSE_RESTORE_CLUSTER_CONFIG": str(cluster_config),
"CLICKHOUSE_NATIVE_RECEIPT_ROOT": str(native_receipt_root),
"CLICKHOUSE_RESTORE_RECEIPT_ROOT": str(restore_receipt_root),
"TEST_MONITOR_SCRIPT": str(monitor_script),
"TEST_MONITOR_LOG": str(monitor_log),
"TEST_COLLECTOR_STATE": str(collector_state),
"TEST_LISTENERS": str(listeners),
"TEST_BACKUP_IDENTITIES": str(identities),
"TEST_COOLDOWN_FILE": str(cooldown_file),
"TEST_LEASE_OBSERVED": str(lease_observed),
"TEST_RESTORE_CONTRACT": str(restore_contract_observed),
"TEST_RESTORE_HOOK": str(restore_hook),
"TEST_INVENTORY_HELPER": str(inventory_helper),
"TEST_BACKUP_DISK_CONFIG": str(backup_disk_config),
"TEST_CLUSTER_CONFIG": str(cluster_config),
"TEST_NATIVE_RECEIPT_ROOT": str(native_receipt_root),
"TEST_RESTORE_RECEIPT_ROOT": str(restore_receipt_root),
"TEST_FLOCK_MARKER": str(flock_marker),
"FAKE_MONITOR_MODE": monitor_mode,
"FAKE_BACKUP_SLEEP_SECONDS": str(backup_sleep_seconds),
"FAKE_DUPLICATE_CRON": "1" if duplicate_cron else "0",
"FAKE_DROP_LISTENER": "1" if drop_listener else "0",
"FAKE_BACKUP_STOPS_COLLECTOR": "1" if expect_collector_stop else "0",
"FAKE_RECEIPT_FAULT": native_receipt_state,
"FAKE_DOCKER_TIMEOUT": "1" if docker_timeout else "0",
}
started_at = int(time.time())
result = subprocess.run(
["bash", str(WRAPPER), f"--{mode}"],
env=env,
text=True,
capture_output=True,
timeout=15,
check=False,
)
paths = {
"cooldown": cooldown_file,
"receipts": receipt_root / run_id / "receipts.jsonl",
"rollback_metadata": receipt_root / run_id / "cooldown.rollback.json",
"identities": identities,
"lease_observed": lease_observed,
"collector_state": collector_state,
"listeners": listeners,
"restore_contract": restore_contract_observed,
"restore_hook": restore_hook,
"flock_marker": flock_marker,
"native_receipts": native_receipt_root / run_id / "receipts.jsonl",
"restore_receipts": restore_receipt_root / run_id / "receipts.jsonl",
"restore_status": restore_receipt_root / run_id / "status.json",
"backup_output": receipt_root / run_id / "backup-output.log",
}
return result, paths, str(started_at)
def _receipts(path: Path) -> list[dict[str, object]]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
def _terminal(receipts: list[dict[str, object]], stage: str) -> str:
return str([item for item in receipts if item["stage"] == stage][-1]["terminal"])
def test_existing_lease_is_atomically_replaced_and_exactly_restored(
tmp_path: Path,
) -> None:
result, paths, started_at = _run_canary(tmp_path)
assert result.returncode == 0, result.stdout + result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
assert stat.S_IMODE(paths["cooldown"].stat().st_mode) == 0o640
assert int(paths["lease_observed"].read_text(encoding="utf-8")) >= (
int(started_at) + 7
)
identities = paths["identities"].read_text(encoding="utf-8").split("\t")
assert identities[0] == "trace-P0-OBS-002"
assert identities[1].startswith("canary-")
assert identities[2] == "P0-OBS-002\n"
receipts = _receipts(paths["receipts"])
assert _terminal(receipts, "rollback") == "restored"
assert _terminal(receipts, "runtime_operation") == "pass"
assert _terminal(receipts, "closure_writeback") == "partial_degraded"
assert _terminal(receipts, "terminal") == "partial_degraded"
metadata = json.loads(paths["rollback_metadata"].read_text(encoding="utf-8"))
assert metadata["trace_id"] == "trace-P0-OBS-002"
assert metadata["prior_present"] == 1
stages = [str(item["stage"]) for item in receipts]
ordered = [
"sensor_source",
"normalized_asset_identity",
"source_of_truth_diff",
"ai_decision",
"risk_policy",
"check",
"execution",
"post_verifier",
"rollback",
"closure_writeback",
"terminal",
]
assert [stages.index(stage) for stage in ordered] == sorted(
stages.index(stage) for stage in ordered
)
def test_absent_lease_is_removed_after_success(tmp_path: Path) -> None:
result, paths, _ = _run_canary(tmp_path, prior_lease=False)
assert result.returncode == 0, result.stdout + result.stderr
assert not paths["cooldown"].exists()
receipts = _receipts(paths["receipts"])
assert _terminal(receipts, "rollback") == "restored"
assert _terminal(receipts, "terminal") == "partial_degraded"
def test_online_native_canary_keeps_collector_running_and_does_not_arm_lease(
tmp_path: Path,
) -> None:
result, paths, _ = _run_canary(tmp_path, expect_collector_stop=False)
assert result.returncode == 0, result.stdout + result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
assert not paths["lease_observed"].exists()
assert paths["collector_state"].read_text(encoding="utf-8") == "true\n"
assert paths["restore_contract"].read_text(encoding="utf-8").split("\t") == [
str(paths["restore_hook"]),
str(tmp_path / "clickhouse-restore-inventory.py"),
str(tmp_path / "backup_disk.xml"),
f"{tmp_path / 'isolated-cluster.xml'}\n",
]
receipts = _receipts(paths["receipts"])
assert paths["native_receipts"].is_file()
assert paths["restore_receipts"].is_file()
restore_status = json.loads(paths["restore_status"].read_text(encoding="utf-8"))
assert restore_status["terminal"] == "verified"
assert restore_status["check_table_count"] == 44
assert restore_status["check_table_pass_count"] == 44
assert any(
item["stage"] == "native_restore_receipt_verifier"
and item["terminal"] == "pass"
for item in receipts
)
assert any(
item["stage"] == "lease" and item["terminal"] == "not_required"
for item in receipts
)
assert _terminal(receipts, "rollback") == "no_write"
assert _terminal(receipts, "terminal") == "partial_degraded"
def test_online_native_canary_is_not_coupled_to_legacy_monitor_contract(
tmp_path: Path,
) -> None:
result, paths, _ = _run_canary(
tmp_path,
expect_collector_stop=False,
omit_monitor_contract=True,
duplicate_cron=True,
)
assert result.returncode == 0, result.stdout + result.stderr
assert not paths["lease_observed"].exists()
receipts = _receipts(paths["receipts"])
source_diff = [
item for item in receipts if item["stage"] == "source_of_truth_diff"
][-1]
assert "monitor_and_cooldown_contract_not_applicable" in source_diff["detail"]
assert _terminal(receipts, "terminal") == "partial_degraded"
def test_auto_repair_contamination_fails_and_restores_lease(tmp_path: Path) -> None:
result, paths, _ = _run_canary(tmp_path, monitor_mode="auto_repair")
assert result.returncode != 0
assert "monitor_auto_repair_contaminated_canary" in result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
receipts = _receipts(paths["receipts"])
assert _terminal(receipts, "rollback") == "restored"
assert _terminal(receipts, "terminal") == "failed"
def test_cron_crossing_requires_monitor_cooldown_receipt(tmp_path: Path) -> None:
result, paths, _ = _run_canary(
tmp_path,
monitor_mode="none",
backup_sleep_seconds=1.2,
monitor_interval_seconds=1,
)
assert result.returncode != 0
assert "cron_crossed_without_collector_cooldown_receipt" in result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
def test_cron_crossing_with_detected_and_cooldown_receipts_passes(
tmp_path: Path,
) -> None:
result, paths, _ = _run_canary(
tmp_path,
monitor_mode="cooldown",
backup_sleep_seconds=1.2,
monitor_interval_seconds=1,
)
assert result.returncode == 0, result.stdout + result.stderr
receipts = _receipts(paths["receipts"])
monitor_receipt = [
item for item in receipts if item["stage"] == "monitor_verifier"
][-1]
assert monitor_receipt["terminal"] == "pass"
assert "cron_crossed_1" in str(monitor_receipt["detail"])
def test_missing_monitor_contract_fails_before_lease_or_backup(tmp_path: Path) -> None:
result, paths, _ = _run_canary(tmp_path, omit_monitor_contract=True)
assert result.returncode != 0
assert "monitor_contract_ambiguous_fragment_count_0" in result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
assert not paths["lease_observed"].exists()
receipts = _receipts(paths["receipts"])
assert _terminal(receipts, "rollback") == "not_armed"
assert _terminal(receipts, "terminal") == "failed"
def test_ambiguous_cron_contract_fails_before_lease_or_backup(
tmp_path: Path,
) -> None:
result, paths, _ = _run_canary(tmp_path, duplicate_cron=True)
assert result.returncode != 0
assert "monitor_cron_contract_ambiguous_count_2" in result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
assert not paths["lease_observed"].exists()
def test_post_backup_listener_failure_fails_and_restores_lease(
tmp_path: Path,
) -> None:
result, paths, _ = _run_canary(tmp_path, drop_listener=True)
assert result.returncode != 0
assert "collector_post_backup_runtime_failed" in result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
receipts = _receipts(paths["receipts"])
assert _terminal(receipts, "rollback") == "restored"
assert _terminal(receipts, "terminal") == "failed"
def test_wrapper_deployment_and_static_contracts_are_owned() -> None:
wrapper = WRAPPER.read_text(encoding="utf-8")
assert "BACKUP_SKIP_RETENTION_CLEANUP=1" in wrapper
assert 'TRACE_ID="${TRACE_ID:-}"' in wrapper
assert 'RUN_ID="${RUN_ID:-}"' in wrapper
assert 'WORK_ITEM_ID="${WORK_ITEM_ID:-}"' in wrapper
assert "monitor_auto_repair_contaminated_canary" in wrapper
assert "collector_post_backup_runtime_failed" in wrapper
assert "/backup/scripts/clickhouse-native-restore-drill.sh" in wrapper
assert "/backup/scripts/clickhouse-restore-inventory.py" in wrapper
assert "/backup/config/signoz/clickhouse/config.d/backup_disk.xml" in wrapper
assert (
"/backup/config/signoz/clickhouse/restore-drill/config.d/isolated-cluster.xml"
in wrapper
)
assert (
'CLICKHOUSE_NATIVE_POST_VERIFY_HOOK="${CLICKHOUSE_NATIVE_POST_VERIFY_HOOK}"'
in wrapper
)
assert (
'CLICKHOUSE_NATIVE_RECEIPT_ROOT="${CLICKHOUSE_NATIVE_RECEIPT_ROOT}"' in wrapper
)
assert (
'CLICKHOUSE_RESTORE_RECEIPT_ROOT="${CLICKHOUSE_RESTORE_RECEIPT_ROOT}"'
in wrapper
)
main_source = wrapper[wrapper.index("main() {") :]
assert main_source.index("acquire_canary_lock") < main_source.index(
"require_native_restore_contract"
)
assert "same_run_native_restore_receipt_validation_failed" in wrapper
assert '"check_table_count"' in wrapper
assert "mandatory_isolated_restore_verifier" in wrapper
assert "SIGNOZ_CANARY_DOCKER_TIMEOUT_SECONDS" in wrapper
assert 'emit_receipt "runtime_operation" "pass"' in wrapper
assert 'emit_receipt "closure_writeback" "partial_degraded"' in wrapper
assert WRAPPER.name in PLAYBOOK.read_text(encoding="utf-8")
assert str(WRAPPER.relative_to(ROOT)) in ANSIBLE_VALIDATE.read_text(
encoding="utf-8"
)
assert str(WRAPPER.relative_to(ROOT)) in READINESS_AUDIT.read_text(encoding="utf-8")
def test_check_mode_proves_contract_without_lease_or_backup_write(
tmp_path: Path,
) -> None:
result, paths, _ = _run_canary(tmp_path, mode="check")
assert result.returncode == 0, result.stdout + result.stderr
assert paths["cooldown"].read_text(encoding="utf-8") == "123\n"
assert stat.S_IMODE(paths["cooldown"].stat().st_mode) == 0o640
assert not paths["identities"].exists()
assert not paths["lease_observed"].exists()
check_receipts = _receipts(paths["receipts"])
assert _terminal(check_receipts, "rollback") == "no_write"
assert _terminal(check_receipts, "terminal") == "check_pass_no_write"
@pytest.mark.parametrize("contract_state", ["missing", "symlink"])
def test_native_check_fails_closed_for_unsafe_restore_contract(
tmp_path: Path,
contract_state: str,
) -> None:
result, paths, _ = _run_canary(
tmp_path,
mode="check",
expect_collector_stop=False,
native_restore_contract=contract_state,
)
assert result.returncode != 0
assert "native_restore_contract_CLICKHOUSE_NATIVE_POST_VERIFY_HOOK" in result.stderr
assert paths["flock_marker"].read_text(encoding="utf-8") == "held\n"
assert not paths["identities"].exists()
receipts = _receipts(paths["receipts"])
assert _terminal(receipts, "terminal") == "failed"
@pytest.mark.parametrize(
"receipt_state",
[
"no_hook",
"forged_identity",
"missing_status",
"symlink_status",
"malformed_restore_receipt",
],
)
def test_native_apply_rejects_missing_or_forged_machine_receipts(
tmp_path: Path,
receipt_state: str,
) -> None:
result, paths, _ = _run_canary(
tmp_path,
expect_collector_stop=False,
native_receipt_state=receipt_state,
)
assert result.returncode != 0
assert "========== SigNoz ClickHouse native 備份完成 (1s) ==========" in paths[
"backup_output"
].read_text(encoding="utf-8")
assert (
"same_run_native_restore_receipt_validation_failed" in result.stderr
or "same_run_native_restore_receipt_missing_unreadable_or_symlink"
in result.stderr
)
assert paths["flock_marker"].read_text(encoding="utf-8") == "held\n"
receipts = _receipts(paths["receipts"])
assert not any(
item["stage"] == "native_restore_receipt_verifier"
and item["terminal"] == "pass"
for item in receipts
)
assert _terminal(receipts, "terminal") == "failed"
def test_docker_metadata_timeout_fails_before_backup(tmp_path: Path) -> None:
result, paths, _ = _run_canary(
tmp_path,
mode="check",
expect_collector_stop=False,
docker_timeout=True,
)
assert result.returncode != 0
assert "collector_runtime_metadata_unreadable" in result.stderr
assert not paths["identities"].exists()
receipts = _receipts(paths["receipts"])
assert _terminal(receipts, "terminal") == "failed"
def test_wrapper_is_valid_bash() -> None:
result = subprocess.run(
["bash", "-n", str(WRAPPER)],
text=True,
capture_output=True,
check=False,
)
assert result.returncode == 0, result.stderr
Reference in New Issue View Git Blame Copy Permalink