#!/usr/bin/env python3 """Generate a self-contained packed-backup CronJob manifest offline.""" import argparse import base64 import hashlib import json import os import re import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Mapping, NoReturn, Optional, Sequence, Tuple from urllib.parse import urlsplit from zoneinfo import ZoneInfo, ZoneInfoNotFoundError import yaml from cronjob_generator.input_inspection import ( InputInspectionError, MysqlConnectionInfo, ParsedGpgPublicKey, parse_gpg_public_key, parse_mysql_defaults, ) from packed_backup_workflow_spec import ( MAX_SPEC_BYTES, SpecError, UniqueKeySafeLoader, WorkflowSpec, parse_workflow_spec, ) CONFIG_SCHEMA_VERSION = 1 MAX_CONFIG_BYTES = 128 * 1024 MAX_GPG_BYTES = 256 * 1024 MAX_MYSQL_BYTES = 256 * 1024 MAX_WEBHOOK_BYTES = 8 * 1024 KUBERNETES_DNS_LABEL_MAX_LENGTH = 63 SCHEDULED_EXECUTION_ID_LENGTH = 10 RUN_ID_HASH_LENGTH = 8 APP_NAME = "on-demand-export-orchestrator" WORKFLOW_NAMESPACE = "tidb-serverless" GPG_CONFIG_KEY = "pubkey.asc" LARK_SECRET_KEY = "webhook" MYSQL_SECRET_KEY = "mysql.cnf" WORKFLOW_CONFIG_KEY = "workflow.yaml" NODE_GROUP_LABEL = "serverless.tidbcloud.com/node" DEFAULT_TIME_ZONE = "Etc/UTC" DEFAULT_STARTING_DEADLINE_SECONDS = 3600 DEFAULT_TTL_SECONDS_AFTER_FINISHED = 30 * 24 * 60 * 60 DEFAULT_SUCCESSFUL_JOBS_HISTORY_LIMIT = 3 DEFAULT_FAILED_JOBS_HISTORY_LIMIT = 5 DEFAULT_SUSPEND = False DEFAULT_KUBERNETES_CONTEXT_NAME = "unknown" DEFAULT_IMAGE_PULL_POLICY = "Always" DEFAULT_NODE_GROUP = "general" ANSI_BOLD = "\x1b[1m" ANSI_DIM = "\x1b[2m" ANSI_RESET = "\x1b[0m" class GeneratorError(ValueError): """The generator input is invalid or the manifest cannot be written.""" class Parser(argparse.ArgumentParser): def error(self, message: str) -> NoReturn: raise GeneratorError(message) class LiteralString(str): """A string that should use YAML literal block style.""" class ManifestDumper(yaml.SafeDumper): pass def _represent_literal( dumper: ManifestDumper, value: LiteralString, ) -> yaml.ScalarNode: return dumper.represent_scalar("tag:yaml.org,2002:str", value, style="|") ManifestDumper.add_representer(LiteralString, _represent_literal) @dataclass(frozen=True) class ControllerResources: requests: Dict[str, str] limits: Dict[str, str] def as_dict(self) -> Dict[str, Dict[str, str]]: return { "requests": dict(self.requests), "limits": dict(self.limits), } @dataclass(frozen=True) class CronJobSettings: name: str namespace: str schedule: str time_zone: str starting_deadline_seconds: int ttl_seconds_after_finished: int successful_jobs_history_limit: int failed_jobs_history_limit: int suspend: bool kubernetes_context_name: str service_account_name: str controller_image: str image_pull_policy: str node_group: str resources: Optional[ControllerResources] @dataclass(frozen=True) class GeneratorConfig: cron_job: CronJobSettings workflow: WorkflowSpec def _mapping(value: Any, path: str) -> Mapping[str, Any]: if not isinstance(value, dict): raise GeneratorError(f"{path} must be a mapping") if not all(isinstance(key, str) for key in value): raise GeneratorError(f"{path} keys must be strings") return value def _keys( value: Mapping[str, Any], path: str, required: Sequence[str], optional: Sequence[str] = (), ) -> None: required_set = set(required) allowed = required_set | set(optional) missing = sorted(required_set - set(value)) unknown = sorted(set(value) - allowed) if missing: raise GeneratorError( f"{path} is missing required field(s): {', '.join(missing)}" ) if unknown: raise GeneratorError( f"{path} contains unknown field(s): {', '.join(unknown)}" ) def _string(value: Any, path: str) -> str: if not isinstance(value, str) or not value: raise GeneratorError(f"{path} must be a non-empty string") if value != value.strip() or any(character in value for character in "\r\n\0"): raise GeneratorError( f"{path} must not contain leading, trailing, or control whitespace" ) return value def _integer(value: Any, path: str, minimum: int = 0) -> int: if isinstance(value, bool) or not isinstance(value, int) or value < minimum: qualifier = "positive" if minimum == 1 else "non-negative" raise GeneratorError(f"{path} must be a {qualifier} integer") return value def _boolean(value: Any, path: str) -> bool: if not isinstance(value, bool): raise GeneratorError(f"{path} must be a boolean") return value def _kubernetes_name(value: Any, path: str, maximum: int = 63) -> str: parsed = _string(value, path) if ( len(parsed) > maximum or re.fullmatch(r"[a-z0-9](?:[-a-z0-9]*[a-z0-9])?", parsed) is None ): raise GeneratorError( f"{path} must be a Kubernetes DNS label of at most {maximum} characters" ) return parsed def _label_value(value: Any, path: str) -> str: parsed = _string(value, path) if ( len(parsed) > 63 or re.fullmatch( r"[A-Za-z0-9](?:[-A-Za-z0-9_.]*[A-Za-z0-9])?", parsed, ) is None ): raise GeneratorError(f"{path} must be a Kubernetes label value") return parsed def _image(value: Any, path: str) -> str: parsed = _string(value, path) if any(character.isspace() for character in parsed): raise GeneratorError(f"{path} must not contain whitespace") return parsed def _schedule(value: Any, path: str) -> str: parsed = _string(value, path) if "TZ=" in parsed or "CRON_TZ=" in parsed: raise GeneratorError(f"{path} must use cronJob.timeZone instead of TZ syntax") if len(parsed.split()) != 5: raise GeneratorError(f"{path} must contain exactly five cron fields") return parsed def _time_zone(value: Any, path: str) -> str: parsed = _string(value, path) try: ZoneInfo(parsed) except (ZoneInfoNotFoundError, ValueError) as exc: raise GeneratorError(f"{path} must be an available IANA time zone") from exc return parsed def _resource_quantity(value: Any, path: str) -> str: parsed = _string(value, path) if any(character.isspace() for character in parsed): raise GeneratorError(f"{path} must not contain whitespace") return parsed def _resources(value: Any, path: str) -> ControllerResources: payload = _mapping(value, path) _keys(payload, path, ("requests", "limits")) result: Dict[str, Dict[str, str]] = {} for group in ("requests", "limits"): group_path = f"{path}.{group}" group_payload = _mapping(payload[group], group_path) _keys(group_payload, group_path, ("cpu", "memory")) result[group] = { "cpu": _resource_quantity(group_payload["cpu"], f"{group_path}.cpu"), "memory": _resource_quantity( group_payload["memory"], f"{group_path}.memory", ), } return ControllerResources( requests=result["requests"], limits=result["limits"], ) def _parse_cron_job(value: Any) -> CronJobSettings: payload = _mapping(value, "cronJob") required = ( "name", "schedule", "serviceAccountName", "controllerImage", ) optional = ( "namespace", "timeZone", "startingDeadlineSeconds", "ttlSecondsAfterFinished", "successfulJobsHistoryLimit", "failedJobsHistoryLimit", "suspend", "kubernetesContextName", "imagePullPolicy", "nodeGroup", "resources", ) _keys(payload, "cronJob", required, optional) image_pull_policy = _string( payload.get("imagePullPolicy", DEFAULT_IMAGE_PULL_POLICY), "cronJob.imagePullPolicy", ) if image_pull_policy not in ("Always", "IfNotPresent", "Never"): raise GeneratorError( "cronJob.imagePullPolicy must be Always, IfNotPresent, or Never" ) namespace = _kubernetes_name( payload.get("namespace", WORKFLOW_NAMESPACE), "cronJob.namespace", ) if namespace != WORKFLOW_NAMESPACE: raise GeneratorError( f"cronJob.namespace must be {WORKFLOW_NAMESPACE!r}; " "the workflow controller and stage generators use that namespace" ) return CronJobSettings( name=_kubernetes_name(payload["name"], "cronJob.name", maximum=52), namespace=namespace, schedule=_schedule(payload["schedule"], "cronJob.schedule"), time_zone=_time_zone( payload.get("timeZone", DEFAULT_TIME_ZONE), "cronJob.timeZone", ), starting_deadline_seconds=_integer( payload.get( "startingDeadlineSeconds", DEFAULT_STARTING_DEADLINE_SECONDS, ), "cronJob.startingDeadlineSeconds", minimum=1, ), ttl_seconds_after_finished=_integer( payload.get( "ttlSecondsAfterFinished", DEFAULT_TTL_SECONDS_AFTER_FINISHED, ), "cronJob.ttlSecondsAfterFinished", ), successful_jobs_history_limit=_integer( payload.get( "successfulJobsHistoryLimit", DEFAULT_SUCCESSFUL_JOBS_HISTORY_LIMIT, ), "cronJob.successfulJobsHistoryLimit", ), failed_jobs_history_limit=_integer( payload.get( "failedJobsHistoryLimit", DEFAULT_FAILED_JOBS_HISTORY_LIMIT, ), "cronJob.failedJobsHistoryLimit", ), suspend=_boolean( payload.get("suspend", DEFAULT_SUSPEND), "cronJob.suspend", ), kubernetes_context_name=_string( payload.get( "kubernetesContextName", DEFAULT_KUBERNETES_CONTEXT_NAME, ), "cronJob.kubernetesContextName", ), service_account_name=_kubernetes_name( payload["serviceAccountName"], "cronJob.serviceAccountName", ), controller_image=_image( payload["controllerImage"], "cronJob.controllerImage", ), image_pull_policy=image_pull_policy, node_group=_label_value( payload.get("nodeGroup", DEFAULT_NODE_GROUP), "cronJob.nodeGroup", ), resources=( _resources(payload["resources"], "cronJob.resources") if payload.get("resources") is not None else None ), ) def _derived_workflow_payload( value: Any, cron_job: CronJobSettings, ) -> Dict[str, Any]: payload = _mapping(value, "workflow") _keys( payload, "workflow", ("schemaVersion", "pack", "export", "watermark"), ) stages: Dict[str, Dict[str, Any]] = {} for stage in ("pack", "export", "watermark"): stage_payload = _mapping(payload[stage], f"workflow.{stage}") if "nodeGroup" in stage_payload: raise GeneratorError( f"workflow.{stage}.nodeGroup is derived from cronJob.nodeGroup" ) stages[stage] = dict(stage_payload) return { "schemaVersion": payload["schemaVersion"], "nodeGroup": cron_job.node_group, "ttlSecondsAfterFinished": cron_job.ttl_seconds_after_finished, "mysql": { "secretName": f"{cron_job.name}-mysql", "secretKey": MYSQL_SECRET_KEY, }, **stages, } def parse_generator_config(payload: Any) -> GeneratorConfig: root = _mapping(payload, "generator config") _keys(root, "generator config", ("schemaVersion", "cronJob", "workflow")) version = root["schemaVersion"] if isinstance(version, bool) or version != CONFIG_SCHEMA_VERSION: raise GeneratorError( f"schemaVersion must be the integer {CONFIG_SCHEMA_VERSION}" ) cron_job = _parse_cron_job(root["cronJob"]) try: workflow = parse_workflow_spec( _derived_workflow_payload(root["workflow"], cron_job) ) except SpecError as exc: raise GeneratorError(str(exc)) from exc normalized_workflow = yaml.safe_dump( workflow.as_dict(), sort_keys=False, allow_unicode=True, ).encode("utf-8") if len(normalized_workflow) > MAX_SPEC_BYTES: raise GeneratorError(f"normalized workflow exceeds {MAX_SPEC_BYTES} bytes") config = GeneratorConfig( cron_job=cron_job, workflow=workflow, ) validate_no_placeholders(config) validate_generated_names(config) return config def validate_no_placeholders(config: GeneratorConfig) -> None: values = { "cronJob.controllerImage": config.cron_job.controller_image, "workflow.pack.cseCtlImage": config.workflow.pack.cse_ctl_image, "workflow.export.outPrefix": config.workflow.export.out_prefix, "workflow.export.cseCtlImage": config.workflow.export.cse_ctl_image, "workflow.export.dumplingImage": config.workflow.export.dumpling_image, "workflow.watermark.image": config.workflow.watermark.image, } if config.workflow.export.runner_image is not None: values["workflow.export.runnerImage"] = config.workflow.export.runner_image for path, value in values.items(): if "<" in value or ">" in value: raise GeneratorError(f"{path} still contains an example placeholder") def load_generator_config(path: str) -> GeneratorConfig: config_path = Path(path) try: raw = config_path.read_bytes() except OSError as exc: raise GeneratorError( f"cannot read generator config {str(config_path)!r} " f"({type(exc).__name__})" ) from exc if not raw: raise GeneratorError("generator config is empty") if len(raw) > MAX_CONFIG_BYTES: raise GeneratorError(f"generator config exceeds {MAX_CONFIG_BYTES} bytes") try: text = raw.decode("utf-8") except UnicodeDecodeError as exc: raise GeneratorError("generator config is not valid UTF-8") from exc try: payload = yaml.load(text, Loader=UniqueKeySafeLoader) except yaml.YAMLError as exc: raise GeneratorError(f"generator config contains invalid YAML: {exc}") from exc return parse_generator_config(payload) def resource_names_for_run_id(run_id: str, shards: int) -> Tuple[str, ...]: pack = f"pack-backup-offline-{run_id}" export = f"dumpling-packed-export-{run_id}" watermark = f"packed-backup-watermark-{run_id}" return ( f"{pack}-cluster-ids", f"{pack}-shard-{shards - 1}-of-{shards}", f"{export}-init", f"{export}-task-{shards - 1}-of-{shards}", watermark, f"{watermark}-input", ) def execution_id_length_budget(cron_job_name: str, shards: int) -> int: empty_execution_run_id = "cron--" + "x" * RUN_ID_HASH_LENGTH controller_job_budget = ( KUBERNETES_DNS_LABEL_MAX_LENGTH - len(cron_job_name) - 1 ) downstream_budgets = ( KUBERNETES_DNS_LABEL_MAX_LENGTH - len(name) for name in resource_names_for_run_id(empty_execution_run_id, shards) ) return min(controller_job_budget, *downstream_budgets) def validate_generated_names(config: GeneratorConfig) -> None: cron_job = config.cron_job lark_secret_name = f"{cron_job.name}-lark" fixed_names = ( config.workflow.mysql.secret_name, f"{cron_job.name}-workflow-spec", f"{cron_job.name}-gpg-public-key", lark_secret_name, ) too_long = [name for name in fixed_names if len(name) > 63] if too_long: raise GeneratorError( f"cronJob.name makes generated resource name {too_long[0]!r} exceed 63 characters" ) maximum_execution_id_length = execution_id_length_budget( cron_job.name, config.workflow.pack.shards, ) if maximum_execution_id_length < SCHEDULED_EXECUTION_ID_LENGTH: raise GeneratorError( "cronJob.name and workflow.pack.shards leave room for only " f"{maximum_execution_id_length} execution ID characters; " f"scheduled execution IDs require {SCHEDULED_EXECUTION_ID_LENGTH}" ) longest_controller_job_name = ( f"{cron_job.name}-" + "x" * maximum_execution_id_length ) if len(longest_controller_job_name) > KUBERNETES_DNS_LABEL_MAX_LENGTH: raise GeneratorError( f"maximum Controller Job name {longest_controller_job_name!r} " "exceeds 63 characters" ) longest_run_id = ( "cron-" + "x" * maximum_execution_id_length + "-" + "x" * RUN_ID_HASH_LENGTH ) for name in resource_names_for_run_id( longest_run_id, config.workflow.pack.shards, ): if len(name) > 63: raise GeneratorError( "workflow.pack.shards and the maximum CronJob RUN_ID make " f"downstream resource name {name!r} exceed 63 characters" ) def _read_regular_file(path: str, description: str, maximum: int) -> bytes: source = Path(path) try: stat_result = source.stat() except OSError as exc: raise GeneratorError( f"cannot stat {description} file {str(source)!r} ({type(exc).__name__})" ) from exc if not source.is_file(): raise GeneratorError(f"{description} is not a regular file: {str(source)!r}") if stat_result.st_size == 0: raise GeneratorError(f"{description} file is empty") if stat_result.st_size > maximum: raise GeneratorError(f"{description} file exceeds {maximum} bytes") try: payload = source.read_bytes() except OSError as exc: raise GeneratorError( f"cannot read {description} file {str(source)!r} ({type(exc).__name__})" ) from exc if not payload: raise GeneratorError(f"{description} file is empty") if len(payload) > maximum: raise GeneratorError(f"{description} file exceeds {maximum} bytes") return payload def inspect_gpg_public_key(payload: bytes) -> ParsedGpgPublicKey: try: return parse_gpg_public_key(payload) except InputInspectionError as exc: raise GeneratorError(str(exc)) from exc def validate_gpg_public_key(payload: bytes) -> str: return inspect_gpg_public_key(payload).armored def inspect_mysql_defaults(payload: bytes) -> MysqlConnectionInfo: try: return parse_mysql_defaults(payload) except InputInspectionError as exc: raise GeneratorError(str(exc)) from exc def validate_mysql_defaults(payload: bytes) -> None: inspect_mysql_defaults(payload) def validate_lark_webhook(value: str) -> str: webhook = value.strip() if not webhook or any(character in webhook for character in "\r\n\0"): raise GeneratorError("Lark webhook must be one non-empty line") if len(webhook.encode("utf-8")) > MAX_WEBHOOK_BYTES: raise GeneratorError(f"Lark webhook exceeds {MAX_WEBHOOK_BYTES} bytes") try: parsed = urlsplit(webhook) hostname = parsed.hostname parsed.port except ValueError as exc: raise GeneratorError("Lark webhook must be a valid HTTPS URL") from exc if ( parsed.scheme != "https" or not hostname or parsed.username is not None or parsed.password is not None or not parsed.path.startswith("/") or parsed.path == "/" or parsed.fragment or any(character.isspace() for character in webhook) ): raise GeneratorError("Lark webhook must be a valid HTTPS URL") return webhook def read_lark_webhook(file_path: Optional[str], environment_name: Optional[str]) -> str: if file_path is not None: payload = _read_regular_file( file_path, "Lark webhook", MAX_WEBHOOK_BYTES, ) try: value = payload.decode("utf-8") except UnicodeDecodeError as exc: raise GeneratorError("Lark webhook file must be valid UTF-8") from exc return validate_lark_webhook(value) if environment_name is None: raise GeneratorError("one explicit Lark webhook source is required") if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", environment_name) is None: raise GeneratorError("--lark-webhook-env must name a valid environment variable") value = os.environ.get(environment_name) if value is None: raise GeneratorError( f"Lark webhook environment variable {environment_name!r} is not set" ) return validate_lark_webhook(value) def _labels(cron_job_name: str) -> Dict[str, str]: return { "app.kubernetes.io/name": APP_NAME, "app.kubernetes.io/instance": cron_job_name, } def _security_context() -> Dict[str, Any]: return { "allowPrivilegeEscalation": False, "capabilities": {"drop": ["ALL"]}, "readOnlyRootFilesystem": True, "seccompProfile": {"type": "RuntimeDefault"}, } RUN_ID_PROGRAM = '''\ import hashlib import re import sys if len(sys.argv) != 4: raise SystemExit( "expected Controller Job name, CronJob name, and execution ID length budget" ) job_name = sys.argv[1] cron_job_name = sys.argv[2] try: maximum_execution_id_length = int(sys.argv[3]) except ValueError: raise SystemExit("execution ID length budget must be an integer") if maximum_execution_id_length < 1: raise SystemExit("execution ID length budget must be positive") dns_label = r"[a-z0-9](?:[-a-z0-9]*[a-z0-9])?" if len(job_name) > 63 or re.fullmatch(dns_label, job_name) is None: raise SystemExit( "controller Job name must be a Kubernetes DNS label of at most 63 characters" ) prefix = f"{cron_job_name}-" if not job_name.startswith(prefix): raise SystemExit( "controller Job name must start with the configured CronJob name followed by '-'" ) execution_id = job_name[len(prefix):] if ( len(execution_id) > maximum_execution_id_length or re.fullmatch(dns_label, execution_id) is None ): raise SystemExit( "execution ID stripped from the Controller Job name must be a Kubernetes " f"DNS label of 1-{maximum_execution_id_length} characters" ) digest = hashlib.sha256(job_name.encode("utf-8")).hexdigest()[:8] print(f"cron-{execution_id}-{digest}") ''' def derive_run_id( controller_job_name: str, cron_job_name: str, maximum_execution_id_length: int, ) -> str: if ( len(controller_job_name) > KUBERNETES_DNS_LABEL_MAX_LENGTH or re.fullmatch( r"[a-z0-9](?:[-a-z0-9]*[a-z0-9])?", controller_job_name, ) is None ): raise GeneratorError( "controller Job name must be a Kubernetes DNS label of at most 63 characters" ) if maximum_execution_id_length < 1: raise GeneratorError("execution ID length budget must be positive") prefix = f"{cron_job_name}-" if not controller_job_name.startswith(prefix): raise GeneratorError( "controller Job name must start with the configured CronJob name followed by '-'" ) execution_id = controller_job_name[len(prefix):] if ( len(execution_id) > maximum_execution_id_length or re.fullmatch( r"[a-z0-9](?:[-a-z0-9]*[a-z0-9])?", execution_id, ) is None ): raise GeneratorError( "execution ID stripped from the Controller Job name must be a Kubernetes " f"DNS label of 1-{maximum_execution_id_length} characters" ) digest = hashlib.sha256(controller_job_name.encode("utf-8")).hexdigest()[ :RUN_ID_HASH_LENGTH ] return f"cron-{execution_id}-{digest}" def controller_script() -> str: return ( 'RUN_ID="$(\n' ' python3 - "${CONTROLLER_JOB_NAME}" "${CRONJOB_NAME}" ' '"${MAX_EXECUTION_ID_LENGTH}" <<\'PY\'\n' + RUN_ID_PROGRAM + 'PY\n' + ')"\n' + 'echo "controller identity: job=${CONTROLLER_JOB_NAME} run_id=${RUN_ID}"\n' + 'exec python3 /app/run_packed_backup_export.py \\\n' + ' --name "${RUN_ID}" \\\n' + ' --spec /run/workflow/workflow.yaml \\\n' + ' "$@"\n' ) def build_manifest( config: GeneratorConfig, gpg_public_key: str, mysql_defaults: bytes, lark_webhook: str, ) -> Tuple[Dict[str, Any], ...]: cron_job = config.cron_job workflow = config.workflow labels = _labels(cron_job.name) workflow_config_map = f"{cron_job.name}-workflow-spec" gpg_config_map = f"{cron_job.name}-gpg-public-key" lark_secret = f"{cron_job.name}-lark" maximum_execution_id_length = execution_id_length_budget( cron_job.name, workflow.pack.shards, ) workflow_yaml = yaml.safe_dump( workflow.as_dict(), sort_keys=False, allow_unicode=True, ) mysql_secret_document = { "apiVersion": "v1", "kind": "Secret", "metadata": { "name": workflow.mysql.secret_name, "namespace": cron_job.namespace, "labels": dict(labels), }, "type": "Opaque", "data": { workflow.mysql.secret_key: base64.b64encode(mysql_defaults).decode("ascii") }, } lark_secret_document = { "apiVersion": "v1", "kind": "Secret", "metadata": { "name": lark_secret, "namespace": cron_job.namespace, "labels": dict(labels), }, "type": "Opaque", "data": { LARK_SECRET_KEY: base64.b64encode(lark_webhook.encode("utf-8")).decode( "ascii" ) }, } gpg_config_map_document = { "apiVersion": "v1", "kind": "ConfigMap", "metadata": { "name": gpg_config_map, "namespace": cron_job.namespace, "labels": dict(labels), }, "data": {GPG_CONFIG_KEY: LiteralString(gpg_public_key)}, } workflow_config_map_document = { "apiVersion": "v1", "kind": "ConfigMap", "metadata": { "name": workflow_config_map, "namespace": cron_job.namespace, "labels": dict(labels), }, "data": {WORKFLOW_CONFIG_KEY: LiteralString(workflow_yaml)}, } pod_labels = dict(labels) cron_job_document = { "apiVersion": "batch/v1", "kind": "CronJob", "metadata": { "name": cron_job.name, "namespace": cron_job.namespace, "labels": dict(labels), }, "spec": { "schedule": cron_job.schedule, "timeZone": cron_job.time_zone, "concurrencyPolicy": "Forbid", "startingDeadlineSeconds": cron_job.starting_deadline_seconds, "successfulJobsHistoryLimit": cron_job.successful_jobs_history_limit, "failedJobsHistoryLimit": cron_job.failed_jobs_history_limit, "suspend": cron_job.suspend, "jobTemplate": { "metadata": {"labels": dict(labels)}, "spec": { "backoffLimit": 0, "ttlSecondsAfterFinished": ( cron_job.ttl_seconds_after_finished ), "completions": 1, "parallelism": 1, "podReplacementPolicy": "TerminatingOrFailed", "template": { "metadata": {"labels": dict(pod_labels)}, "spec": { "serviceAccountName": cron_job.service_account_name, "automountServiceAccountToken": True, "restartPolicy": "Never", "nodeSelector": { "kubernetes.io/arch": "arm64", NODE_GROUP_LABEL: cron_job.node_group, }, "tolerations": [ { "key": NODE_GROUP_LABEL, "operator": "Equal", "value": cron_job.node_group, "effect": "NoSchedule", } ], "containers": [ { "name": "orchestrator", "image": cron_job.controller_image, "imagePullPolicy": cron_job.image_pull_policy, "command": ["bash", "-ec"], "args": [ LiteralString(controller_script()), "controller", "--mysql-defaults-file", "/run/mysql/mysql.cnf", "--local-gpg", "/run/gpg/pubkey.asc", ], "env": [ { "name": "CONTROLLER_JOB_NAME", "valueFrom": { "fieldRef": { "fieldPath": ( "metadata.labels['batch.kubernetes.io/job-name']" ) } }, }, { "name": "CRONJOB_NAME", "value": cron_job.name, }, { "name": "MAX_EXECUTION_ID_LENGTH", "value": str(maximum_execution_id_length), }, { "name": "KUBERNETES_CONTEXT_NAME", "value": cron_job.kubernetes_context_name, }, { "name": "LARK_BOT_WEBHOOK", "valueFrom": { "secretKeyRef": { "name": lark_secret, "key": LARK_SECRET_KEY, } }, }, { "name": "AWS_REGION", "value": workflow.watermark.aws_region, }, { "name": "AWS_DEFAULT_REGION", "value": workflow.watermark.aws_region, }, ], **( {"resources": cron_job.resources.as_dict()} if cron_job.resources is not None else {} ), "securityContext": _security_context(), "volumeMounts": [ { "name": "mysql-defaults", "mountPath": "/run/mysql", "readOnly": True, }, { "name": "gpg-public-key", "mountPath": "/run/gpg", "readOnly": True, }, { "name": "workflow-spec", "mountPath": "/run/workflow", "readOnly": True, }, { "name": "temporary", "mountPath": "/tmp", }, ], } ], "volumes": [ { "name": "mysql-defaults", "secret": { "secretName": workflow.mysql.secret_name, "defaultMode": 0o400, "items": [ { "key": workflow.mysql.secret_key, "path": "mysql.cnf", } ], }, }, { "name": "gpg-public-key", "configMap": { "name": gpg_config_map, "defaultMode": 0o444, "items": [ { "key": GPG_CONFIG_KEY, "path": "pubkey.asc", } ], }, }, { "name": "workflow-spec", "configMap": { "name": workflow_config_map, "items": [ { "key": WORKFLOW_CONFIG_KEY, "path": "workflow.yaml", } ], }, }, {"name": "temporary", "emptyDir": {}}, ], }, }, }, }, }, } return ( mysql_secret_document, lark_secret_document, gpg_config_map_document, workflow_config_map_document, cron_job_document, ) def serialize_manifest(documents: Sequence[Mapping[str, Any]]) -> str: return yaml.dump_all( list(documents), Dumper=ManifestDumper, explicit_start=True, sort_keys=False, allow_unicode=True, default_flow_style=False, ) def _resolved_path(path: str) -> Path: return Path(path).resolve(strict=False) def validate_output_target(output: str, input_paths: Sequence[str]) -> Path: output_path = Path(output) if not output_path.name: raise GeneratorError("--output must name a file") parent = output_path.parent if not parent.is_dir(): raise GeneratorError(f"output directory does not exist: {str(parent)!r}") resolved_output = _resolved_path(output) for input_path in input_paths: if resolved_output == _resolved_path(input_path): raise GeneratorError("--output must not overwrite an input file") return output_path def write_output_file(path: Path, content: str, force: bool) -> None: flags = os.O_WRONLY | os.O_CREAT flags |= os.O_TRUNC if force else os.O_EXCL if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW descriptor = -1 try: descriptor = os.open(path, flags, 0o600) os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as output: descriptor = -1 output.write(content) except FileExistsError as exc: raise GeneratorError( f"output file already exists: {str(path)!r}; pass --force to replace it" ) from exc except OSError as exc: raise GeneratorError( f"cannot write output file {str(path)!r} ({type(exc).__name__})" ) from exc finally: if descriptor >= 0: os.close(descriptor) def write_stdout(content: str) -> None: try: sys.stdout.write(content) sys.stdout.flush() except OSError as exc: raise GeneratorError( f"cannot write manifest to stdout ({type(exc).__name__})" ) from exc def _stderr_supports_style() -> bool: return ( sys.stderr.isatty() and os.environ.get("TERM") != "dumb" and "NO_COLOR" not in os.environ ) def _styled_summary_text(text: str, style: str, enabled: bool) -> str: if not enabled: return text return f"{style}{text}{ANSI_RESET}" def emit_generation_summary( output_path: Optional[Path], gpg_public_key: ParsedGpgPublicKey, mysql: MysqlConnectionInfo, ) -> None: styled = _stderr_supports_style() def subdued(text: str) -> str: return _styled_summary_text(text, ANSI_DIM, styled) def emphasized(text: str) -> str: return _styled_summary_text(text, ANSI_BOLD, styled) for key in gpg_public_key.keys: holders = json.dumps(list(key.user_ids), ensure_ascii=True) print( subdued("GPG public key: version=") + emphasized(str(key.version)) + subdued(" fingerprint=") + emphasized(key.fingerprint) + subdued(" holders=") + emphasized(holders), file=sys.stderr, ) port_source = "explicit" if mysql.port_is_explicit else "default" print( subdued("MySQL target: host=") + emphasized(json.dumps(mysql.host)) + subdued(" port=") + emphasized(f"{mysql.port} ({port_source})") + subdued(" user=") + emphasized(json.dumps(mysql.user)) + subdued(" database=") + emphasized(json.dumps(mysql.database)), file=sys.stderr, ) if output_path is None: destination = "stdout" else: destination = f"{output_path} with mode 0600" print( subdued("wrote 5 Kubernetes resources to ") + emphasized(destination), file=sys.stderr, ) def build_parser() -> Parser: parser = Parser( description=( "Generate multi-document YAML containing packed-backup " "CronJob inputs and the CronJob itself. The command is fully offline." ), formatter_class=argparse.ArgumentDefaultsHelpFormatter, allow_abbrev=False, ) parser.add_argument( "--config", required=True, metavar="PATH", help="CronJob deployment and workflow YAML", ) parser.add_argument( "--gpg-public-key", required=True, metavar="PATH", help="explicit local ASCII-armored OpenPGP public-key source", ) parser.add_argument( "--mysql-defaults-file", required=True, metavar="PATH", help="explicit local my.cnf source; there is no home-directory default", ) lark = parser.add_mutually_exclusive_group(required=True) lark.add_argument( "--lark-webhook-file", metavar="PATH", help="file containing the Lark webhook URL", ) lark.add_argument( "--lark-webhook-env", metavar="NAME", help="explicit environment variable containing the Lark webhook URL", ) parser.add_argument( "--output", metavar="PATH", default="-", help="write YAML to PATH; '-' or omission writes to stdout", ) parser.add_argument( "--force", action="store_true", help="replace an existing --output file", ) return parser def run(argv: Optional[Sequence[str]] = None) -> Optional[Path]: parser = build_parser() args = parser.parse_args(argv) input_paths = [ args.config, args.gpg_public_key, args.mysql_defaults_file, ] if args.lark_webhook_file is not None: input_paths.append(args.lark_webhook_file) output_path = None if args.output != "-": output_path = validate_output_target(args.output, input_paths) elif args.force: raise GeneratorError("--force requires --output PATH") config = load_generator_config(args.config) gpg_payload = _read_regular_file( args.gpg_public_key, "GPG public key", MAX_GPG_BYTES, ) gpg_public_key = inspect_gpg_public_key(gpg_payload) mysql_defaults = _read_regular_file( args.mysql_defaults_file, "my.cnf", MAX_MYSQL_BYTES, ) mysql_connection = inspect_mysql_defaults(mysql_defaults) webhook = read_lark_webhook( args.lark_webhook_file, args.lark_webhook_env, ) documents = build_manifest( config, gpg_public_key.armored, mysql_defaults, webhook, ) manifest = serialize_manifest(documents) if output_path is None: write_stdout(manifest) else: write_output_file(output_path, manifest, args.force) emit_generation_summary(output_path, gpg_public_key, mysql_connection) return output_path def main(argv: Optional[Sequence[str]] = None) -> int: try: run(argv) except GeneratorError as exc: print(f"generate_packed_backup_cronjob.py: error: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())