#!/usr/bin/env python3 """Reconcile one named packed-backup-to-CSV export workflow. The process is reentrant: the exact backup and immutable S3 cluster selection metadata are stored in a ConfigMap, while phase progress is derived from Jobs. Failed Jobs require explicit intervention, existing S3 output is preserved, and generated Jobs carry the workflow-configured Kubernetes TTL. """ import argparse import hashlib import json import os import re import shutil import subprocess import sys import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Set, Tuple from urllib.parse import urlencode, urlsplit import pymysql from boto3.session import Session from botocore.config import Config from packed_backup_notification import send_notification from packed_backup_workflow_spec import ( SpecError, WorkflowSpec, load_workflow_spec, ) SCRIPT_DIR = Path(__file__).resolve().parent GENERATE_PACK = SCRIPT_DIR / "generate-pack-backup-offline.sh" LIST_PACK_TASKS = SCRIPT_DIR / "list-completed-pack-backup-offline-tasks.sh" GENERATE_EXPORT = SCRIPT_DIR / "generate_dumpling_packed_backup_export.py" GENERATE_WATERMARK = SCRIPT_DIR / "generate_packed_backup_watermark.py" NAMESPACE = "tidb-serverless" TIKV_API_CONTAINER = "tikv-api" TIKV_API_POD = "tikv-api-0" TIKV_API_STATEFUL_SET = "tikv-api" INSTANCE_LABEL = "app.kubernetes.io/instance" ACTIVE_CLUSTER_IDS_TABLE = "active_cluster_ids" CLUSTER_BLOCKLIST_TABLE = "cluster_blocklist" COMPLETE_MARKER_BACKUP_TIME_FORMAT = "%Y-%m-%dT%H:%M:%SZ" WATERMARK_CONTAINER = "packed-backup-watermark" WORKFLOW_CONFIG_MAP_APP_NAME = "packed-backup-csv-export-workflow" WORKFLOW_NAME_LABEL = "cse.tidbcloud.com/workflow-name" RESOURCE_ROLE_LABEL = "cse.tidbcloud.com/resource-role" WORKFLOW_BACKUP_NAME_DATA_KEY = "backup-name" WORKFLOW_CLUSTER_ID_OBJECT_URI_DATA_KEY = "cluster-id-object-uri" WORKFLOW_CLUSTER_ID_OBJECT_REGION_DATA_KEY = "cluster-id-object-region" WORKFLOW_CLUSTER_ID_SHA256_DATA_KEY = "cluster-id-object-sha256" WORKFLOW_CLUSTER_ID_BYTES_DATA_KEY = "cluster-id-object-bytes" WORKFLOW_CLUSTER_ID_COUNT_DATA_KEY = "cluster-id-count" CLUSTER_ID_OBJECT_ROOT = "packed-backup-csv-export/cluster-id-lists" CLUSTER_ID_SELECT_FETCH_SIZE = 1000 S3_READ_CHUNK_BYTES = 1024 * 1024 JOB_POLL_INTERVAL_SECONDS = 10 JOB_HEARTBEAT_INTERVAL_SECONDS = 300 class WorkflowError(Exception): pass @dataclass(frozen=True) class DfsStorageConfig: bucket: str region: str @dataclass(frozen=True) class ClusterIdObject: bucket: str key: str region: str sha256: str byte_count: int cluster_count: int @property def uri(self) -> str: return f"s3://{self.bucket}/{self.key}" @dataclass(frozen=True) class WorkflowRecord: name: str cluster_id_object: ClusterIdObject backup_name: str @dataclass(frozen=True) class WorkflowEnsureResult: record: Optional[WorkflowRecord] backup_name: str backup_cutoff: datetime def log(message: str) -> None: timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") print(f"{timestamp} {message}", file=sys.stderr, flush=True) def log_output(description: str, output: str) -> None: for line in output.strip().splitlines(): log(f"{description}: {line}") def non_empty(value: str) -> str: if not value: raise argparse.ArgumentTypeError("must not be empty") return value def kubernetes_name(value: str) -> str: if ( len(value) > 63 or re.fullmatch(r"[a-z0-9]([-a-z0-9]*[a-z0-9])?", value) is None ): raise argparse.ArgumentTypeError( "must be a Kubernetes DNS label of at most 63 characters" ) return value def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( "Reconcile a named packed-backup-to-CSV workflow from the " "resources already present in Kubernetes. Missing Pack, Export, " "or Watermark stages perform real Kubernetes changes." ), epilog=( "KUBERNETES_CONTEXT_NAME is included as the context in results and " "notifications. When LARK_BOT_WEBHOOK is set, one short " "notification is sent on final success or failure." ), formatter_class=argparse.ArgumentDefaultsHelpFormatter, allow_abbrev=False, ) workflow = parser.add_argument_group("workflow identity") workflow.add_argument( "--name", type=kubernetes_name, required=True, metavar="NAME", help="stable workflow name reused by every reconciliation", ) workflow.add_argument( "--spec", type=non_empty, required=True, metavar="PATH", help="declarative workflow YAML file", ) backup = parser.add_argument_group("backup discovery") backup.add_argument( "--backup-from-prefix", type=non_empty, default=datetime.now(timezone.utc).strftime("%Y%m%d"), metavar="PREFIX", help="backup API from_prefix; the first returned backup is selected", ) mysql = parser.add_argument_group("MySQL cluster selection") mysql.add_argument( "--mysql-defaults-file", type=non_empty, required=True, metavar="PATH", help="MySQL option file read by PyMySQL; its contents are never logged", ) local = parser.add_argument_group("local invocation inputs") local.add_argument( "--local-gpg", type=non_empty, required=True, metavar="PATH", help="local OpenPGP public-key file used by Export", ) return parser def command_failure_output(stdout: str, stderr: str) -> str: output = stderr.strip() or stdout.strip() or "(no command output)" limit = 8000 if len(output) > limit: output = output[:limit] + "\n...[truncated]" return output def run_command( command: Sequence[str], description: str, input_text: Optional[str] = None, ) -> str: try: result = subprocess.run( list(command), input=input_text, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", check=False, ) except OSError as exc: raise WorkflowError(f"cannot start {description}: {exc}") from exc if result.returncode != 0: output = command_failure_output(result.stdout, result.stderr) raise WorkflowError( f"{description} failed with exit code {result.returncode}:\n{output}" ) if result.stderr.strip(): log_output(description, result.stderr) return result.stdout def parse_json(text: str, description: str) -> Any: try: return json.loads(text) except json.JSONDecodeError as exc: raise WorkflowError( f"{description} returned invalid JSON at line {exc.lineno} " f"column {exc.colno}: {exc.msg}" ) from exc def backup_cutoff_from_name(backup_name: str) -> datetime: try: parsed = datetime.strptime(backup_name, "%Y%m%d/%H%M%S.meta") except ValueError as exc: raise WorkflowError( f"backup name does not contain an expected UTC timestamp: " f"{backup_name!r}" ) from exc return parsed.replace(tzinfo=timezone.utc) def canonical_json(value: Any) -> str: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) def job_manifest_with_ttl( text: str, description: str, ttl_seconds_after_finished: int, ) -> str: manifest = text.strip() if not manifest: raise WorkflowError(f"{description} generator produced an empty manifest") payload = parse_json(manifest, f"{description} manifest") if not isinstance(payload, dict): raise WorkflowError(f"{description} manifest is not a JSON object") items = payload.get("items") if payload.get("kind") != "List" or not isinstance(items, list): raise WorkflowError( f"{description} manifest is not a Kubernetes List" ) for index, item in enumerate(items): if not isinstance(item, dict): raise WorkflowError( f"{description} manifest item {index} is not an object" ) if item.get("kind") != "Job": continue job_spec = item.get("spec") if not isinstance(job_spec, dict): raise WorkflowError( f"{description} manifest Job item {index} has no spec" ) job_spec["ttlSecondsAfterFinished"] = ttl_seconds_after_finished return canonical_json(payload) def mysql_defaults_path(value: str) -> Path: return Path(value).expanduser() def mysql_operation_error(operation: str, exc: Exception) -> WorkflowError: return WorkflowError(f"{operation} failed ({type(exc).__name__})") def quietly_rollback_mysql(connection: Any) -> None: try: connection.rollback() except Exception: pass def quietly_close_mysql(connection: Any) -> None: try: connection.close() except Exception: pass def open_mysql_connection(defaults_file: str) -> Any: connection = None try: connection = pymysql.connect( read_default_file=str(mysql_defaults_path(defaults_file)), read_default_group="client", connect_timeout=10, read_timeout=30, write_timeout=30, autocommit=False, charset="utf8mb4", cursorclass=pymysql.cursors.SSCursor, ) with connection.cursor() as cursor: cursor.execute("SET time_zone = '+00:00'") cursor.execute("SELECT 1") if cursor.fetchone() != (1,): raise WorkflowError("MySQL connection check returned an invalid row") connection.rollback() return connection except Exception as exc: if connection is not None: quietly_close_mysql(connection) raise mysql_operation_error("MySQL connection", exc) from exc def select_cluster_ids( connection: Any, backup_cutoff: datetime, ) -> List[str]: cutoff = backup_cutoff.astimezone(timezone.utc).replace(tzinfo=None) query = ( f"SELECT active_clusters.cluster_id " f"FROM {ACTIVE_CLUSTER_IDS_TABLE} AS active_clusters " "WHERE active_clusters.updated_at > active_clusters.s3_data_updated_at " "AND active_clusters.updated_at <= %s " "AND NOT EXISTS (" f"SELECT 1 FROM {CLUSTER_BLOCKLIST_TABLE} AS blocklist " "WHERE blocklist.cluster_id = active_clusters.cluster_id" ") " "ORDER BY active_clusters.cluster_id" ) try: with connection.cursor() as cursor: cursor.execute(query, (cutoff,)) cluster_ids = [] while True: rows = cursor.fetchmany(CLUSTER_ID_SELECT_FETCH_SIZE) if not rows: break cluster_ids.extend(row[0] for row in rows) connection.rollback() except Exception as exc: quietly_rollback_mysql(connection) raise mysql_operation_error( "active cluster selection query", exc, ) from exc return cluster_ids def parse_positive_decimal(value: Any, description: str) -> int: if not isinstance(value, str) or re.fullmatch(r"[1-9][0-9]*", value) is None: raise WorkflowError(f"{description} must be a positive decimal integer") return int(value) def parse_s3_object_uri(value: Any, description: str) -> Tuple[str, str]: if not isinstance(value, str): raise WorkflowError(f"{description} must be an s3:// URI") parsed = urlsplit(value) if parsed.scheme != "s3" or not parsed.netloc: raise WorkflowError(f"{description} must be an s3:// URI") if parsed.query or parsed.fragment: raise WorkflowError( f"{description} must not contain a query or fragment" ) object_key = parsed.path[1:] if parsed.path.startswith("/") else parsed.path if not object_key: raise WorkflowError(f"{description} must include an object key") return parsed.netloc, object_key def cluster_id_object_from_workflow_data(data: Dict[str, Any]) -> ClusterIdObject: bucket, key = parse_s3_object_uri( data.get(WORKFLOW_CLUSTER_ID_OBJECT_URI_DATA_KEY), "workflow cluster ID object URI", ) region = data.get(WORKFLOW_CLUSTER_ID_OBJECT_REGION_DATA_KEY) if ( not isinstance(region, str) or not region or any(character.isspace() for character in region) ): raise WorkflowError( "workflow ConfigMap has no valid cluster ID object region" ) sha256 = data.get(WORKFLOW_CLUSTER_ID_SHA256_DATA_KEY) if not isinstance(sha256, str) or re.fullmatch(r"[0-9a-f]{64}", sha256) is None: raise WorkflowError( "workflow ConfigMap has no valid cluster ID object SHA-256" ) byte_count = parse_positive_decimal( data.get(WORKFLOW_CLUSTER_ID_BYTES_DATA_KEY), "workflow cluster ID object byte count", ) cluster_count = parse_positive_decimal( data.get(WORKFLOW_CLUSTER_ID_COUNT_DATA_KEY), "workflow cluster ID count", ) return ClusterIdObject( bucket=bucket, key=key, region=region, sha256=sha256, byte_count=byte_count, cluster_count=cluster_count, ) def load_workflow_record( kubectl: str, workflow_name: str, pack_instance: str, ) -> Optional[WorkflowRecord]: config_map_name = f"{pack_instance}-cluster-ids" output = run_command( [ kubectl, "--request-timeout=30s", "-n", NAMESPACE, "get", "configmap", config_map_name, "--ignore-not-found", "-o", "json", ], "load workflow ConfigMap", ) if not output.strip(): return None payload = parse_json(output, "workflow ConfigMap") if not isinstance(payload, dict): raise WorkflowError("workflow ConfigMap is not an object") metadata = payload.get("metadata") if not isinstance(metadata, dict): raise WorkflowError("workflow ConfigMap has no metadata") if metadata.get("name") != config_map_name: raise WorkflowError("workflow ConfigMap has the wrong name") if metadata.get("namespace") != NAMESPACE: raise WorkflowError("workflow ConfigMap has the wrong namespace") labels = metadata.get("labels") if isinstance(metadata, dict) else None if not isinstance(labels, dict): raise WorkflowError("workflow ConfigMap has no labels") if ( labels.get("app.kubernetes.io/name") != WORKFLOW_CONFIG_MAP_APP_NAME ): raise WorkflowError("workflow ConfigMap has the wrong app label") if labels.get(INSTANCE_LABEL) != pack_instance: raise WorkflowError( "workflow ConfigMap has the wrong instance label" ) if labels.get(WORKFLOW_NAME_LABEL) != workflow_name: raise WorkflowError("workflow ConfigMap has the wrong workflow label") resource_role = labels.get(RESOURCE_ROLE_LABEL) if resource_role is None: log( "warning: legacy workflow ConfigMap has no resource-role label" ) elif resource_role != "cluster-selection": raise WorkflowError("workflow ConfigMap has the wrong resource-role label") data = payload.get("data") if not isinstance(data, dict): raise WorkflowError("workflow ConfigMap has no data") cluster_id_object = cluster_id_object_from_workflow_data(data) backup_name = data.get(WORKFLOW_BACKUP_NAME_DATA_KEY) if not isinstance(backup_name, str) or not backup_name: raise WorkflowError("workflow ConfigMap has no valid backup-name data") backup_cutoff_from_name(backup_name) return WorkflowRecord( name=config_map_name, cluster_id_object=cluster_id_object, backup_name=backup_name, ) def cluster_id_content_from_ids(cluster_ids: Sequence[str]) -> bytes: if not cluster_ids: raise WorkflowError("cluster ID list is empty") seen: Set[str] = set() for index, cluster_id in enumerate(cluster_ids): if ( not isinstance(cluster_id, str) or not cluster_id or cluster_id != cluster_id.strip() or any(character in cluster_id for character in ("\x00", "\r", "\n")) ): raise WorkflowError(f"cluster ID at index {index} is invalid") if cluster_id in seen: raise WorkflowError("cluster ID list contains duplicates") seen.add(cluster_id) return ("\n".join(cluster_ids) + "\n").encode("utf-8") def parse_cluster_id_object_content(content: bytes) -> List[str]: if not content: raise WorkflowError("cluster ID object is empty") try: text = content.decode("utf-8") except UnicodeDecodeError as exc: raise WorkflowError("cluster ID object is not valid UTF-8") from exc if "\x00" in text: raise WorkflowError("cluster ID object contains a NUL byte") cluster_ids = [line.strip() for line in text.splitlines() if line.strip()] if not cluster_ids: raise WorkflowError("cluster ID object contains no cluster IDs") if len(cluster_ids) != len(set(cluster_ids)): raise WorkflowError("cluster ID object contains duplicate cluster IDs") return cluster_ids def workflow_record_manifest( workflow_name: str, pack_instance: str, cluster_id_object: ClusterIdObject, backup_name: str, ) -> str: metadata: Dict[str, Any] = { "name": f"{pack_instance}-cluster-ids", "namespace": NAMESPACE, "labels": { "app.kubernetes.io/name": WORKFLOW_CONFIG_MAP_APP_NAME, INSTANCE_LABEL: pack_instance, WORKFLOW_NAME_LABEL: workflow_name, RESOURCE_ROLE_LABEL: "cluster-selection", }, } data = { WORKFLOW_BACKUP_NAME_DATA_KEY: backup_name, WORKFLOW_CLUSTER_ID_OBJECT_URI_DATA_KEY: cluster_id_object.uri, WORKFLOW_CLUSTER_ID_OBJECT_REGION_DATA_KEY: cluster_id_object.region, WORKFLOW_CLUSTER_ID_SHA256_DATA_KEY: cluster_id_object.sha256, WORKFLOW_CLUSTER_ID_BYTES_DATA_KEY: str(cluster_id_object.byte_count), WORKFLOW_CLUSTER_ID_COUNT_DATA_KEY: str(cluster_id_object.cluster_count), } manifest = { "apiVersion": "v1", "kind": "ConfigMap", "metadata": metadata, "data": data, } return canonical_json(manifest) def create_workflow_record( kubectl: str, workflow_name: str, pack_instance: str, cluster_id_object: ClusterIdObject, backup_name: str, ) -> WorkflowRecord: manifest = workflow_record_manifest( workflow_name, pack_instance, cluster_id_object, backup_name, ) output = run_command( [kubectl, "--request-timeout=30s", "create", "-f", "-"], "create workflow ConfigMap", input_text=manifest, ) if output.strip(): log_output("create workflow ConfigMap", output) record = load_workflow_record( kubectl, workflow_name, pack_instance, ) if record is None: raise WorkflowError("created workflow ConfigMap cannot be loaded") return record def cluster_id_object_key(workflow_name: str) -> str: return f"{CLUSTER_ID_OBJECT_ROOT}/{workflow_name}/cluster-ids" def load_dfs_storage_config(kubectl: str) -> DfsStorageConfig: output = run_command( [ kubectl, "--request-timeout=30s", "-n", NAMESPACE, "get", "statefulset", TIKV_API_STATEFUL_SET, "-o", "json", ], "load DFS storage configuration", ) payload = parse_json(output, "DFS source StatefulSet") try: containers = payload["spec"]["template"]["spec"]["containers"] except (KeyError, TypeError) as exc: raise WorkflowError( "DFS source StatefulSet has no container list" ) from exc if not isinstance(containers, list): raise WorkflowError("DFS source StatefulSet container list is invalid") matches = [ container for container in containers if isinstance(container, dict) and container.get("name") == TIKV_API_CONTAINER ] if len(matches) != 1: raise WorkflowError( f"expected exactly one {TIKV_API_CONTAINER!r} container in " "DFS source StatefulSet" ) env = matches[0].get("env") if not isinstance(env, list): raise WorkflowError("DFS source container has no environment list") values: Dict[str, str] = {} for name in ("DFS_S3_BUCKET", "DFS_S3_REGION"): entries = [ entry for entry in env if isinstance(entry, dict) and entry.get("name") == name ] if len(entries) != 1: raise WorkflowError( f"expected exactly one {name} entry in DFS source container" ) value = entries[0].get("value") if not isinstance(value, str) or not value: raise WorkflowError(f"{name} must have a non-empty literal value") values[name] = value bucket = values["DFS_S3_BUCKET"] if ( any(character.isspace() for character in bucket) or any(character in bucket for character in ("/", "?", "#")) ): raise WorkflowError("DFS_S3_BUCKET has an invalid literal value") region = values["DFS_S3_REGION"] if any(character.isspace() for character in region): raise WorkflowError("DFS_S3_REGION has an invalid literal value") return DfsStorageConfig(bucket=bucket, region=region) def s3_error_code(exc: Exception) -> Optional[str]: response = getattr(exc, "response", None) if not isinstance(response, dict): return None error = response.get("Error") if not isinstance(error, dict) or error.get("Code") is None: return None return str(error["Code"]) def s3_operation_error(operation: str, exc: Exception) -> WorkflowError: code = s3_error_code(exc) suffix = f", code={code}" if code is not None else "" return WorkflowError( f"{operation} failed ({type(exc).__name__}{suffix})" ) def create_s3_client(region: str) -> Any: try: session = Session(region_name=region) return session.client( "s3", region_name=region, config=Config( connect_timeout=10, read_timeout=60, retries={"mode": "standard", "total_max_attempts": 10}, ), ) except Exception as exc: raise s3_operation_error("S3 client initialization", exc) from exc def read_s3_body(response: Dict[str, Any], description: str) -> bytes: body = response.get("Body") if body is None or not callable(getattr(body, "read", None)): raise WorkflowError(f"{description} returned no readable body") chunks = [] try: while True: chunk = body.read(S3_READ_CHUNK_BYTES) if not chunk: break if not isinstance(chunk, bytes): raise WorkflowError(f"{description} returned a non-bytes body") chunks.append(chunk) except WorkflowError: raise except Exception as exc: raise s3_operation_error(f"read {description}", exc) from exc finally: close = getattr(body, "close", None) if callable(close): close() return b"".join(chunks) def cluster_id_object_from_s3_response( storage: DfsStorageConfig, workflow_name: str, key: str, response: Dict[str, Any], ) -> Tuple[ClusterIdObject, str]: metadata = response.get("Metadata") if not isinstance(metadata, dict): raise WorkflowError("cluster ID object has no metadata") backup_name = metadata.get("backup-name") if not isinstance(backup_name, str) or not backup_name: raise WorkflowError("cluster ID object has no backup-name metadata") backup_cutoff_from_name(backup_name) if metadata.get("workflow-name") != workflow_name: raise WorkflowError("cluster ID object has the wrong workflow metadata") expected_sha256 = metadata.get("sha256") if ( not isinstance(expected_sha256, str) or re.fullmatch(r"[0-9a-f]{64}", expected_sha256) is None ): raise WorkflowError("cluster ID object has invalid SHA-256 metadata") expected_bytes = parse_positive_decimal( metadata.get("byte-count"), "cluster ID object byte-count metadata", ) expected_clusters = parse_positive_decimal( metadata.get("cluster-count"), "cluster ID object cluster-count metadata", ) content = read_s3_body(response, "cluster ID object download") content_length = response.get("ContentLength") if isinstance(content_length, bool) or not isinstance(content_length, int): raise WorkflowError("cluster ID object has no valid ContentLength") if content_length != len(content) or expected_bytes != len(content): raise WorkflowError("cluster ID object byte count does not match metadata") actual_sha256 = hashlib.sha256(content).hexdigest() if actual_sha256 != expected_sha256: raise WorkflowError("cluster ID object SHA-256 does not match metadata") cluster_ids = parse_cluster_id_object_content(content) if len(cluster_ids) != expected_clusters: raise WorkflowError("cluster ID object count does not match metadata") return ( ClusterIdObject( bucket=storage.bucket, key=key, region=storage.region, sha256=actual_sha256, byte_count=len(content), cluster_count=len(cluster_ids), ), backup_name, ) def load_existing_cluster_id_object( s3_client: Any, storage: DfsStorageConfig, workflow_name: str, ) -> Optional[Tuple[ClusterIdObject, str]]: key = cluster_id_object_key(workflow_name) try: response = s3_client.get_object(Bucket=storage.bucket, Key=key) except Exception as exc: if s3_error_code(exc) in ("404", "NoSuchKey", "NotFound"): return None raise s3_operation_error("load cluster ID object", exc) from exc if not isinstance(response, dict): raise WorkflowError("cluster ID object download returned a non-object") return cluster_id_object_from_s3_response( storage, workflow_name, key, response, ) def validate_cluster_id_object( s3_client: Any, workflow_name: str, cluster_id_object: ClusterIdObject, backup_name: str, ) -> None: try: response = s3_client.head_object( Bucket=cluster_id_object.bucket, Key=cluster_id_object.key, ) except Exception as exc: raise s3_operation_error("validate cluster ID object", exc) from exc if not isinstance(response, dict): raise WorkflowError("cluster ID object metadata returned a non-object") metadata = response.get("Metadata") if not isinstance(metadata, dict): raise WorkflowError("cluster ID object has no metadata") expected_metadata = { "workflow-name": workflow_name, "backup-name": backup_name, "sha256": cluster_id_object.sha256, "byte-count": str(cluster_id_object.byte_count), "cluster-count": str(cluster_id_object.cluster_count), } if any(metadata.get(key) != value for key, value in expected_metadata.items()): raise WorkflowError("cluster ID object metadata does not match workflow record") content_length = response.get("ContentLength") if content_length != cluster_id_object.byte_count: raise WorkflowError("cluster ID object size does not match workflow record") def upload_cluster_id_object( s3_client: Any, storage: DfsStorageConfig, workflow_name: str, cluster_ids: Sequence[str], backup_name: str, ) -> ClusterIdObject: content = cluster_id_content_from_ids(cluster_ids) key = cluster_id_object_key(workflow_name) cluster_id_object = ClusterIdObject( bucket=storage.bucket, key=key, region=storage.region, sha256=hashlib.sha256(content).hexdigest(), byte_count=len(content), cluster_count=len(cluster_ids), ) metadata = { "workflow-name": workflow_name, "backup-name": backup_name, "sha256": cluster_id_object.sha256, "byte-count": str(cluster_id_object.byte_count), "cluster-count": str(cluster_id_object.cluster_count), } try: s3_client.put_object( Bucket=storage.bucket, Key=key, Body=content, ContentType="text/plain; charset=utf-8", Metadata=metadata, IfNoneMatch="*", ) return cluster_id_object except Exception as exc: if s3_error_code(exc) not in ("412", "PreconditionFailed"): raise s3_operation_error("create cluster ID object", exc) from exc existing = load_existing_cluster_id_object( s3_client, storage, workflow_name, ) if existing is None: raise WorkflowError( "cluster ID object create was preconditioned but no object exists" ) existing_object, existing_backup_name = existing if existing_object != cluster_id_object or existing_backup_name != backup_name: raise WorkflowError( "existing cluster ID object conflicts with the current selection" ) return existing_object def parse_s3_prefix_uri(value: str, description: str) -> Tuple[str, str]: parsed = urlsplit(value) if parsed.scheme != "s3" or not parsed.netloc: raise WorkflowError(f"{description} must be an s3:// URI") if parsed.query or parsed.fragment: raise WorkflowError( f"{description} must not contain a query or fragment" ) object_prefix = parsed.path.lstrip("/").rstrip("/") if not object_prefix: raise WorkflowError(f"{description} must include an object prefix") return parsed.netloc, object_prefix def validate_pack_task_backup( tasks: Sequence[Dict[str, Any]], backup_name: str, ) -> None: for task in tasks: task_name = task.get("task_name") if not isinstance(task_name, str) or not task_name: raise WorkflowError( "completed Pack task has no non-empty task_name" ) task_prefix = task.get("pack_backup_prefix") if not isinstance(task_prefix, str): raise WorkflowError( "completed Pack task has no string pack_backup_prefix" ) unused_bucket, object_prefix = parse_s3_prefix_uri( task_prefix, "packed-backup task prefix", ) expected_suffix = ( f"/packed_backup/native/{backup_name}/{task_name}" ) if not f"/{object_prefix}".endswith(expected_suffix): raise WorkflowError( f"saved Pack task {task_name!r} does not belong to backup " f"{backup_name!r}" ) def require_runtime(args: argparse.Namespace) -> Dict[str, str]: commands = {} for name in ("bash", "jq", "kubectl"): path = shutil.which(name) if path is None: raise WorkflowError(f"{name} is required") commands[name] = path for path in ( GENERATE_PACK, LIST_PACK_TASKS, GENERATE_EXPORT, GENERATE_WATERMARK, ): if not path.is_file(): raise WorkflowError(f"required helper script is missing: {path}") gpg_path = Path(args.local_gpg) if not gpg_path.is_file(): raise WorkflowError( f"--local-gpg is not a regular file: '{args.local_gpg}'" ) defaults_path = mysql_defaults_path(args.mysql_defaults_file) if not defaults_path.is_file(): raise WorkflowError( f"--mysql-defaults-file is not a regular file: " f"{args.mysql_defaults_file!r}" ) return commands def discover_backup( kubectl: str, pod_name: str, from_prefix: str, ) -> str: url = ( "https://localhost:19500/api/v1/x/backups?" + urlencode({"from_prefix": from_prefix}) ) output = run_command( [ kubectl, "--request-timeout=90s", "-n", NAMESPACE, "exec", pod_name, "-c", TIKV_API_CONTAINER, "--", "curl", "--silent", "--show-error", "--fail", "--connect-timeout", "10", "--max-time", "60", "-k", url, ], "backup discovery", ) payload = parse_json(output, "backup discovery") if not isinstance(payload, dict): raise WorkflowError("backup discovery response must be a JSON object") backups = payload.get("data") if not isinstance(backups, list): raise WorkflowError("backup discovery response data must be an array") if not backups: raise WorkflowError( f"no backup returned for from_prefix '{from_prefix}'" ) first = backups[0] if not isinstance(first, dict): raise WorkflowError("the first backup entry must be a JSON object") backup_name = first.get("name") if not isinstance(backup_name, str) or not backup_name: raise WorkflowError("the first backup entry has no non-empty name") return backup_name def apply_manifest(kubectl: str, manifest: str, description: str) -> None: if not manifest.strip(): raise WorkflowError(f"{description} generator produced an empty manifest") output = run_command( [kubectl, "--request-timeout=30s", "apply", "-f", "-"], f"apply {description} manifest", input_text=manifest, ) if output.strip(): log_output(f"apply {description}", output) def condition_is_true(job: Dict[str, Any], condition_types: Sequence[str]) -> bool: status = job.get("status") if not isinstance(status, dict): return False conditions = status.get("conditions") if not isinstance(conditions, list): return False return any( isinstance(condition, dict) and condition.get("type") in condition_types and condition.get("status") == "True" for condition in conditions ) def failed_job_details(jobs: Sequence[Dict[str, Any]]) -> str: details = [] for job in jobs: metadata = job.get("metadata") name = metadata.get("name", "") if isinstance(metadata, dict) else "" status = job.get("status") conditions = status.get("conditions", []) if isinstance(status, dict) else [] messages = [] if isinstance(conditions, list): for condition in conditions: if ( isinstance(condition, dict) and condition.get("type") in ("Failed", "FailureTarget") and condition.get("status") == "True" ): reason = condition.get("reason") or "-" message = str(condition.get("message") or "-").replace("\n", " ") messages.append(f"{condition.get('type')}({reason}: {message})") details.append(f"{name}: {'; '.join(messages) or 'failed'}") return "; ".join(details) def kubernetes_job_name(job: Dict[str, Any]) -> str: metadata = job.get("metadata") name = metadata.get("name") if isinstance(metadata, dict) else None if not isinstance(name, str) or not name: raise WorkflowError("Kubernetes Job has no non-empty metadata.name") return name def list_instance_jobs( kubectl: str, instance: str, description: str, ) -> List[Dict[str, Any]]: selector = f"{INSTANCE_LABEL}={instance}" output = run_command( [ kubectl, "--request-timeout=30s", "-n", NAMESPACE, "get", "jobs", "-l", selector, "-o", "json", ], f"list {description} Jobs", ) payload = parse_json(output, f"{description} Job List") if not isinstance(payload, dict) or not isinstance(payload.get("items"), list): raise WorkflowError( f"{description} Job query returned an invalid Kubernetes List" ) jobs = payload["items"] if not all(isinstance(job, dict) for job in jobs): raise WorkflowError(f"{description} Job List contains a non-object") names = [kubernetes_job_name(job) for job in jobs] if len(set(names)) != len(names): raise WorkflowError(f"{description} Job List contains duplicate names") return sorted(jobs, key=kubernetes_job_name) def validate_expected_job_set( jobs: Sequence[Dict[str, Any]], expected_names: Sequence[str], description: str, ) -> None: actual = {kubernetes_job_name(job) for job in jobs} expected = set(expected_names) if actual != expected: missing = sorted(expected - actual) unexpected = sorted(actual - expected) details = [] if missing: details.append(f"missing={','.join(missing)}") if unexpected: details.append(f"unexpected={','.join(unexpected)}") raise WorkflowError( f"{description} Job set does not match the saved workflow: " + "; ".join(details) ) def validate_observed_job_subset( jobs: Sequence[Dict[str, Any]], expected_names: Sequence[str], description: str, ) -> Set[str]: actual = {kubernetes_job_name(job) for job in jobs} unexpected = sorted(actual - set(expected_names)) if unexpected: raise WorkflowError( f"{description} contains unexpected Job(s): " + ",".join(unexpected) ) return actual def manifest_with_missing_jobs( manifest: str, expected_names: Sequence[str], missing_names: Set[str], description: str, ) -> str: payload = parse_json(manifest, f"{description} manifest") if ( not isinstance(payload, dict) or payload.get("kind") != "List" or not isinstance(payload.get("items"), list) ): raise WorkflowError( f"{description} manifest is not a Kubernetes List" ) manifest_job_names = [] selected_items = [] for item in payload["items"]: if not isinstance(item, dict): raise WorkflowError( f"{description} manifest contains a non-object item" ) if item.get("kind") != "Job": selected_items.append(item) continue metadata = item.get("metadata") name = metadata.get("name") if isinstance(metadata, dict) else None if not isinstance(name, str) or not name: raise WorkflowError( f"{description} manifest contains a Job without a name" ) manifest_job_names.append(name) if name in missing_names: selected_items.append(item) if set(manifest_job_names) != set(expected_names): raise WorkflowError( f"{description} manifest Job set does not match its expected names" ) if not missing_names: raise WorkflowError( f"{description} missing-Job manifest requested with no missing Jobs" ) return canonical_json({**payload, "items": selected_items}) def expected_pack_job_names( pack_instance: str, shard_count: int, ) -> List[str]: return [ f"{pack_instance}-shard-{index}-of-{shard_count}" for index in range(shard_count) ] def expected_export_job_names( export_instance: str, task_count: int, ) -> List[str]: return [ f"{export_instance}-task-{index}-of-{task_count}" for index in range(task_count) ] def expected_watermark_job_names(watermark_instance: str) -> List[str]: return [watermark_instance] def wait_for_jobs( kubectl: str, instance: str, expected_names: Sequence[str], description: str, ) -> None: expected_count = len(expected_names) last_complete_count = -1 next_heartbeat = time.monotonic() while True: jobs = list_instance_jobs( kubectl, instance, description, ) validate_expected_job_set(jobs, expected_names, description) failed = [ job for job in jobs if condition_is_true(job, ("Failed", "FailureTarget")) ] if failed: raise WorkflowError( f"{description} Job failure: {failed_job_details(failed)}" ) complete_count = sum( condition_is_true(job, ("Complete",)) for job in jobs ) if complete_count == expected_count: log(f"all {expected_count} {description} Job(s) are complete") return now = time.monotonic() if complete_count != last_complete_count or now >= next_heartbeat: log( f"waiting for {description} Jobs: " f"complete={complete_count}/{expected_count}" ) last_complete_count = complete_count next_heartbeat = now + JOB_HEARTBEAT_INTERVAL_SECONDS time.sleep(JOB_POLL_INTERVAL_SECONDS) def reconcile_job_phase( kubectl: str, instance: str, expected_names: Sequence[str], description: str, manifest: Optional[str], observed_jobs: Optional[Sequence[Dict[str, Any]]] = None, ) -> Tuple[bool, float]: if any(len(name) > 63 for name in expected_names): raise WorkflowError( f"{description} contains a Job name longer than 63 characters" ) jobs = ( list_instance_jobs(kubectl, instance, description) if observed_jobs is None else list(observed_jobs) ) observed_names = validate_observed_job_subset( jobs, expected_names, description, ) missing_names = set(expected_names) - observed_names created = bool(missing_names) if missing_names: if manifest is None: raise WorkflowError( f"{description} has missing Jobs but no manifest was generated" ) missing_manifest = manifest_with_missing_jobs( manifest, expected_names, missing_names, description, ) apply_manifest(kubectl, missing_manifest, description) log( f"created {len(missing_names)} missing {description} Job(s) for " f"instance {instance!r}" ) else: log( f"reusing {len(jobs)} existing {description} Job(s) for " f"instance {instance!r}" ) already_complete = ( not missing_names and bool(jobs) and not any( condition_is_true(job, ("Failed", "FailureTarget")) for job in jobs ) and all( condition_is_true(job, ("Complete",)) for job in jobs ) ) if already_complete: log(f"all {len(expected_names)} {description} Job(s) already complete") return created, 0.0 started_at = time.monotonic() wait_for_jobs( kubectl, instance, expected_names, description, ) return created, time.monotonic() - started_at def generate_pack_manifest( spec: WorkflowSpec, bash: str, workflow_name: str, pack_instance: str, backup_name: str, cluster_id_object_key_value: str, ) -> str: command = [ bash, str(GENERATE_PACK), "--name", pack_instance, "--workflow-name", workflow_name, "--shards", str(spec.pack.shards), "--image", spec.pack.cse_ctl_image, "--node-group", spec.node_group, "--cpu.request", spec.pack.resources.cpu_request, "--memory.request", spec.pack.resources.memory_request, "--cluster-id-dfs-object-key", cluster_id_object_key_value, "--output", "json", ] if spec.pack.ebs_request: command.extend(["--ebs.request", spec.pack.ebs_request]) if spec.pack.ebs_storage_class: command.extend( ["--ebs.storage-class", spec.pack.ebs_storage_class] ) command.extend(spec.pack.generator_args) command.extend( [ "--", "--backup-name", backup_name, *spec.pack.command_args, ] ) return run_command(command, "generate Pack manifest") def collect_pack_tasks( bash: str, pack_instance: str, expected_count: Optional[int], ) -> List[Dict[str, Any]]: output = run_command( [ bash, str(LIST_PACK_TASKS), "--selector", f"{INSTANCE_LABEL}={pack_instance}", ], "collect completed Pack tasks", ) tasks = parse_json(output, "completed Pack task collector") if not isinstance(tasks, list): raise WorkflowError("completed Pack task collector must return an array") if expected_count is not None and len(tasks) != expected_count: raise WorkflowError( f"completed Pack task collector returned {len(tasks)} tasks; " f"expected {expected_count}" ) if not tasks: raise WorkflowError("completed Pack task collector returned no tasks") if not all(isinstance(task, dict) for task in tasks): raise WorkflowError("completed Pack task collector returned a non-object task") return tasks def generate_export_manifest( args: argparse.Namespace, spec: WorkflowSpec, workflow_name: str, export_instance: str, tasks: Sequence[Dict[str, Any]], ) -> str: command = [ sys.executable, str(GENERATE_EXPORT), "--name", export_instance, "--workflow-name", workflow_name, "--batch-task-json=-", "--output=json", "--local-gpg", args.local_gpg, "--dumpling-image", spec.export.dumpling_image, "--cse-ctl-image", spec.export.cse_ctl_image, "--node-group", spec.node_group, "--cpu.request", spec.export.resources.cpu_request, "--memory.request", spec.export.resources.memory_request, ] if spec.export.runner_image: command.extend(["--runner-image", spec.export.runner_image]) command.extend(spec.export.generator_args) command.extend( [ "--", "--out-prefix", spec.export.out_prefix, "--flat", ] ) if spec.export.exclude_meta_name_regexp: command.extend( [ "--exclude-meta-name-regexp", spec.export.exclude_meta_name_regexp, ] ) command.extend(spec.export.worker_args) command.extend( [ "--", *spec.export.dumpling_args, ] ) task_json = json.dumps(tasks, ensure_ascii=False, separators=(",", ":")) return run_command( command, "generate Export manifest", input_text=task_json, ) def generate_watermark_manifest( spec: WorkflowSpec, workflow_name: str, watermark_instance: str, export_instance: str, tasks: Sequence[Dict[str, Any]], backup_cutoff: datetime, ) -> str: command = [ sys.executable, str(GENERATE_WATERMARK), "--name", watermark_instance, "--workflow-name", workflow_name, "--tasks-json=-", "--mysql-secret", spec.mysql.secret_name, "--mysql-secret-key", spec.mysql.secret_key, "--out-prefix", spec.export.out_prefix, "--export-instance", export_instance, "--backup-time", backup_cutoff.strftime(COMPLETE_MARKER_BACKUP_TIME_FORMAT), "--aws-region", spec.watermark.aws_region, "--runner-image", spec.watermark.image, "--node-group", spec.node_group, "--cpu.request", spec.watermark.resources.cpu_request, "--memory.request", spec.watermark.resources.memory_request, ] command.extend(spec.watermark.generator_args) command.extend(["--", *spec.watermark.worker_args]) task_json = json.dumps(tasks, ensure_ascii=False, separators=(",", ":")) return run_command( command, "generate Watermark manifest", input_text=task_json, ) def ensure_workflow_record( args: argparse.Namespace, commands: Dict[str, str], workflow_name: str, pack_instance: str, result: Dict[str, Any], ) -> WorkflowEnsureResult: result["stage"] = "workflow_reconcile" storage = load_dfs_storage_config(commands["kubectl"]) s3_client = create_s3_client(storage.region) record = load_workflow_record( commands["kubectl"], workflow_name, pack_instance, ) record_created = False cluster_ids: List[str] = [] if record is not None: cluster_id_object = record.cluster_id_object if cluster_id_object.key != cluster_id_object_key(workflow_name): raise WorkflowError( "workflow cluster ID object key is not the deterministic " "key for this RUN_ID" ) if cluster_id_object.bucket != storage.bucket: raise WorkflowError( "workflow cluster ID object bucket no longer matches " "DFS_S3_BUCKET" ) if cluster_id_object.region != storage.region: raise WorkflowError( "workflow cluster ID object region no longer matches " "DFS_S3_REGION" ) validate_cluster_id_object( s3_client, workflow_name, cluster_id_object, record.backup_name, ) backup_name = record.backup_name log( f"reusing backup {backup_name!r} and cluster ID object " f"{cluster_id_object.uri!r} from workflow ConfigMap " f"{record.name!r}" ) else: orphaned_pack_jobs = list_instance_jobs( commands["kubectl"], pack_instance, "Pack", ) if orphaned_pack_jobs: raise WorkflowError( "Pack Jobs exist without the named workflow ConfigMap" ) recovered = load_existing_cluster_id_object( s3_client, storage, workflow_name, ) if recovered is not None: cluster_id_object, backup_name = recovered record = create_workflow_record( commands["kubectl"], workflow_name, pack_instance, cluster_id_object, backup_name, ) record_created = True result["workflow_record_created"] = True log( f"recovered workflow ConfigMap {record.name!r} from " f"cluster ID object {cluster_id_object.uri!r}" ) else: log( f"discovering backup from_prefix={args.backup_from_prefix!r} " f"pod={TIKV_API_POD}" ) backup_name = discover_backup( commands["kubectl"], TIKV_API_POD, args.backup_from_prefix, ) backup_cutoff = backup_cutoff_from_name(backup_name) result["backup_name"] = backup_name result["backup_cutoff"] = ( backup_cutoff.isoformat().replace("+00:00", "Z") ) if record is None: log("connecting to MySQL") mysql_connection = open_mysql_connection(args.mysql_defaults_file) try: cluster_ids = select_cluster_ids( mysql_connection, backup_cutoff, ) finally: quietly_close_mysql(mysql_connection) if cluster_ids: cluster_id_object = upload_cluster_id_object( s3_client, storage, workflow_name, cluster_ids, backup_name, ) record = create_workflow_record( commands["kubectl"], workflow_name, pack_instance, cluster_id_object, backup_name, ) record_created = True result["workflow_record_created"] = True selected_clusters = ( record.cluster_id_object.cluster_count if record is not None else len(cluster_ids) ) result["selected_clusters"] = selected_clusters if record is not None: result["cluster_id_object_uri"] = record.cluster_id_object.uri result["cluster_id_object_sha256"] = record.cluster_id_object.sha256 result["cluster_id_object_bytes"] = record.cluster_id_object.byte_count if record is None: log("no clusters require export; no S3 object or ConfigMap created") elif record_created: log( f"created workflow ConfigMap {record.name!r} for " f"{selected_clusters} cluster(s)" ) else: log( f"loaded {selected_clusters} original cluster(s) from workflow " f"record {record.name!r}" ) return WorkflowEnsureResult( record=record, backup_name=backup_name, backup_cutoff=backup_cutoff, ) def ensure_pack_jobs( spec: WorkflowSpec, commands: Dict[str, str], workflow_name: str, pack_instance: str, workflow: WorkflowEnsureResult, result: Dict[str, Any], ) -> List[Dict[str, Any]]: result["stage"] = "pack_reconcile" if workflow.record is None: raise WorkflowError("cannot create Pack Jobs without a workflow record") record = workflow.record expected_names = expected_pack_job_names( pack_instance, spec.pack.shards, ) jobs = list_instance_jobs( commands["kubectl"], pack_instance, "Pack", ) observed_names = validate_observed_job_subset( jobs, expected_names, "Pack", ) manifest = None if set(expected_names) - observed_names: manifest = job_manifest_with_ttl( generate_pack_manifest( spec, commands["bash"], workflow_name, pack_instance, workflow.backup_name, record.cluster_id_object.key, ), "Pack", spec.ttl_seconds_after_finished, ) jobs_created, duration_seconds = reconcile_job_phase( commands["kubectl"], pack_instance, expected_names, "Pack", manifest, observed_jobs=jobs, ) result["pack_jobs"] = len(expected_names) result["pack_jobs_created"] = jobs_created result["pack_duration_seconds"] = round(duration_seconds, 3) tasks = collect_pack_tasks( commands["bash"], pack_instance, spec.pack.shards, ) result["pack_tasks"] = len(tasks) validate_pack_task_backup(tasks, workflow.backup_name) log(f"collected {len(tasks)} completed Pack task(s)") return tasks def ensure_export_jobs( args: argparse.Namespace, spec: WorkflowSpec, commands: Dict[str, str], workflow_name: str, export_instance: str, tasks: Sequence[Dict[str, Any]], result: Dict[str, Any], ) -> None: result["stage"] = "export_reconcile" expected_names = expected_export_job_names( export_instance, len(tasks), ) jobs = list_instance_jobs( commands["kubectl"], export_instance, "Export", ) observed_names = validate_observed_job_subset( jobs, expected_names, "Export", ) manifest = None if set(expected_names) - observed_names: manifest = job_manifest_with_ttl( generate_export_manifest( args, spec, workflow_name, export_instance, tasks, ), "Export", spec.ttl_seconds_after_finished, ) jobs_created, duration_seconds = reconcile_job_phase( commands["kubectl"], export_instance, expected_names, "Export", manifest, observed_jobs=jobs, ) result["export_jobs"] = len(expected_names) result["export_jobs_created"] = jobs_created result["export_duration_seconds"] = round(duration_seconds, 3) def load_watermark_job_result( kubectl: str, watermark_instance: str, export_instance: str, backup_cutoff: datetime, ) -> Dict[str, Any]: selector = f"{INSTANCE_LABEL}={watermark_instance}" output = run_command( [ kubectl, "--request-timeout=30s", "-n", NAMESPACE, "get", "pods", "-l", selector, "-o", "json", ], "read Watermark Pod status", ) pod_list = parse_json(output, "Watermark Pod List") if ( not isinstance(pod_list, dict) or not isinstance(pod_list.get("items"), list) ): raise WorkflowError( "Watermark Pod query returned an invalid Kubernetes List" ) pods = pod_list["items"] if len(pods) != 1 or not isinstance(pods[0], dict): raise WorkflowError( f"Watermark Job expected one Pod; found {len(pods)}" ) status = pods[0].get("status") container_statuses = ( status.get("containerStatuses") if isinstance(status, dict) else None ) if not isinstance(container_statuses, list): raise WorkflowError("Watermark Pod has no container statuses") matching_statuses = [ item for item in container_statuses if isinstance(item, dict) and item.get("name") == WATERMARK_CONTAINER ] if len(matching_statuses) != 1: raise WorkflowError( f"Watermark Pod expected one {WATERMARK_CONTAINER!r} " f"container status; found {len(matching_statuses)}" ) state = matching_statuses[0].get("state") terminated = state.get("terminated") if isinstance(state, dict) else None if not isinstance(terminated, dict): raise WorkflowError("Watermark container has not terminated") message = terminated.get("message") payload_text = message.strip() if isinstance(message, str) else "" if not payload_text: raise WorkflowError("Watermark container produced no termination message") payload = parse_json(payload_text, "Watermark Job result") if not isinstance(payload, dict): raise WorkflowError("Watermark Job result is not an object") expected_values = { "watermark_instance": watermark_instance, "export_instance": export_instance, "backup_time": backup_cutoff.strftime( COMPLETE_MARKER_BACKUP_TIME_FORMAT ), } for key, expected in expected_values.items(): if payload.get(key) != expected: raise WorkflowError( f"Watermark Job result has unexpected {key!r}" ) if payload.get("status") not in ("succeeded", "failed"): raise WorkflowError("Watermark Job result has an invalid status") count_fields = ( "packed_clusters", "packed_meta_objects", "verified_complete_markers", "unverified_complete_marker_clusters", "mysql_updated_clusters", "newer_complete_marker_clusters", "newer_mysql_watermark_clusters", "mysql_update_failed_clusters", ) for field in count_fields: value = payload.get(field) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise WorkflowError( f"Watermark Job result has an invalid {field!r}" ) if payload["status"] == "failed" and not isinstance( payload.get("error"), str, ): raise WorkflowError("failed Watermark Job result has no error") return payload def copy_watermark_job_result( payload: Dict[str, Any], result: Dict[str, Any], ) -> None: for field in ( "packed_clusters", "packed_meta_objects", "verified_complete_markers", "unverified_complete_marker_clusters", "mysql_updated_clusters", "newer_complete_marker_clusters", "newer_mysql_watermark_clusters", "mysql_update_failed_clusters", ): result[field] = payload[field] result["watermark_worker_status"] = payload["status"] if payload["status"] == "failed": result["watermark_worker_error"] = payload["error"] def ensure_watermark_job( spec: WorkflowSpec, commands: Dict[str, str], workflow_name: str, watermark_instance: str, export_instance: str, tasks: Sequence[Dict[str, Any]], backup_cutoff: datetime, result: Dict[str, Any], ) -> None: result["stage"] = "watermark_reconcile" expected_names = expected_watermark_job_names(watermark_instance) jobs = list_instance_jobs( commands["kubectl"], watermark_instance, "Watermark", ) observed_names = validate_observed_job_subset( jobs, expected_names, "Watermark", ) missing_names = set(expected_names) - observed_names result["watermark_jobs"] = len(expected_names) result["watermark_jobs_created"] = bool(missing_names) manifest = None if missing_names: manifest = job_manifest_with_ttl( generate_watermark_manifest( spec, workflow_name, watermark_instance, export_instance, tasks, backup_cutoff, ), "Watermark", spec.ttl_seconds_after_finished, ) phase_error = None phase_started_at = time.monotonic() try: jobs_created, duration_seconds = reconcile_job_phase( commands["kubectl"], watermark_instance, expected_names, "Watermark", manifest, observed_jobs=jobs, ) result["watermark_jobs_created"] = jobs_created result["watermark_duration_seconds"] = round(duration_seconds, 3) except WorkflowError as exc: phase_error = exc result["watermark_duration_seconds"] = round( time.monotonic() - phase_started_at, 3, ) try: payload = load_watermark_job_result( commands["kubectl"], watermark_instance, export_instance, backup_cutoff, ) except WorkflowError as exc: if phase_error is not None: log(f"warning: cannot read Watermark Job result: {exc}") raise phase_error raise copy_watermark_job_result(payload, result) if phase_error is not None: worker_error = payload.get("error") if isinstance(worker_error, str): raise WorkflowError( f"{phase_error}; Watermark worker: {worker_error}" ) from phase_error raise phase_error if payload["status"] != "succeeded": raise WorkflowError( f"Watermark worker failed: {payload['error']}" ) def emit_result(result: Dict[str, Any]) -> None: json.dump(result, sys.stdout, ensure_ascii=False, sort_keys=True) print() sys.stdout.flush() def _record_notification(result: Dict[str, Any], webhook_url: str) -> None: notification_error = send_notification(webhook_url, result) if notification_error: log(f"warning: {notification_error}") result["notification_error"] = notification_error def finish_success( result: Dict[str, Any], webhook_url: str, ) -> int: result["status"] = "succeeded" result["stage"] = "complete" _record_notification(result, webhook_url) emit_result(result) return 0 def main(argv: Optional[Sequence[str]] = None) -> int: parser = build_parser() args = parser.parse_args(argv) try: spec = load_workflow_spec(args.spec) except SpecError as exc: parser.error(str(exc)) webhook_url = os.environ.get("LARK_BOT_WEBHOOK", "").strip() kubernetes_context = ( os.environ.get("KUBERNETES_CONTEXT_NAME", "").strip() or "unknown" ) run_id = args.name pack_instance = f"pack-backup-offline-{run_id}" export_instance = f"dumpling-packed-export-{run_id}" watermark_instance = f"packed-backup-watermark-{run_id}" workflow_config_map = f"{pack_instance}-cluster-ids" watermark_input_config_map = f"{watermark_instance}-input" result: Dict[str, Any] = { "status": "running", "run_id": run_id, "backup_from_prefix": args.backup_from_prefix, "pack_instance": pack_instance, "export_instance": export_instance, "watermark_instance": watermark_instance, "workflow_config_map": workflow_config_map, "watermark_input_config_map": watermark_input_config_map, "cluster_id_object_uri": None, "cluster_id_object_sha256": None, "cluster_id_object_bytes": 0, "out_prefix": spec.export.out_prefix, "kubernetes_context": kubernetes_context, "workflow_record_created": False, "selected_clusters": 0, "pack_jobs": 0, "pack_jobs_created": False, "pack_tasks": 0, "pack_duration_seconds": 0.0, "export_jobs": 0, "export_jobs_created": False, "export_duration_seconds": 0.0, "watermark_jobs": 0, "watermark_jobs_created": False, "watermark_duration_seconds": 0.0, "watermark_worker_status": "not_started", "packed_clusters": 0, "packed_meta_objects": 0, "verified_complete_markers": 0, "unverified_complete_marker_clusters": 0, "mysql_updated_clusters": 0, "newer_complete_marker_clusters": 0, "newer_mysql_watermark_clusters": 0, "mysql_update_failed_clusters": 0, "stage": "initializing", } try: if any( len(resource_name) > 63 for resource_name in ( pack_instance, export_instance, watermark_instance, workflow_config_map, watermark_input_config_map, ) ): raise WorkflowError( "--name is too long after adding workflow resource prefixes" ) commands = require_runtime(args) workflow = ensure_workflow_record( args, commands, run_id, pack_instance, result, ) if workflow.record is None: log("no clusters require export; no Kubernetes resources created") return finish_success(result, webhook_url) tasks = ensure_pack_jobs( spec, commands, run_id, pack_instance, workflow, result, ) ensure_export_jobs( args, spec, commands, run_id, export_instance, tasks, result, ) ensure_watermark_job( spec, commands, run_id, watermark_instance, export_instance, tasks, workflow.backup_cutoff, result, ) log("all Export and Watermark Jobs completed") return finish_success(result, webhook_url) except WorkflowError as exc: result["status"] = "failed" result["error"] = str(exc) log(f"error: {exc}") _record_notification(result, webhook_url) emit_result(result) return 1 except KeyboardInterrupt: result["status"] = "failed" result["error"] = "interrupted" log("error: interrupted") _record_notification(result, webhook_url) emit_result(result) return 130 if __name__ == "__main__": sys.exit(main())