#!/usr/bin/env python3 """Run the complete-marker and version-path regression in starter staging.""" from __future__ import annotations import hashlib import json import os import random import re import shutil import subprocess import sys import time from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple import pymysql import yaml SCRIPT_DIR = Path(__file__).resolve().parent REPO_ROOT = SCRIPT_DIR.parent if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from integration_test import generate_active_cluster_ids_sql as sql_generator from integration_test import mysql_control CONTROLLER_SERVICE_ACCOUNT = "backup-restore-manus" WORKER_SERVICE_ACCOUNT = "backup-restore" AWS_REGION = "us-east-1" GPG_BUCKET = "juncen-test" GPG_KEY = "pubkey.asc" EXPORT_BUCKET = "staging-us-east-1-f01-export" EXPORT_KEY_PREFIX = ( "system:serviceaccount:tidb-serverless:backup-restore/exported/" ) OUT_PREFIX = f"s3://{EXPORT_BUCKET}/{EXPORT_KEY_PREFIX}" EXPORT_RUNNER_IMAGE = "public.ecr.aws/lambda/python:3.13" PACK_CSE_CTL_IMAGE = ( "us-docker.pkg.dev/pingcap-testing-account/dev/tikv/tikv/image:" "pull-5645-41af0d7-nextgen" ) EXPORT_CSE_CTL_IMAGE = ( "us-docker.pkg.dev/pingcap-testing-account/dev/tikv/tikv/image:" "juncen-exp-export-packed-csv-aws-profile-proof-21e88dd-nextgen" ) DUMPLING_IMAGE = ( "us-docker.pkg.dev/pingcap-testing-account/dev/pingcap/tidb/images/" "dumpling:exp-export-packed-aws-profile-proof-3f1742a" ) TEST_JOB_TTL_SECONDS_AFTER_FINISHED = 3600 BACKUP_NAME_FORMAT = "%Y%m%d/%H%M%S.meta" BACKUP_DATE_FORMAT = "%Y%m%d" UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%SZ" BACKUP_NAME_PATTERN = re.compile(r"^[0-9]{8}/[0-9]{6}[.]meta$") DNS_LABEL_PATTERN = re.compile(r"^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$") PERMISSION_ERROR_PATTERN = re.compile( r"Forbidden|Unauthorized|AccessDenied|Access Denied" ) AUTOSCALER_SAFE_TO_EVICT_ANNOTATION = ( "cluster-autoscaler.kubernetes.io/safe-to-evict" ) class RegressionError(RuntimeError): pass class PermissionDenied(RegressionError): pass @dataclass(frozen=True) class Settings: kubernetes_context: str namespace: str cron_job_name: str node_group: str backup_prefix: str minimum_target_clusters: int pack_shards: int export_parallel: int watermark_parallel: int dumpling_threads: int dumpling_timeout_seconds: int aws_profile: str controller_image: str mysql_defaults_file: Path lark_webhook: str = field(repr=False) run_token: str = "" state_dir: Path = Path() controller_wait_seconds: int = 14400 @property def kubectl(self) -> List[str]: return [ "kubectl", "--context", self.kubernetes_context, "-n", self.namespace, ] @property def mysql_secret(self) -> str: return f"{self.cron_job_name}-mysql" @property def lark_secret(self) -> str: return f"{self.cron_job_name}-lark" @property def gpg_config_map(self) -> str: return f"{self.cron_job_name}-gpg-public-key" @property def workflow_config_map(self) -> str: return f"{self.cron_job_name}-workflow-spec" def positive_environment_integer(name: str, default: int) -> int: value = os.environ.get(name, str(default)) if re.fullmatch(r"[1-9][0-9]*", value) is None: raise RegressionError(f"{name} must be a positive integer") return int(value) def load_settings() -> Settings: run_token = os.environ.get( "RUN_TOKEN", datetime.now(timezone.utc).strftime("%d%H%M%S"), ) state_dir_text = os.environ.get("STATE_DIR") state_dir = ( Path(state_dir_text).expanduser() if state_dir_text else SCRIPT_DIR / "tmp" / f"complete-marker-{run_token}" ) defaults_text = os.environ.get("MYSQL_DEFAULTS_FILE") defaults_file = ( Path(defaults_text).expanduser() if defaults_text else Path("~/.my.cnf").expanduser() ) settings = Settings( kubernetes_context=os.environ.get("K8S_CONTEXT", "starter-staging"), namespace=os.environ.get("K8S_NAMESPACE", "tidb-serverless"), cron_job_name=os.environ.get( "CRONJOB_NAME", "on-demand-packed-export" ), node_group=os.environ.get("NODE_GROUP", "dm-worker"), backup_prefix=os.environ.get( "BACKUP_PREFIX", datetime.now(timezone.utc).strftime("%Y%m%d"), ), minimum_target_clusters=positive_environment_integer( "MIN_TARGET_CLUSTERS", 2 ), pack_shards=positive_environment_integer("PACK_SHARDS", 1), export_parallel=positive_environment_integer("EXPORT_PARALLEL", 1), watermark_parallel=positive_environment_integer( "WATERMARK_PARALLEL", 4 ), dumpling_threads=positive_environment_integer("DUMPLING_THREADS", 4), dumpling_timeout_seconds=positive_environment_integer( "DUMPLING_TIMEOUT_SECONDS", 300 ), aws_profile=os.environ.get("AWS_PROFILE_NAME", "pingcap"), controller_image=os.environ.get("CONTROLLER_IMAGE", ""), mysql_defaults_file=defaults_file, lark_webhook=os.environ.get("LARK_BOT_WEBHOOK", ""), run_token=run_token, state_dir=state_dir, controller_wait_seconds=positive_environment_integer( "CONTROLLER_WAIT_SECONDS", 14400 ), ) validate_settings(settings) return settings def validate_settings(settings: Settings) -> None: for value in ( settings.namespace, settings.cron_job_name, settings.node_group, ): if DNS_LABEL_PATTERN.fullmatch(value) is None or len(value) > 63: raise RegressionError(f"invalid Kubernetes DNS label: {value}") if re.fullmatch(r"[a-z0-9]{4,9}", settings.run_token) is None: raise RegressionError( "RUN_TOKEN must contain 4-9 lowercase letters or digits" ) if settings.minimum_target_clusters < 2: raise RegressionError("MIN_TARGET_CLUSTERS must be at least 2") try: parsed_backup_date = datetime.strptime( settings.backup_prefix, BACKUP_DATE_FORMAT, ) except ValueError as exc: raise RegressionError( f"BACKUP_PREFIX must use {BACKUP_DATE_FORMAT}" ) from exc if parsed_backup_date.strftime(BACKUP_DATE_FORMAT) != settings.backup_prefix: raise RegressionError(f"BACKUP_PREFIX must use {BACKUP_DATE_FORMAT}") if not settings.controller_image or any( character.isspace() for character in settings.controller_image ): raise RegressionError( "CONTROLLER_IMAGE must identify the revision under test" ) if settings.namespace != "tidb-serverless": raise RegressionError( "the workflow generators require K8S_NAMESPACE=tidb-serverless" ) if not settings.mysql_defaults_file.is_file(): raise RegressionError( "MySQL defaults file is not a regular file: " f"{settings.mysql_defaults_file}" ) if not settings.lark_webhook: raise RegressionError("LARK_BOT_WEBHOOK must be exported") if OUT_PREFIX != ( "s3://staging-us-east-1-f01-export/" "system:serviceaccount:tidb-serverless:backup-restore/exported/" ): raise RegressionError("the fixed export path was changed") def log(message: str) -> None: print(f"[e2e] {message}", flush=True) def decode_output(payload: Optional[bytes]) -> str: return (payload or b"").decode("utf-8", errors="replace") def execute( command: Sequence[str], *, input_bytes: Optional[bytes] = None, check: bool = True, echo: bool = False, ) -> subprocess.CompletedProcess[bytes]: completed = subprocess.run( list(command), input=input_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) stdout = decode_output(completed.stdout) stderr = decode_output(completed.stderr) if echo: if stdout: print(stdout, end="" if stdout.endswith("\n") else "\n") if stderr: print( stderr, end="" if stderr.endswith("\n") else "\n", file=sys.stderr, ) if completed.returncode != 0 and check: if PERMISSION_ERROR_PATTERN.search(stderr): raise PermissionDenied(stderr.strip() or "permission denied") if stderr and not echo: print(stderr, end="" if stderr.endswith("\n") else "\n", file=sys.stderr) raise RegressionError( f"command failed with exit {completed.returncode}: " + " ".join(command) ) return completed def parse_json_bytes(payload: bytes, description: str) -> Any: try: return json.loads(payload) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise RegressionError(f"{description} was not valid JSON") from exc def write_private_bytes(path: Path, payload: bytes) -> None: path.write_bytes(payload) path.chmod(0o600) def write_private_text(path: Path, payload: str) -> None: path.write_text(payload, encoding="utf-8") path.chmod(0o600) def write_json(path: Path, payload: Any, *, pretty: bool = False) -> None: if pretty: content = json.dumps(payload, indent=2, sort_keys=True) + "\n" else: content = json.dumps(payload, separators=(",", ":")) + "\n" write_private_text(path, content) def require(condition: bool, message: str) -> None: if not condition: raise RegressionError(message) def split_identity(identity: str) -> Tuple[str, str]: cluster_id, separator, keyspace_name = identity.rpartition("_") if not separator or not cluster_id or not keyspace_name: raise RegressionError( "packed identity does not match " f"_: {identity!r}" ) return cluster_id, keyspace_name def derive_run_id(controller_job_name: str, cron_job_name: str) -> str: prefix = f"{cron_job_name}-" require( controller_job_name.startswith(prefix), "Controller Job name does not start with the CronJob name", ) execution_id = controller_job_name[len(prefix) :] require( DNS_LABEL_PATTERN.fullmatch(execution_id) is not None, f"invalid Controller execution ID: {execution_id}", ) digest = hashlib.sha256(controller_job_name.encode("utf-8")).hexdigest()[:8] return f"cron-{execution_id}-{digest}" def backup_datetime(backup_name: str) -> datetime: try: return datetime.strptime(backup_name, BACKUP_NAME_FORMAT) except ValueError as exc: raise RegressionError( f"backup name must use {BACKUP_NAME_FORMAT}: {backup_name!r}" ) from exc def backup_timestamp(backup_name: str) -> str: return backup_datetime(backup_name).strftime(UTC_TIMESTAMP_FORMAT) def workflow_config(settings: Settings) -> Dict[str, Any]: return { "schemaVersion": 1, "cronJob": { "name": settings.cron_job_name, "namespace": settings.namespace, "schedule": "0 1 * * *", "timeZone": "Etc/UTC", "startingDeadlineSeconds": 300, "ttlSecondsAfterFinished": TEST_JOB_TTL_SECONDS_AFTER_FINISHED, "successfulJobsHistoryLimit": 3, "failedJobsHistoryLimit": 5, "suspend": True, "kubernetesContextName": settings.kubernetes_context, "serviceAccountName": CONTROLLER_SERVICE_ACCOUNT, "controllerImage": settings.controller_image, "imagePullPolicy": "Always", "nodeGroup": settings.node_group, "resources": { "requests": {"cpu": "100m", "memory": "256Mi"}, "limits": {"cpu": "1", "memory": "1Gi"}, }, }, "workflow": { "schemaVersion": 1, "pack": { "shards": settings.pack_shards, "cseCtlImage": PACK_CSE_CTL_IMAGE, "resources": {"cpuRequest": "3", "memoryRequest": "14G"}, "generatorArgs": [], "commandArgs": [], }, "export": { "outPrefix": OUT_PREFIX, "cseCtlImage": EXPORT_CSE_CTL_IMAGE, "dumplingImage": DUMPLING_IMAGE, "runnerImage": EXPORT_RUNNER_IMAGE, "resources": {"cpuRequest": "3", "memoryRequest": "14G"}, "generatorArgs": [], "workerArgs": [ "--parallel", str(settings.export_parallel), "--dumpling-timeout", str(settings.dumpling_timeout_seconds), ], "dumplingArgs": [ "--filesize", "1GiB", "--threads", str(settings.dumpling_threads), ], }, "watermark": { "image": settings.controller_image, "awsRegion": AWS_REGION, "resources": {"cpuRequest": "500m", "memoryRequest": "1Gi"}, "generatorArgs": [], "workerArgs": [ "--parallel", str(settings.watermark_parallel), ], }, }, } class Regression: def __init__(self, settings: Settings) -> None: self.settings = settings self.backup_bucket = "" self.backup_dfs_prefix = "" self.backup_region = "" self.cleanup_targets: List[Tuple[str, str]] = [] self.terminal_run_ids: set[str] = set() self.last_controller_condition = "" self.last_controller_result: Dict[str, Any] = {} def kubectl_raw( self, *arguments: str, input_bytes: Optional[bytes] = None, check: bool = True, echo: bool = False, ) -> subprocess.CompletedProcess[bytes]: return execute( [*self.settings.kubectl, *arguments], input_bytes=input_bytes, check=check, echo=echo, ) def kubectl_text(self, *arguments: str) -> str: return decode_output(self.kubectl_raw(*arguments).stdout) def kubectl_json(self, *arguments: str) -> Dict[str, Any]: payload = parse_json_bytes( self.kubectl_raw(*arguments).stdout, f"kubectl {' '.join(arguments)}", ) if not isinstance(payload, dict): raise RegressionError("kubectl JSON result was not an object") return payload def aws_raw( self, *arguments: str, region: str = AWS_REGION, check: bool = True, ) -> subprocess.CompletedProcess[bytes]: return execute( [ "aws", "--profile", self.settings.aws_profile, "--region", region, *arguments, ], check=check, ) def aws_json(self, *arguments: str, region: str = AWS_REGION) -> Any: return parse_json_bytes( self.aws_raw(*arguments, region=region).stdout, f"aws {' '.join(arguments)}", ) def preflight(self) -> None: for command in ("aws", "kubectl"): if shutil.which(command) is None: raise RegressionError(f"required command is missing: {command}") if self.settings.state_dir.exists(): raise RegressionError( f"state directory already exists: {self.settings.state_dir}" ) self.settings.state_dir.mkdir(parents=True, mode=0o700) self.settings.state_dir.chmod(0o700) contexts = decode_output( execute( [ "kubectl", "config", "get-contexts", self.settings.kubernetes_context, "-o", "name", ] ).stdout ).splitlines() require( self.settings.kubernetes_context in contexts, "kubectl context does not exist: " f"{self.settings.kubernetes_context}", ) for service_account in ( CONTROLLER_SERVICE_ACCOUNT, WORKER_SERVICE_ACCOUNT, ): self.kubectl_raw( "get", f"serviceaccount/{service_account}", "-o", "name", echo=True, ) stateful_set = self.kubectl_json( "get", "statefulset/tikv-api", "-o", "json" ) pod = self.kubectl_json("get", "pod/tikv-api-0", "-o", "json") log( "tikv-api readiness: " f"statefulset={stateful_set.get('metadata', {}).get('name')} " f"ready={stateful_set.get('status', {}).get('readyReplicas', 0)} " f"pod={pod.get('metadata', {}).get('name')} " f"phase={pod.get('status', {}).get('phase')}" ) self.load_backup_storage(stateful_set) jobs = self.kubectl_json( "get", "jobs", "-l", f"app.kubernetes.io/instance={self.settings.cron_job_name}", "-o", "json", ) active_jobs = 0 for job in jobs.get("items", []): conditions = (job.get("status") or {}).get("conditions") or [] terminal = any( condition.get("status") == "True" and condition.get("type") in {"Complete", "Failed"} for condition in conditions ) if not terminal: active_jobs += 1 require( active_jobs == 0, f"{active_jobs} active Controller Job(s) already use " f"{self.settings.cron_job_name}", ) gpg_head = self.aws_json( "s3api", "head-object", "--bucket", GPG_BUCKET, "--key", GPG_KEY, ) write_json(self.settings.state_dir / "gpg-head.json", gpg_head, pretty=True) log(f"preflight passed; state directory: {self.settings.state_dir}") def load_backup_storage(self, stateful_set: Mapping[str, Any]) -> None: containers = ( stateful_set.get("spec", {}) .get("template", {}) .get("spec", {}) .get("containers", []) ) def literal(name: str) -> str: values = [ item.get("value") for container in containers if container.get("name") == "tikv-api" for item in container.get("env", []) if item.get("name") == name and isinstance(item.get("value"), str) ] if len(values) != 1: raise RegressionError(f"expected exactly one literal {name}") return values[0] self.backup_bucket = literal("DFS_S3_BUCKET") self.backup_dfs_prefix = literal("DFS_PREFIX") self.backup_region = literal("DFS_S3_REGION") require(bool(self.backup_bucket), "DFS_S3_BUCKET is empty") require(bool(self.backup_region), "DFS_S3_REGION is empty") def flush_wal(self) -> None: self.kubectl_raw( "exec", "pod/tikv-api-0", "-c", "tikv-api", "--", "curl", "--silent", "--show-error", "--fail", "--connect-timeout", "10", "--max-time", "60", "-k", "-X", "POST", "https://localhost:19500/api/v1/x/cluster/flush_wal", ) log("tikv-api WAL flush completed") def discover_first_backup_name(self, prefix: str) -> str: response = self.kubectl_raw( "exec", "pod/tikv-api-0", "-c", "tikv-api", "--", "curl", "--silent", "--show-error", "--fail", "--connect-timeout", "10", "--max-time", "60", "-k", "--get", "--data-urlencode", f"from_prefix={prefix}", "https://localhost:19500/api/v1/x/backups", ) payload = parse_json_bytes(response.stdout, "tikv-api backup list") data = payload.get("data") if isinstance(payload, dict) else None if not isinstance(data, list): raise RegressionError("tikv-api backup list has no data array") candidates = [] for item in data: if not isinstance(item, dict): continue name = item.get("name") if ( isinstance(name, str) and name.startswith(prefix) and BACKUP_NAME_PATTERN.fullmatch(name) is not None ): candidates.append(name) candidates.sort() if not candidates: raise RegressionError(f"no valid backup matched {prefix}") return candidates[0] def backup_object_key(self, backup_name: str) -> str: normalized_prefix = self.backup_dfs_prefix.strip("/") if normalized_prefix: return f"{normalized_prefix}/backup/{backup_name}" return f"backup/{backup_name}" def download_backup_meta( self, backup_name: str, label: str, ) -> Tuple[bytes, str]: object_key = self.backup_object_key(backup_name) head = self.aws_json( "s3api", "head-object", "--bucket", self.backup_bucket, "--key", object_key, region=self.backup_region, ) write_json( self.settings.state_dir / f"{label}-backup-head.json", head, pretty=True, ) uri = f"s3://{self.backup_bucket}/{object_key}" backup_meta = self.aws_raw( "s3", "cp", uri, "-", "--only-show-errors", region=self.backup_region, ).stdout require(bool(backup_meta), f"downloaded backup metadata is empty: {uri}") write_private_bytes( self.settings.state_dir / f"{label}-backup.meta", backup_meta, ) log(f"downloaded pinned {label} backup metadata: {uri}") return backup_meta, uri def generate_seed_sql( self, earlier_backup_meta: bytes, earlier_source_uri: str, later_backup_meta: bytes, later_source_uri: str, cutoff: datetime, ) -> Tuple[str, List[str]]: ( earlier_cluster_ids, earlier_duplicate_count, earlier_keyspace_count, earlier_non_enabled_keyspace_count, earlier_backup_cluster_id, ) = sql_generator.load_cluster_ids( earlier_backup_meta, earlier_source_uri, ) ( later_cluster_ids, later_duplicate_count, later_keyspace_count, later_non_enabled_keyspace_count, later_backup_cluster_id, ) = sql_generator.load_cluster_ids( later_backup_meta, later_source_uri, ) cluster_ids = sorted( set(earlier_cluster_ids) & set(later_cluster_ids) ) require( len(cluster_ids) >= self.settings.minimum_target_clusters, "adjacent backups have too few common ENABLED clusters: " f"earlier={len(earlier_cluster_ids)} " f"later={len(later_cluster_ids)} common={len(cluster_ids)}", ) sql = sql_generator.build_sql( cluster_ids, cutoff - timedelta(days=30), cutoff - timedelta(days=7), random.Random(0), earlier_source_uri, earlier_backup_cluster_id, earlier_keyspace_count, earlier_non_enabled_keyspace_count, ) sql = ( f"-- Intersected with {later_source_uri}; " f"common ENABLED clusters: {len(cluster_ids)}.\n" + sql ) write_private_text( self.settings.state_dir / "active_cluster_ids.sql", sql, ) write_json( self.settings.state_dir / "backup-cluster-intersection.json", { "earlier": { "source": earlier_source_uri, "backup_cluster_id": earlier_backup_cluster_id, "clusters": earlier_cluster_ids, "duplicate_cluster_ids": earlier_duplicate_count, "keyspaces": earlier_keyspace_count, "non_enabled_keyspaces": ( earlier_non_enabled_keyspace_count ), }, "later": { "source": later_source_uri, "backup_cluster_id": later_backup_cluster_id, "clusters": later_cluster_ids, "duplicate_cluster_ids": later_duplicate_count, "keyspaces": later_keyspace_count, "non_enabled_keyspaces": later_non_enabled_keyspace_count, }, "common_clusters": cluster_ids, }, pretty=True, ) log( "backup fixture intersection: " f"earlier_clusters={len(earlier_cluster_ids)} " f"later_clusters={len(later_cluster_ids)} " f"common_clusters={len(cluster_ids)}" ) return sql, cluster_ids def apply_generated_manifest( self, config_yaml: bytes, gpg_path: Path, ) -> None: generator_environment = dict(os.environ) generator_environment["LARK_BOT_WEBHOOK"] = self.settings.lark_webhook generator = subprocess.Popen( [ str(REPO_ROOT / "generate_packed_backup_cronjob.py"), "--config", "/dev/stdin", "--gpg-public-key", str(gpg_path), "--mysql-defaults-file", str(self.settings.mysql_defaults_file), "--lark-webhook-env", "LARK_BOT_WEBHOOK", ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=generator_environment, ) if generator.stdin is None or generator.stdout is None: generator.kill() generator.wait() raise RegressionError("cannot open CronJob generator pipes") if generator.stderr is None: generator.kill() generator.wait() raise RegressionError("cannot read CronJob generator errors") try: kubectl = subprocess.Popen( [*self.settings.kubectl, "apply", "-f", "-"], stdin=generator.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) except BaseException: generator.kill() generator.communicate() raise generator.stdout.close() try: generator.stdin.write(config_yaml) generator.stdin.close() kubectl_stdout, kubectl_stderr = kubectl.communicate() generator_stderr = generator.stderr.read() generator_return_code = generator.wait() except BaseException: generator.kill() kubectl.kill() generator.wait() kubectl.wait() raise kubectl_output_text = decode_output(kubectl_stdout) kubectl_error_text = decode_output(kubectl_stderr) if kubectl_output_text: print( kubectl_output_text, end="" if kubectl_output_text.endswith("\n") else "\n", ) if kubectl_error_text: print( kubectl_error_text, end="" if kubectl_error_text.endswith("\n") else "\n", file=sys.stderr, ) if PERMISSION_ERROR_PATTERN.search(kubectl_error_text): raise PermissionDenied( kubectl_error_text.strip() or "permission denied" ) if generator_return_code != 0: generator_error_text = decode_output(generator_stderr) if generator_error_text: print(generator_error_text, file=sys.stderr) raise RegressionError( "CronJob generator failed with exit " f"{generator_return_code}" ) if kubectl.returncode != 0: raise RegressionError( f"kubectl apply failed with exit {kubectl.returncode}" ) def deploy_cron_job(self) -> None: gpg_path = self.settings.state_dir / "pubkey.asc" gpg_payload = self.aws_raw( "s3", "cp", f"s3://{GPG_BUCKET}/{GPG_KEY}", "-", "--only-show-errors", ).stdout require(bool(gpg_payload), "downloaded GPG public key is empty") write_private_bytes(gpg_path, gpg_payload) config_yaml = yaml.safe_dump( workflow_config(self.settings), sort_keys=False, allow_unicode=True, ).encode("utf-8") self.apply_generated_manifest(config_yaml, gpg_path) cron_job = self.kubectl_json( "get", f"cronjob/{self.settings.cron_job_name}", "-o", "json", ) spec = cron_job.get("spec") or {} job_spec = ( spec.get("jobTemplate", {}) .get("spec", {}) .get("template", {}) .get("spec", {}) ) orchestrators = [ container for container in job_spec.get("containers", []) if container.get("name") == "orchestrator" ] require(len(orchestrators) == 1, "deployed CronJob has no orchestrator") require(spec.get("suspend") is True, "deployed CronJob is not suspended") require(spec.get("schedule") == "0 1 * * *", "CronJob schedule differs") require( spec.get("concurrencyPolicy") == "Forbid", "CronJob concurrency policy differs", ) require( job_spec.get("serviceAccountName") == CONTROLLER_SERVICE_ACCOUNT, "CronJob Controller service account differs", ) require( orchestrators[0].get("image") == self.settings.controller_image, "CronJob Controller image differs", ) def workflow_inventory(self, run_id: str) -> Mapping[str, Any]: return self.kubectl_json( "get", "jobs,pods,secrets,configmaps,persistentvolumeclaims", "-l", f"cse.tidbcloud.com/workflow-name={run_id}", "-o", "json", ) def controller_inventory(self, run_id: str) -> Mapping[str, Any]: return self.kubectl_json( "get", "jobs,pods", "-l", f"cse.tidbcloud.com/e2e-workflow={run_id}", "-o", "json", ) def assert_cleanup_scope_empty( self, controller_job: str, run_id: str, ) -> None: workflow_items = self.workflow_inventory(run_id).get("items", []) controller_items = self.controller_inventory(run_id).get("items", []) require( not workflow_items, f"refusing to reuse non-empty workflow cleanup scope {run_id}", ) require( not controller_items, f"refusing to reuse non-empty Controller cleanup scope {run_id}", ) require( derive_run_id(controller_job, self.settings.cron_job_name) == run_id, "Controller Job and RUN_ID do not match", ) def create_controller_job( self, execution_id: str, backup_selector: str, ) -> Tuple[str, str]: job_name = f"{self.settings.cron_job_name}-{execution_id}" run_id = derive_run_id(job_name, self.settings.cron_job_name) existing = self.kubectl_text( "get", f"job/{job_name}", "--ignore-not-found=true", "-o", "name", ).strip() require(not existing, f"Controller Job already exists: {job_name}") self.assert_cleanup_scope_empty(job_name, run_id) self.cleanup_targets.append((job_name, run_id)) job = self.kubectl_json( "create", "job", job_name, f"--from=cronjob/{self.settings.cron_job_name}", "--dry-run=client", "-o", "json", ) template = job.setdefault("spec", {}).setdefault("template", {}) pod_spec = template.setdefault("spec", {}) orchestrators = [ container for container in pod_spec.get("containers", []) if container.get("name") == "orchestrator" ] require( len(orchestrators) == 1, "generated Controller Job has no orchestrator", ) orchestrator_args = orchestrators[0].setdefault("args", []) require( isinstance(orchestrator_args, list), "generated Controller Job has invalid orchestrator args", ) orchestrator_args.extend(["--backup-from-prefix", backup_selector]) metadata = job.setdefault("metadata", {}) metadata.setdefault("labels", {})[ "cse.tidbcloud.com/e2e-workflow" ] = run_id template_metadata = template.setdefault("metadata", {}) template_metadata.setdefault("labels", {})[ "cse.tidbcloud.com/e2e-workflow" ] = run_id template_metadata.setdefault("annotations", {})[ AUTOSCALER_SAFE_TO_EVICT_ANNOTATION ] = "false" self.kubectl_raw( "apply", "-f", "-", input_bytes=json.dumps(job).encode("utf-8"), echo=True, ) applied = self.kubectl_json( "get", f"job/{job_name}", "-o", "json" ) labels = applied.get("metadata", {}).get("labels", {}) pod_labels = ( applied.get("spec", {}) .get("template", {}) .get("metadata", {}) .get("labels", {}) ) require( labels.get("app.kubernetes.io/instance") == self.settings.cron_job_name, "Controller Job instance label differs", ) require( labels.get("cse.tidbcloud.com/e2e-workflow") == run_id, "Controller Job cleanup label differs", ) require( pod_labels.get("cse.tidbcloud.com/e2e-workflow") == run_id, "Controller Pod cleanup label differs", ) return job_name, run_id def kubectl_with_retries( self, arguments: Sequence[str], description: str, ) -> bytes: last_error = b"" for attempt in range(1, 6): completed = self.kubectl_raw(*arguments, check=False) if completed.returncode == 0: return completed.stdout last_error = completed.stderr error_text = decode_output(last_error) if PERMISSION_ERROR_PATTERN.search(error_text): raise PermissionDenied(error_text.strip() or "permission denied") log(f"transient kubectl failure for {description}; retry {attempt}/5") time.sleep(attempt) if last_error: print(decode_output(last_error), file=sys.stderr) raise RegressionError(f"cannot read {description} after 5 attempts") def wait_controller_job(self, job_name: str) -> Mapping[str, Any]: started_at = time.monotonic() condition = "Running" while True: job_payload = parse_json_bytes( self.kubectl_with_retries( ["get", f"job/{job_name}", "-o", "json"], f"Controller Job {job_name}", ), f"Controller Job {job_name}", ) conditions = ( job_payload.get("status", {}).get("conditions", []) if isinstance(job_payload, dict) else [] ) condition = next( ( item.get("type") for item in conditions if item.get("status") == "True" and item.get("type") in {"Complete", "Failed"} ), "Running", ) if condition != "Running": self.terminal_run_ids.add( derive_run_id(job_name, self.settings.cron_job_name) ) break elapsed = int(time.monotonic() - started_at) require( elapsed < self.settings.controller_wait_seconds, f"timed out waiting for Controller Job {job_name}", ) if elapsed % 30 < 10: log(f"waiting for {job_name}: {elapsed}s") time.sleep(10) log_payload = self.kubectl_with_retries( ["logs", f"job/{job_name}", "-c", "orchestrator"], f"Controller log {job_name}", ) log_path = self.settings.state_dir / f"{job_name}.log" write_private_bytes(log_path, log_payload) lines = decode_output(log_payload).splitlines() require(bool(lines), f"Controller log is empty: {job_name}") try: result = json.loads(lines[-1]) except json.JSONDecodeError as exc: raise RegressionError( f"Controller log did not end in result JSON: {job_name}" ) from exc require( isinstance(result, dict), f"Controller result is not an object: {job_name}", ) expected_run_id = derive_run_id( job_name, self.settings.cron_job_name, ) require( result.get("run_id") == expected_run_id, "Controller result RUN_ID differs from the documented " "Controller Job identity", ) write_json( self.settings.state_dir / f"{job_name}.result.json", result, pretty=True, ) self.last_controller_condition = condition self.last_controller_result = result log(f"Controller {job_name} ended with {condition}") summary_keys = ( "status", "stage", "run_id", "backup_name", "selected_clusters", "packed_clusters", "verified_complete_markers", "mysql_updated_clusters", "unverified_complete_marker_clusters", "watermark_worker_status", "error", ) print( json.dumps( {key: result.get(key) for key in summary_keys}, indent=2, sort_keys=True, ) ) return result def result_count(self, name: str) -> int: value = self.last_controller_result.get(name) if type(value) is not int: raise RegressionError(f"Controller result has invalid {name}") return value def assert_accounted_result( self, expected_clusters: int, expected_identities: int, expected_backup_name: str, skipped_clusters: Sequence[str], ) -> None: result = self.last_controller_result require(result.get("status") == "succeeded", "Controller did not succeed") require(result.get("stage") == "complete", "Controller did not complete") require( result.get("backup_name") == expected_backup_name, "Controller used a different backup", ) require( self.result_count("selected_clusters") == expected_clusters, "selected cluster count differs from input", ) require( self.result_count("packed_clusters") == expected_clusters, "packed cluster count differs from input", ) require( self.result_count("packed_meta_objects") == expected_identities, "packed identity count differs from Export results", ) require( self.result_count("mysql_update_failed_clusters") == 0, "one or more MySQL updates failed", ) require( result.get("watermark_worker_status") == "succeeded", "Watermark worker did not succeed", ) verified = self.result_count("verified_complete_markers") require( 0 <= verified <= expected_identities, "verified complete marker count is outside the identity set", ) skipped_count = ( self.result_count("unverified_complete_marker_clusters") + self.result_count("newer_complete_marker_clusters") + self.result_count("newer_mysql_watermark_clusters") ) require( skipped_count == len(skipped_clusters), "Controller skipped count differs from marker classification", ) require( self.result_count("mysql_updated_clusters") + skipped_count == expected_clusters, "updated and skipped clusters do not account for every input", ) require( self.last_controller_condition == "Complete", "successful Controller result did not have Complete condition", ) def assert_noop_result(self, expected_backup_name: str) -> None: result = self.last_controller_result require(result.get("status") == "succeeded", "no-op did not succeed") require(result.get("stage") == "complete", "no-op did not complete") require( result.get("backup_name") == expected_backup_name, "no-op used a different backup", ) for name in ( "selected_clusters", "pack_jobs", "export_jobs", "watermark_jobs", "mysql_updated_clusters", "unverified_complete_marker_clusters", "mysql_update_failed_clusters", "newer_complete_marker_clusters", "newer_mysql_watermark_clusters", ): require(self.result_count(name) == 0, f"no-op has nonzero {name}") require( self.last_controller_condition == "Complete", "no-op Controller result did not have Complete condition", ) def collect_export_identities( self, run_id: str, label: str, ) -> List[str]: jobs = self.kubectl_json( "get", "jobs", "-l", "cse.tidbcloud.com/workflow-name=" f"{run_id},cse.tidbcloud.com/stage=export", "-o", "json", ) job_names = sorted( item.get("metadata", {}).get("name") for item in jobs.get("items", []) if isinstance(item.get("metadata", {}).get("name"), str) ) require(bool(job_names), f"no Export Jobs found for workflow {run_id}") combined_parts = [] for job_name in job_names: payload = self.kubectl_raw( "logs", f"job/{job_name}", "-c", "offline-dumpling", ).stdout write_private_bytes( self.settings.state_dir / f"{label}.{job_name}.log", payload, ) combined_parts.append(decode_output(payload)) combined_log = "\n".join( part.rstrip("\n") for part in combined_parts ) + "\n" write_private_text( self.settings.state_dir / f"{label}.export.log", combined_log, ) patterns = ( re.compile( r"(?:^|\s)target=" + re.escape(OUT_PREFIX) + r"([^/\s]+)/" ), re.compile( r"(?:^|\s)marker=" + re.escape(OUT_PREFIX) + r"([^/\s]+)/complete" ), ) identities = sorted( { match.group(1) for pattern in patterns for match in pattern.finditer(combined_log) } ) require(bool(identities), "no identities were found in Export logs") write_json( self.settings.state_dir / f"{label}-identities.json", identities, ) return identities def validate_identity_clusters( self, clusters: Sequence[str], identities: Sequence[str], ) -> None: observed = sorted({split_identity(identity)[0] for identity in identities}) if observed != list(clusters): missing = sorted(set(clusters) - set(observed)) unexpected = sorted(set(observed) - set(clusters)) raise RegressionError( "packed identity clusters differ from initialized clusters: " f"missing={missing[:3]!r} unexpected={unexpected[:3]!r}" ) log(f"identity mapping: clusters={len(clusters)} identities={len(identities)}") def marker_key(self, identity: str) -> str: return f"{EXPORT_KEY_PREFIX}{identity}/complete" def marker_snapshot(self, identity: str, label: str) -> Dict[str, Any]: key = self.marker_key(identity) head = self.aws_raw( "s3api", "head-object", "--bucket", EXPORT_BUCKET, "--key", key, "--output", "json", check=False, ) if head.returncode == 0: head_payload = parse_json_bytes(head.stdout, f"marker head {identity}") body_payload = self.aws_raw( "s3", "cp", f"s3://{EXPORT_BUCKET}/{key}", "-", "--only-show-errors", ).stdout body = decode_output(body_payload).rstrip("\n") snapshot = { "exists": True, "body": body, "content_length": head_payload.get("ContentLength"), "etag": head_payload.get("ETag"), "last_modified": head_payload.get("LastModified"), "metadata": head_payload.get("Metadata"), } else: error_text = decode_output(head.stderr) if PERMISSION_ERROR_PATTERN.search(error_text): raise PermissionDenied(error_text.strip() or "permission denied") if re.search( r"(^|[^0-9])404([^0-9]|$)|Not Found|NoSuchKey", error_text, ): snapshot = {"exists": False} else: if error_text: print(error_text, file=sys.stderr) raise RegressionError(f"cannot read marker for {identity}") write_json( self.settings.state_dir / f"{label}.marker.json", snapshot, pretty=True, ) return snapshot def object_count(self, key_prefix: str) -> int: result = self.aws_raw( "s3api", "list-objects-v2", "--bucket", EXPORT_BUCKET, "--prefix", key_prefix, "--max-keys", "1", "--query", "KeyCount", "--output", "text", ) try: return int(decode_output(result.stdout).strip()) except ValueError as exc: raise RegressionError( f"invalid object count for prefix {key_prefix}" ) from exc def assert_pointer_has_objects(self, identity: str, pointer: str) -> None: count = self.object_count(f"{EXPORT_KEY_PREFIX}{identity}/{pointer}/") require( count > 0, f"expected retained output for {identity}/{pointer}", ) def assert_pointer_has_no_objects(self, identity: str, pointer: str) -> None: count = self.object_count(f"{EXPORT_KEY_PREFIX}{identity}/{pointer}/") require( count == 0, f"stale output was retained for {identity}/{pointer}", ) def direct_legacy_count(self, identity: str) -> int: identity_prefix = f"{EXPORT_KEY_PREFIX}{identity}/" payload = self.aws_json( "s3api", "list-objects-v2", "--bucket", EXPORT_BUCKET, "--prefix", identity_prefix, "--output", "json", ) count = 0 for item in payload.get("Contents") or []: key = item.get("Key") if not isinstance(key, str) or not key.startswith(identity_prefix): continue relative = key[len(identity_prefix) :] if "/" not in relative and relative.endswith(".gpg"): count += 1 return count def assert_no_legacy_objects(self, identities: Sequence[str]) -> None: require(bool(identities), "successful identity set is empty") for identity in identities: count = self.direct_legacy_count(identity) require( count == 0, f"identity {identity} has {count} direct legacy object(s)", ) def snapshot_marker_set( self, identities: Sequence[str], label: str, ) -> List[Dict[str, Any]]: require(bool(identities), "cannot snapshot an empty identity set") records = [] for index, identity in enumerate(identities): snapshot = self.marker_snapshot(identity, f"{label}-{index}") records.append({**snapshot, "identity": identity}) records.sort(key=lambda record: record["identity"]) write_json( self.settings.state_dir / f"{label}.markers.json", records, pretty=True, ) return records def classify_marker_set( self, clusters: Sequence[str], identities: Sequence[str], backup_time: str, export_instance: str, label: str, ) -> Dict[str, Any]: expected_pointer = f"{backup_time}:{export_instance}" groups: Dict[str, List[Tuple[str, bool]]] = { cluster: [] for cluster in clusters } valid_identities = [] invalid_markers = [] for index, identity in enumerate(identities): snapshot = self.marker_snapshot(identity, f"{label}-{index}") cluster_id, unused_keyspace = split_identity(identity) require( cluster_id in groups, f"marker snapshot has unexpected cluster: {cluster_id!r}", ) metadata = snapshot.get("metadata") valid = ( snapshot.get("exists") is True and snapshot.get("body") == expected_pointer and snapshot.get("content_length") == len(expected_pointer) and isinstance(metadata, dict) and metadata.get("backup-time") == backup_time and metadata.get("job-instance-name") == export_instance ) groups[cluster_id].append((identity, valid)) (valid_identities if valid else invalid_markers).append(identity) empty_clusters = [ cluster for cluster, entries in groups.items() if not entries ] require( not empty_clusters, "no packed identity found for input cluster(s): " f"{empty_clusters[:3]!r}", ) successful_clusters = sorted( cluster for cluster, entries in groups.items() if all(valid for unused_identity, valid in entries) ) successful_cluster_set = set(successful_clusters) successful_identities = sorted( identity for cluster, entries in groups.items() if cluster in successful_cluster_set for identity, unused_valid in entries ) classification = { "input_clusters": list(clusters), "successful_clusters": successful_clusters, "skipped_clusters": sorted( set(clusters) - successful_cluster_set ), "successful_identities": successful_identities, "valid_identities": sorted(valid_identities), "invalid_markers": sorted(invalid_markers), } write_json( self.settings.state_dir / f"{label}-marker-classification.json", classification, pretty=True, ) for identity in classification["valid_identities"]: self.assert_pointer_has_objects(identity, expected_pointer) log( f"{label} markers: input={len(clusters)} " f"successful={len(successful_clusters)} " f"skipped={len(classification['skipped_clusters'])} " f"invalid={len(invalid_markers)}" ) return classification def assert_eligible_cluster_set( self, expected: Sequence[str], cutoff: datetime, label: str, ) -> None: actual = mysql_control.list_eligible_clusters( self.settings.mysql_defaults_file, cutoff, self.settings.minimum_target_clusters, ) write_json( self.settings.state_dir / f"{label}.eligible-clusters.json", actual, ) require( actual == list(expected), f"eligible cluster set differs from skipped set after {label}", ) def temporary_resource_inventory( self, run_id: str, ) -> Mapping[str, Any]: return self.kubectl_json( "get", "secrets,configmaps", "-l", f"cse.tidbcloud.com/workflow-name={run_id}", "-o", "json", ) def validate_temporary_resource_inventory( self, payload: Mapping[str, Any], run_id: str, ) -> Sequence[Mapping[str, Any]]: items = payload.get("items") if not isinstance(items, list): raise RegressionError("temporary resource inventory has no items") allowed = { f"pack-backup-offline-{run_id}-cluster-ids": ( "ConfigMap", "cluster-selection", ), f"dumpling-packed-export-{run_id}-init": ( "ConfigMap", "export-init", ), f"packed-backup-watermark-{run_id}-input": ( "ConfigMap", "watermark-input", ), } for item in items: if not isinstance(item, dict): raise RegressionError("temporary resource is not an object") kind = item.get("kind") metadata = item.get("metadata") or {} name = metadata.get("name") labels = metadata.get("labels") or {} require( kind in {"Secret", "ConfigMap"}, f"unexpected temporary resource kind: {kind!r}", ) require( labels.get("cse.tidbcloud.com/workflow-name") == run_id, f"temporary resource label mismatch on {kind}/{name}", ) if not isinstance(name, str) or name not in allowed: raise RegressionError( f"unexpected temporary resource: {kind}/{name}" ) expected_kind, expected_role = allowed[name] require( kind == expected_kind, f"temporary resource kind mismatch on {kind}/{name}", ) require( labels.get("cse.tidbcloud.com/resource-role") == expected_role, f"temporary resource role mismatch on {kind}/{name}", ) return items def cleanup_temporary_resources( self, run_id: str, ) -> None: selector = f"cse.tidbcloud.com/workflow-name={run_id}" inventory = self.temporary_resource_inventory(run_id) write_json( self.settings.state_dir / f"{run_id}.temporary-cleanup-inventory.json", inventory, pretty=True, ) items = self.validate_temporary_resource_inventory(inventory, run_id) if items: self.kubectl_raw( "delete", "secrets,configmaps", "-l", selector, "--ignore-not-found=true", "--wait=true", echo=True, ) remaining = self.temporary_resource_inventory(run_id).get("items", []) require( not remaining, f"temporary resource cleanup left resources for {run_id}", ) log(f"cleaned temporary Secret/ConfigMap resources for {run_id}") def cleanup(self) -> bool: cleanup_failed = False for controller_job, run_id in reversed(self.cleanup_targets): if run_id not in self.terminal_run_ids: print( "[e2e] Controller did not reach a terminal condition; " "retaining temporary resources for " f"{controller_job}", file=sys.stderr, ) cleanup_failed = True continue try: self.cleanup_temporary_resources(run_id) except PermissionDenied as exc: print( "[e2e] permission error during cleanup; stopping " f"immediately: {exc}", file=sys.stderr, ) return False except Exception as exc: print( "[e2e] error: failed to clean temporary resources for " f"{run_id}: {exc}", file=sys.stderr, ) cleanup_failed = True return not cleanup_failed def inventory_run(self, controller_job: str, run_id: str) -> List[str]: lines = [f"job.batch/{controller_job}"] names = self.kubectl_text( "get", "jobs,pods,secrets,configmaps,persistentvolumeclaims", "-l", f"cse.tidbcloud.com/workflow-name={run_id}", "-o", "name", ) lines.extend(line for line in names.splitlines() if line) return lines def enable_cron_job(self) -> None: self.kubectl_raw( "patch", f"cronjob/{self.settings.cron_job_name}", "--type=merge", "-p", '{"spec":{"suspend":false}}', ) cron_job = self.kubectl_json( "get", f"cronjob/{self.settings.cron_job_name}", "-o", "json", ) spec = cron_job.get("spec", {}) service_account = ( spec.get("jobTemplate", {}) .get("spec", {}) .get("template", {}) .get("spec", {}) .get("serviceAccountName") ) log( f"cronjob={self.settings.cron_job_name} " f"schedule={spec.get('schedule')} suspend={spec.get('suspend')} " f"serviceAccount={service_account}" ) def write_resource_inventory( self, runs: Sequence[Tuple[str, str]], ) -> None: lines = [ f"configmap/{self.settings.gpg_config_map}", f"configmap/{self.settings.workflow_config_map}", f"secret/{self.settings.mysql_secret}", f"secret/{self.settings.lark_secret}", f"cronjob.batch/{self.settings.cron_job_name}", ] for controller_job, run_id in runs: lines.extend(self.inventory_run(controller_job, run_id)) content = "\n".join(lines) + "\n" write_private_text( self.settings.state_dir / "resource-inventory.txt", content, ) print(content, end="") def run(self) -> None: self.preflight() self.flush_wal() later_date = datetime.strptime( self.settings.backup_prefix, BACKUP_DATE_FORMAT, ) earlier_date_prefix = (later_date - timedelta(days=1)).strftime( BACKUP_DATE_FORMAT ) earlier_backup_name = self.discover_first_backup_name( earlier_date_prefix ) later_backup_name = self.discover_first_backup_name( self.settings.backup_prefix ) earlier_backup_selector = earlier_backup_name.removesuffix(".meta") later_backup_selector = later_backup_name.removesuffix(".meta") for backup_name, backup_selector in ( (earlier_backup_name, earlier_backup_selector), (later_backup_name, later_backup_selector), ): pinned_backup_name = self.discover_first_backup_name( backup_selector ) require( pinned_backup_name == backup_name, f"exact backup lookup changed {backup_name} to " f"{pinned_backup_name}", ) earlier_backup_time = backup_timestamp(earlier_backup_name) later_backup_time = backup_timestamp(later_backup_name) earlier_cutoff = backup_datetime(earlier_backup_name) later_cutoff = backup_datetime(later_backup_name) require( earlier_cutoff < later_cutoff, "adjacent backup cutoffs are not increasing", ) earlier_backup_meta, earlier_source_uri = self.download_backup_meta( earlier_backup_name, "b1", ) later_backup_meta, later_source_uri = self.download_backup_meta( later_backup_name, "b2", ) seed_sql, expected_target_clusters = self.generate_seed_sql( earlier_backup_meta, earlier_source_uri, later_backup_meta, later_source_uri, earlier_cutoff, ) log( f"pinned adjacent first backups B1={earlier_backup_name} " f"({earlier_backup_time}) B2={later_backup_name} " f"({later_backup_time})" ) first_execution = f"a{self.settings.run_token}" second_execution = f"b{self.settings.run_token}" third_execution = f"c{self.settings.run_token}" first_run = derive_run_id( f"{self.settings.cron_job_name}-{first_execution}", self.settings.cron_job_name, ) second_run = derive_run_id( f"{self.settings.cron_job_name}-{second_execution}", self.settings.cron_job_name, ) third_run = derive_run_id( f"{self.settings.cron_job_name}-{third_execution}", self.settings.cron_job_name, ) first_export = f"dumpling-packed-export-{first_run}" second_export = f"dumpling-packed-export-{second_run}" third_export = f"dumpling-packed-export-{third_run}" first_pointer = f"{earlier_backup_time}:{first_export}" second_pointer = f"{later_backup_time}:{second_export}" self.deploy_cron_job() seed_result = mysql_control.seed_database( self.settings.mysql_defaults_file, seed_sql, self.settings.minimum_target_clusters, ) log(f"database seed: {json.dumps(seed_result, sort_keys=True)}") prepared = mysql_control.prepare_database( self.settings.mysql_defaults_file, earlier_cutoff, self.settings.minimum_target_clusters, ) log(f"database prepare: {json.dumps(prepared, sort_keys=True)}") target_clusters = mysql_control.list_clusters( self.settings.mysql_defaults_file, self.settings.minimum_target_clusters, ) require( target_clusters == expected_target_clusters, "seeded database differs from the backup cluster intersection", ) write_json( self.settings.state_dir / "target-clusters.json", target_clusters, ) target_cluster_count = len(target_clusters) log( f"round 1/3: export all {target_cluster_count} clusters " f"from B1 {earlier_backup_name}" ) mysql_control.assert_all_clusters_eligible( self.settings.mysql_defaults_file, earlier_cutoff, self.settings.minimum_target_clusters, ) first_job, observed_first_run = self.create_controller_job( first_execution, earlier_backup_selector, ) require(observed_first_run == first_run, "round 1 RUN_ID changed") self.wait_controller_job(first_job) first_identities = self.collect_export_identities(first_run, "round-1") self.validate_identity_clusters(target_clusters, first_identities) first_classification = self.classify_marker_set( target_clusters, first_identities, earlier_backup_time, first_export, "round-1", ) first_skipped_clusters = first_classification["skipped_clusters"] first_success_identities = first_classification[ "successful_identities" ] self.assert_accounted_result( target_cluster_count, len(first_identities), earlier_backup_name, first_skipped_clusters, ) self.assert_eligible_cluster_set( first_skipped_clusters, earlier_cutoff, "round-1", ) self.assert_no_legacy_objects(first_success_identities) for identity in first_success_identities: self.assert_pointer_has_objects(identity, first_pointer) log("round 1 passed: every input is updated or explicitly skipped") log( "round 2/3: preserve B1 watermarks, simulate updates between " "B1 and B2, then export B2" ) advanced = mysql_control.advance_updated_at( self.settings.mysql_defaults_file, earlier_cutoff, later_cutoff, self.settings.minimum_target_clusters, ) log(f"database advance: {json.dumps(advanced, sort_keys=True)}") mysql_control.assert_all_clusters_eligible( self.settings.mysql_defaults_file, later_cutoff, self.settings.minimum_target_clusters, ) second_target_clusters = mysql_control.list_clusters( self.settings.mysql_defaults_file, self.settings.minimum_target_clusters, ) require( second_target_clusters == target_clusters, "round 2 target cluster set changed", ) write_json( self.settings.state_dir / "round-2-target-clusters.json", second_target_clusters, ) second_job, observed_second_run = self.create_controller_job( second_execution, later_backup_selector, ) require(observed_second_run == second_run, "round 2 RUN_ID changed") self.wait_controller_job(second_job) second_identities = self.collect_export_identities(second_run, "round-2") self.validate_identity_clusters(target_clusters, second_identities) second_classification = self.classify_marker_set( target_clusters, second_identities, later_backup_time, second_export, "round-2", ) second_skipped_clusters = second_classification["skipped_clusters"] second_success_identities = second_classification[ "successful_identities" ] self.assert_accounted_result( target_cluster_count, len(second_identities), later_backup_name, second_skipped_clusters, ) self.assert_eligible_cluster_set( second_skipped_clusters, later_cutoff, "round-2", ) self.assert_no_legacy_objects(second_success_identities) require( bool(first_success_identities), "round 1 completed no identity; pointer cleanup was not exercised", ) common_success_identities = sorted( set(first_success_identities) & set(second_success_identities) ) common_success_clusters = sorted( {split_identity(identity)[0] for identity in common_success_identities} ) require( bool(common_success_identities), "no shared identity completed in both rounds; older-version " "cleanup was not exercised", ) write_json( self.settings.state_dir / "round-1-and-2-successful-clusters.json", common_success_clusters, ) write_json( self.settings.state_dir / "round-1-and-2-successful-identities.json", common_success_identities, ) for identity in common_success_identities: self.assert_pointer_has_no_objects(identity, first_pointer) self.assert_pointer_has_objects(identity, second_pointer) log( "round 2 passed: common successes point to B2 and their B1 " "versions were removed" ) log("round 3/3: do not update MySQL; retry only eligible rows") third_input_clusters = mysql_control.list_eligible_clusters( self.settings.mysql_defaults_file, later_cutoff, self.settings.minimum_target_clusters, ) require( third_input_clusters == second_skipped_clusters, "round 3 input differs from round 2 skipped cluster set", ) write_json( self.settings.state_dir / "round-3-input-clusters.json", third_input_clusters, ) third_cluster_set = set(third_input_clusters) unaffected_identities = sorted( identity for identity in second_identities if split_identity(identity)[0] not in third_cluster_set ) write_json( self.settings.state_dir / "round-3-unaffected-identities.json", unaffected_identities, ) before_markers: Optional[List[Dict[str, Any]]] = None if not third_input_clusters: before_markers = self.snapshot_marker_set( second_identities, "round-3-before", ) elif unaffected_identities: before_markers = self.snapshot_marker_set( unaffected_identities, "round-3-before-unaffected", ) third_job, observed_third_run = self.create_controller_job( third_execution, later_backup_selector, ) require(observed_third_run == third_run, "round 3 RUN_ID changed") self.wait_controller_job(third_job) if not third_input_clusters: self.assert_noop_result(later_backup_name) after_markers = self.snapshot_marker_set( second_identities, "round-3-after", ) require( before_markers == after_markers, "no-op changed one or more complete markers", ) third_skipped_clusters: Sequence[str] = [] self.assert_eligible_cluster_set([], later_cutoff, "round-3") log("round 3 passed: zero inputs and markers unchanged") else: third_identities = self.collect_export_identities( third_run, "round-3", ) self.validate_identity_clusters( third_input_clusters, third_identities, ) third_classification = self.classify_marker_set( third_input_clusters, third_identities, later_backup_time, third_export, "round-3", ) third_skipped_clusters = third_classification["skipped_clusters"] self.assert_accounted_result( len(third_input_clusters), len(third_identities), later_backup_name, third_skipped_clusters, ) self.assert_eligible_cluster_set( third_skipped_clusters, later_cutoff, "round-3", ) if unaffected_identities: after_markers = self.snapshot_marker_set( unaffected_identities, "round-3-after-unaffected", ) require( before_markers == after_markers, "round 3 changed a marker outside its eligible input", ) log("round 3 passed: every retry input is accounted for") log( f"completed rows are at {later_backup_time}; skipped rows remain " "eligible" ) self.enable_cron_job() self.write_resource_inventory( [ (first_job, first_run), (second_job, second_run), (third_job, third_run), ] ) log("all three rounds passed; CronJob schedule is enabled") log( "temporary Secret/ConfigMap resources will be cleaned at script " "exit; completed Jobs are left to their TTL" ) log(f"evidence: {self.settings.state_dir}") def main() -> int: os.umask(0o077) regression: Optional[Regression] = None permission_denied = False status = 0 try: regression = Regression(load_settings()) regression.run() except PermissionDenied as exc: permission_denied = True status = 1 print(f"[e2e] permission error; stopping immediately: {exc}", file=sys.stderr) except (OSError, RegressionError, ValueError, pymysql.MySQLError) as exc: status = 1 print(f"[e2e] error: {type(exc).__name__}: {exc}", file=sys.stderr) except KeyboardInterrupt: status = 130 print("[e2e] interrupted", file=sys.stderr) if regression is not None and not permission_denied: if not regression.cleanup() and status == 0: status = 1 elif regression is not None and regression.cleanup_targets: print( "[e2e] cleanup was skipped after a permission error; " f"inspect {regression.settings.state_dir}", file=sys.stderr, ) return status if __name__ == "__main__": raise SystemExit(main())