#!/usr/bin/env python3 """Generate the prod AWS BYOC customer-pull image list. The generator treats image-sync-byoc-prod.yaml as the customer-pull image name allowlist. Deployment repositories are used only to resolve the exact tag for each allowlisted non-kernel image in the prod AWS BYOC scope. """ from __future__ import annotations import argparse import copy import dataclasses import difflib import io import re import sys from collections import defaultdict from pathlib import Path from typing import Any, Iterable import yaml as pyyaml from ruamel.yaml import YAML from ruamel.yaml.comments import CommentedMap, CommentedSeq DBAAS_REGISTRY = "380838443567.dkr.ecr.us-west-2.amazonaws.com" O11Y_REGISTRY = "557537366020.dkr.ecr.us-west-2.amazonaws.com" PULL_IMAGE_SETS = ("kernelImages", "dbaasImages", "o11yImages", "o11yCharts") GENERATED_IMAGE_SETS = ("dbaasImages", "o11yImages", "o11yCharts") IMAGE_SET_REGISTRY = { "dbaasImages": DBAAS_REGISTRY, "o11yImages": O11Y_REGISTRY, "o11yCharts": O11Y_REGISTRY, } IMAGE_SET_REPOSITORY_SET = { "dbaasImages": "tidbRepos", "o11yImages": "o11yRepos", "o11yCharts": "o11yRepos", } SOURCE_ROOTS = { "nextgen-regional-control-plane": "dbaasImages", "dedicated-cluster-ng-service": "dbaasImages", } PRIORITY_BYOC_VARIANT = 0 PRIORITY_PROD_AWS_VARIANT = 1 PRIORITY_GENERAL_VARIANT = 2 PRIORITY_BASE = 3 PRIORITY_NAMES = { PRIORITY_BYOC_VARIANT: "BYOC-specific variant or managed override", PRIORITY_PROD_AWS_VARIANT: "prod AWS variant", PRIORITY_GENERAL_VARIANT: "general variant", PRIORITY_BASE: "base config", } O11Y_ENV = "prod" O11Y_VENDOR = "aws" O11Y_LOGICAL_REGION = "byoc-pull-generator" O11Y_MANAGED_VALUE_FILES = ( "infra-v2/ops-server/variants/byoc-aws-values-tmpl.yaml", "infra-v2/kube-prometheus-stack/variants/prod-aws-byoc-values-tmpl.yaml", "infra-v2/cert-manager/variants/values-aws-tmpl.yaml", "infra-v2/vm-operator/variants/values-prod-tmpl.yaml", "infra-v2/eventrouter/variants/values_tmpl.yaml", "infra-v2/volume-autoscaler/variants/values_tmpl.yaml", "infra-v2/vector/variants/aws-values-tmpl.yaml", "infra-v2/velero/variants/aws-values-tmpl.yaml", "cloud-addon/aws-lb-controller/variants/prod/values-tmpl.yaml", "cloud-addon/external-dns/variants/prod/values-aws-tmpl.yaml", "cloud-addon/cluster-autoscaler/variants/prod/values-aws-tmpl.yaml", "cloud-addon/ingress-nginx-write/variants/aws/prod-aws-values-tmpl.yaml", "cloud-addon/ingress-nginx-read/variants/aws/byoc-aws-values-tmpl.yaml", "dataplane/vmcluster/variants/prod-aws-byoc-values-tmpl.yaml", "dataplane/loki/variants/aws-values-tmpl.yaml", "dataplane/alertmanager/variants/aws-byoc-values-tmpl.yaml", "drhouse/diagnosis-query/variants/aws/values-tmpl.yaml", "drhouse/o11ydiagnosis-deltalake/variants/aws/values-tmpl.yaml", ) O11Y_LOCAL_CHART_DIRS = ( "ops-server", "kube-prometheus-stack", "cert-manager", "o11y-mixin", "eventrouter", "volume-autoscaler", "vector", "velero", "aws-load-balancer-controller", "external-dns", "ingress-nginx-v411", "vmcluster", "alertmanager", "regional-server", "diagnosis-query", "o11ydiagnosis-deltalake", ) O11Y_EXTERNAL_CHARTS = ( ("observability-charts/victoria-metrics-operator", "infra-v2/deploy-with-ops.sh", "VM_OPERATOR_CHART_VERSION"), ("observability-charts/cluster-autoscaler", "cloud-addon/deploy-with-ops.sh", "CLUSTER_AUTOSCALER_VERSION"), ("observability-charts/loki", "dataplane/deploy-with-ops.sh", "LOKI_CHART_VERSION"), ) DRHOUSE_IMAGE_TAG_REFS = ( ("diagnosis-query", ("image", "tag"), "diagnosis-query"), ("o11ydiagnosis-deltalake", ("image", "tag"), "o11ydiagnosis"), ("o11ydiagnosis-deltalake", ("taskCollector", "image", "tag"), "observability-regional-task-collector"), ) KERNEL_IMAGE_NAMES = { "tidbcloud/pd", "tidbcloud/tidb", "tidbcloud/tiflash", "tidbcloud/tikv", "tidbcloud/ticdc", "tidbcloud/br", "tidbcloud/tiproxy", } PLACEHOLDER_REGISTRIES = { "IMAGE_REGISTRY", "TIDB_IMAGE_REGISTRY", "TMPL_IMAGE_REGISTRY", "TMPL_TIDB_IMAGE_REGISTRY", "TPML_IMAGE_REGISTRY", "TPML_TIDB_IMAGE_REGISTRY", } IMAGE_REF_RE = re.compile( r"(?\d{12}):(?Proot|role/[A-Za-z0-9+=,.@_/-]{1,128})$" ) LITERAL_TAG_RE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.+@-]*$") MUTABLE_TAGS = {"latest"} @dataclasses.dataclass(frozen=True) class ImageRef: name: str tag: str source: str | None image_set: str file: str priority: int @dataclasses.dataclass(frozen=True) class RequiredImage: name: str source: str tag: str image_set: str upstream_source: str | None = None class IndentedSafeDumper(pyyaml.SafeDumper): def increase_indent(self, flow: bool = False, indentless: bool = False) -> Any: return super().increase_indent(flow, False) def roundtrip_yaml() -> YAML: yaml = YAML() yaml.preserve_quotes = True yaml.width = 120 yaml.indent(mapping=2, sequence=4, offset=2) return yaml def load_roundtrip_yaml(path: Path) -> Any: yaml = roundtrip_yaml() with path.open() as f: return yaml.load(f) def dump_roundtrip_yaml(data: Any) -> str: yaml = roundtrip_yaml() output = io.StringIO() yaml.dump(data, output) return output.getvalue() def dump_generated_yaml(data: Any) -> str: return pyyaml.dump( data, sort_keys=False, default_flow_style=False, width=120, Dumper=IndentedSafeDumper, ) def write_if_changed(path: Path, content: str, check: bool, changed: list[Path]) -> None: old = path.read_text() if path.exists() else "" if old == content: return changed.append(path) if check: print_diff(path, old, content) return path.write_text(content) def print_diff(path: Path, old: str, new: str) -> None: diff = difflib.unified_diff( old.splitlines(), new.splitlines(), fromfile=str(path), tofile=str(path), lineterm="", ) print("\n".join(diff)) def is_registry(first_segment: str) -> bool: return ( first_segment in PLACEHOLDER_REGISTRIES or "." in first_segment or ":" in first_segment or first_segment == "localhost" ) def parse_image_ref( image_ref: str, image_set: str, file: str, priority: int, *, allow_bare_repository: bool = False, ) -> ImageRef | None: if "@" in image_ref: return None if image_ref.startswith(("http://", "https://", "//")): return None if ":" not in image_ref: return None repository, tag = image_ref.rsplit(":", 1) if ( "/" not in repository and not allow_bare_repository and not repository.startswith(("dbaas", "tidbcloud", "third-party", "observability")) ): return None first, _, rest = repository.partition("/") if not rest: if allow_bare_repository: return ImageRef(name=repository, tag=tag, source=None, image_set=image_set, file=file, priority=priority) return None if is_registry(first): name = rest source = None if first in PLACEHOLDER_REGISTRIES else repository else: name = repository source = None if not name or not tag: return None return ImageRef(name=name, tag=tag, source=source, image_set=image_set, file=file, priority=priority) def extract_image_refs_from_text(text: str, image_set: str, file: str, priority: int) -> list[ImageRef]: refs = [] for match in IMAGE_REF_RE.finditer(text): parsed = parse_image_ref(match.group(1), image_set, file, priority) if parsed: refs.append(parsed) return refs def image_from_parts(repository: Any, tag: Any, registry: Any | None = None) -> str | None: if repository is None or tag is None: return None repo = str(repository).strip() image_tag = str(tag).strip() if not repo or not image_tag or image_tag in {"None", "null"}: return None if registry is not None and str(registry).strip(): registry_s = str(registry).strip().rstrip("/") first_segment = repo.split("/", 1)[0] if not repo.startswith(registry_s + "/") and not is_registry(first_segment): repo = f"{registry_s}/{repo.lstrip('/')}" return f"{repo}:{image_tag}" def walk_yaml(value: Any, image_set: str, file: str, priority: int) -> Iterable[ImageRef]: if isinstance(value, dict): image = image_from_parts(value.get("repository"), value.get("tag"), value.get("registry")) if image: parsed = parse_image_ref(image, image_set, file, priority, allow_bare_repository=True) if parsed: yield parsed image = image_from_parts(value.get("image"), value.get("tag"), value.get("registry")) if image: parsed = parse_image_ref(image, image_set, file, priority, allow_bare_repository=True) if parsed: yield parsed image = image_from_parts(value.get("newName"), value.get("newTag")) if image: parsed = parse_image_ref(image, image_set, file, priority, allow_bare_repository=True) if parsed: yield parsed for child in value.values(): yield from walk_yaml(child, image_set, file, priority) elif isinstance(value, list): for child in value: yield from walk_yaml(child, image_set, file, priority) elif isinstance(value, str): yield from extract_image_refs_from_text(value, image_set, file, priority) def has_any_marker(text: str, markers: Iterable[str]) -> bool: return any(marker in text for marker in markers) def classify_source_file(path: Path, root: Path) -> int | None: if path.suffix not in {".yaml", ".yml"}: return None try: rel = path.relative_to(root) except ValueError: rel = path parts = tuple(part.lower() for part in rel.parts) rel_text = rel.as_posix().lower() if any(part in {"env", "envs"} for part in parts): return None if has_any_marker(rel_text, ["/dev/", "dev-", "/staging/", "staging-"]): return None if has_any_marker(rel_text, ["azure", "alicloud", "aliyun", "gcp"]): return None in_variants = "variants" in parts if in_variants: is_byoc_variant = "byoc" in rel_text and "aws" in rel_text if is_byoc_variant: return PRIORITY_BYOC_VARIANT provider_markers = ("aws", "byoc", "azure", "alicloud", "aliyun", "gcp") is_prod_aws = "prod-aws" in rel_text or "aws-prod" in rel_text is_provider_specific = any(marker in rel_text for marker in provider_markers) if is_prod_aws: return PRIORITY_PROD_AWS_VARIANT if not is_provider_specific: return PRIORITY_GENERAL_VARIANT return None return PRIORITY_BASE def collect_source_files(root: Path) -> list[tuple[Path, int]]: if not root.exists(): raise FileNotFoundError(f"source root does not exist: {root}") files = [] for path in root.rglob("*"): if not path.is_file(): continue priority = classify_source_file(path, root) if priority is not None: files.append((path, priority)) return sorted(files, key=lambda item: (item[1], str(item[0]))) def collect_image_refs_from_yaml_file(path: Path, image_set: str, priority: int) -> tuple[list[ImageRef], bool]: text = path.read_text() refs: list[ImageRef] = [] try: docs = list(pyyaml.safe_load_all(text)) except pyyaml.YAMLError: return extract_image_refs_from_text(text, image_set, str(path), priority), True for doc in docs: refs.extend(walk_yaml(doc, image_set, str(path), priority)) return refs, False def collect_app_delivery_images(source_roots: dict[str, Path]) -> list[ImageRef]: refs: list[ImageRef] = [] seen: set[ImageRef] = set() parse_fallbacks: list[Path] = [] for source_root, root in source_roots.items(): image_set = SOURCE_ROOTS[source_root] for path, priority in collect_source_files(root): file_refs, used_fallback = collect_image_refs_from_yaml_file(path, image_set, priority) if used_fallback: parse_fallbacks.append(path) for ref in file_refs: if ref not in seen: refs.append(ref) seen.add(ref) if parse_fallbacks: print( f"warning: used text image scanning fallback for {len(parse_fallbacks)} templated YAML files", file=sys.stderr, ) for path in parse_fallbacks[:10]: print(f" - {path}", file=sys.stderr) if len(parse_fallbacks) > 10: print(f" - ... {len(parse_fallbacks) - 10} more", file=sys.stderr) return refs def read_chart_metadata(chart_dir: Path) -> tuple[str, str]: chart_file = chart_dir / "Chart.yaml" data = pyyaml.safe_load(chart_file.read_text()) or {} name = str(data.get("name", "")).strip() version = str(data.get("version", "")).strip() if not name or not version: raise ValueError(f"{chart_file}: chart name and version are required") return name, version def read_shell_chart_version(script: Path, variable: str) -> str: text = script.read_text() direct = re.search( rf"^\s*{re.escape(variable)}\s*=\s*(?P['\"]?)(?P[A-Za-z0-9_.+-]+)(?P=quote)\s*$", text, re.M, ) if direct: return direct.group("value") defaulted = re.search(rf"\$\{{{re.escape(variable)}:-(?P[A-Za-z0-9_.+-]+)\}}", text) if defaulted: return defaulted.group("value") raise ValueError(f"{script}: unable to resolve {variable}") def get_nested(data: Any, path: Iterable[str]) -> Any: current = data for part in path: if not isinstance(current, dict) or part not in current: return None current = current[part] return current def resolve_drhouse_image_tag_file(component_dir: Path) -> Path: env_file = component_dir / "envs" / f"{O11Y_ENV}-{O11Y_VENDOR}-{O11Y_LOGICAL_REGION}" / "image-tag.yaml" if env_file.exists(): return env_file return component_dir / "image-tag.yaml" def read_auto_deploy_regional_server_tag(auto_deploy_root: Path) -> tuple[str, Path]: vars_file = auto_deploy_root / "shoot" / "vars.tf" text = vars_file.read_text() variable_match = re.search(r'variable\s+"regional_server_img_tag"\s*\{(?P.*?)\n\}', text, re.S) if not variable_match: raise ValueError(f"{vars_file}: regional_server_img_tag variable not found") default_match = re.search(r'default\s*=\s*"(?P[^"]+)"', variable_match.group("body")) if not default_match: raise ValueError(f"{vars_file}: regional_server_img_tag default not found") return default_match.group("tag"), vars_file def collect_observability_images(observability_root: Path, auto_deploy_root: Path) -> list[ImageRef]: deployments = observability_root / "deployments" charts = deployments / "charts" refs: list[ImageRef] = [] seen: set[ImageRef] = set() parse_fallbacks: list[Path] = [] def add(ref: ImageRef) -> None: if ref not in seen: refs.append(ref) seen.add(ref) def add_yaml_refs(path: Path, priority: int) -> None: if not path.exists(): return file_refs, used_fallback = collect_image_refs_from_yaml_file(path, "o11yImages", priority) if used_fallback: parse_fallbacks.append(path) for ref in file_refs: add(ref) for rel_path in O11Y_MANAGED_VALUE_FILES: add_yaml_refs(deployments / rel_path, PRIORITY_BYOC_VARIANT) for chart_dir_name in O11Y_LOCAL_CHART_DIRS: chart_dir = charts / chart_dir_name chart_file = chart_dir / "Chart.yaml" if not chart_file.exists(): continue chart_name, chart_version = read_chart_metadata(chart_dir) add( ImageRef( name=f"observability-charts/{chart_name}", tag=chart_version, source=None, image_set="o11yCharts", file=str(chart_file), priority=PRIORITY_BYOC_VARIANT, ) ) add_yaml_refs(chart_dir / "values.yaml", PRIORITY_BASE) for name, script_rel_path, variable in O11Y_EXTERNAL_CHARTS: script = deployments / script_rel_path if not script.exists(): continue add( ImageRef( name=name, tag=read_shell_chart_version(script, variable), source=None, image_set="o11yCharts", file=f"{script}:{variable}", priority=PRIORITY_BYOC_VARIANT, ) ) for component, tag_path, image_name in DRHOUSE_IMAGE_TAG_REFS: image_tag_file = resolve_drhouse_image_tag_file(deployments / "drhouse" / component) if not image_tag_file.exists(): continue data = pyyaml.safe_load(image_tag_file.read_text()) or {} tag = get_nested(data, tag_path) if tag is None or str(tag).strip() == "": continue add( ImageRef( name=image_name, tag=str(tag).strip(), source=None, image_set="o11yImages", file=str(image_tag_file), priority=PRIORITY_BYOC_VARIANT, ) ) regional_tag, vars_file = read_auto_deploy_regional_server_tag(auto_deploy_root) for image_name in ("observability-regional-server", "observability-regional-data-server"): add( ImageRef( name=image_name, tag=regional_tag, source=None, image_set="o11yImages", file=str(vars_file), priority=PRIORITY_BYOC_VARIANT, ) ) if parse_fallbacks: print( f"warning: used text image scanning fallback for {len(parse_fallbacks)} observability templated YAML files", file=sys.stderr, ) for path in parse_fallbacks[:10]: print(f" - {path}", file=sys.stderr) if len(parse_fallbacks) > 10: print(f" - ... {len(parse_fallbacks) - 10} more", file=sys.stderr) return refs def collect_deployment_images( app_source_roots: dict[str, Path], observability_root: Path, auto_deploy_root: Path, ) -> list[ImageRef]: return [ *collect_app_delivery_images(app_source_roots), *collect_observability_images(observability_root, auto_deploy_root), ] def repo_tags(entry: dict[str, Any]) -> list[str]: tags = entry.get("tags", {}).get("allow", []) if tags is None: return [] return [str(tag) for tag in tags] def is_mutable_tag(tag: str) -> bool: return tag.lower() in MUTABLE_TAGS def ensure_repository_set(config: Any, set_name: str) -> CommentedSeq: repo_sets = config.setdefault("repositorySets", CommentedMap()) repos = repo_sets.setdefault(set_name, CommentedSeq()) return repos def make_repo_entry(name: str, source: str, tag: str) -> CommentedMap: entry = CommentedMap() entry["name"] = name entry["source"] = source tags = CommentedMap() allow = CommentedSeq() allow.append(tag) tags["allow"] = allow entry["tags"] = tags return entry def find_entry(repos: Iterable[dict[str, Any]], name: str, source: str | None = None) -> dict[str, Any] | None: for entry in repos: if str(entry.get("name")) != name: continue if source is not None and str(entry.get("source")) != source: continue return entry return None def ensure_tag(entry: dict[str, Any], tag: str) -> bool: tags = entry.setdefault("tags", CommentedMap()) allow = tags.setdefault("allow", CommentedSeq()) existing = [str(item) for item in allow] if tag in existing: return False allow.append(tag) return True def set_exact_tag(entry: dict[str, Any], tag: str) -> bool: tags = entry.setdefault("tags", CommentedMap()) allow = tags.setdefault("allow", CommentedSeq()) existing = [str(item) for item in allow] if existing == [tag]: return False new_allow = CommentedSeq() new_allow.append(tag) tags["allow"] = new_allow return True def source_registry(source: str) -> str | None: first, _, rest = source.partition("/") if not rest: return None if is_registry(first): return first return None def desired_pingcap_source(image_set: str, name: str) -> str: return f"{IMAGE_SET_REGISTRY[image_set]}/{name}" def should_add_to_prod_sync(source: str, desired_source: str) -> bool: if not source or source == desired_source: return False if any(source.startswith(f"{registry}/") for registry in IMAGE_SET_REGISTRY.values()): return False registry = source_registry(source) if registry and ECR_REGISTRY_RE.fullmatch(registry): return False return True def build_prod_index(prod_config: dict[str, Any]) -> set[tuple[str, str, str]]: index = set() for repos in prod_config.get("repositorySets", {}).values(): for entry in repos or []: for tag in repo_tags(entry): index.add((str(entry.get("name")), str(entry.get("source")), tag)) return index def ensure_prod_mirror( prod_config: Any, prod_index: set[tuple[str, str, str]], name: str, source: str, tag: str, ) -> bool: if (name, source, tag) in prod_index: return False common_repos = ensure_repository_set(prod_config, "commonRepositories") common_repos.append(make_repo_entry(name=name, source=source, tag=tag)) prod_index.add((name, source, tag)) return True def byoc_entries_for_image_set(byoc_config: Any, image_set: str) -> CommentedSeq: return ensure_repository_set(byoc_config, IMAGE_SET_REPOSITORY_SET[image_set]) def is_observability_chart_name(name: str) -> bool: return name.startswith("observability-charts/") def byoc_name_belongs_to_image_set(image_set: str, name: str) -> bool: if image_set == "dbaasImages": return name not in KERNEL_IMAGE_NAMES if image_set == "o11yImages": return not is_observability_chart_name(name) if image_set == "o11yCharts": return is_observability_chart_name(name) return True def output_name_belongs_to_image_set(image_set: str, name: str) -> bool: if image_set == "o11yCharts": return is_observability_chart_name(name) return not is_observability_chart_name(name) def build_byoc_name_source_indexes( byoc_config: Any, ) -> tuple[dict[str, set[str]], set[str], dict[str, str]]: names: dict[str, set[str]] = {image_set: set() for image_set in GENERATED_IMAGE_SETS} eligible_names: set[str] = set() by_source: dict[str, str] = {} for image_set in GENERATED_IMAGE_SETS: for entry in byoc_entries_for_image_set(byoc_config, image_set): name = str(entry.get("name")) if not byoc_name_belongs_to_image_set(image_set, name): continue names[image_set].add(name) eligible_names.add(name) source = entry.get("source") if source: by_source.setdefault(str(source), name) return names, eligible_names, by_source def match_byoc_name( ref: ImageRef, eligible_names: set[str], byoc_source_index: dict[str, str], ) -> str | None: if ref.name in eligible_names and output_name_belongs_to_image_set(ref.image_set, ref.name): return ref.name third_party_name = f"third-party/{ref.name}" if third_party_name in eligible_names and output_name_belongs_to_image_set(ref.image_set, third_party_name): return third_party_name if ref.source and ref.source in byoc_source_index: name = byoc_source_index[ref.source] if output_name_belongs_to_image_set(ref.image_set, name): return name return None def format_deployment_tag_conflict( image_set: str, name: str, priority: int, by_tag: dict[str, list[ImageRef]], ) -> list[str]: lines = [ f"{image_set}:{name} resolves to multiple tags at {PRIORITY_NAMES[priority]} priority:", ] for tag, refs_for_tag in sorted(by_tag.items()): files = sorted({ref.file for ref in refs_for_tag}) for file in files: lines.append(f" - {tag}: {file}") return lines def resolve_required_images( refs: Iterable[ImageRef], byoc_config: Any, *, fail_on_conflict: bool = True, ) -> list[RequiredImage]: byoc_names, eligible_names, byoc_source_index = build_byoc_name_source_indexes(byoc_config) grouped: dict[tuple[str, str], dict[int, list[ImageRef]]] = defaultdict(lambda: defaultdict(list)) for ref in refs: if ref.name in KERNEL_IMAGE_NAMES: continue byoc_name = match_byoc_name(ref, eligible_names, byoc_source_index) if byoc_name is None: continue resolved_ref = dataclasses.replace(ref, name=byoc_name) grouped[(resolved_ref.image_set, resolved_ref.name)][resolved_ref.priority].append(resolved_ref) required = [] skipped_mutable_tags: list[tuple[str, str, str, int, list[str]]] = [] skipped_conflicts: list[tuple[str, str, int, dict[str, list[ImageRef]]]] = [] skipped_names: set[tuple[str, str]] = set() for (image_set, name), refs_by_priority in sorted(grouped.items()): priority = min(refs_by_priority) selected_refs = refs_by_priority[priority] by_tag: dict[str, list[ImageRef]] = defaultdict(list) for ref in selected_refs: by_tag[ref.tag].append(ref) if len(by_tag) > 1: if fail_on_conflict: raise ValueError("\n".join(format_deployment_tag_conflict(image_set, name, priority, by_tag))) skipped_conflicts.append((image_set, name, priority, by_tag)) skipped_names.add((image_set, name)) continue tag = next(iter(by_tag)) if is_mutable_tag(tag): skipped_mutable_tags.append( ( image_set, name, tag, priority, sorted({ref.file for ref in selected_refs}), ) ) skipped_names.add((image_set, name)) continue desired_source = desired_pingcap_source(image_set, name) upstream_source = None for ref in selected_refs: if ref.source and should_add_to_prod_sync(ref.source, desired_source): upstream_source = ref.source break required.append( RequiredImage( name=name, source=desired_source, tag=tag, image_set=image_set, upstream_source=upstream_source, ) ) report_deployment_tag_conflicts(skipped_conflicts) report_mutable_deployment_tags(skipped_mutable_tags) report_unresolved_byoc_images(byoc_names, required, skipped_names) return required def report_deployment_tag_conflicts(items: Iterable[tuple[str, str, int, dict[str, list[ImageRef]]]]) -> None: items = list(items) if not items: return print("warning: BYOC-listed generated images resolved to multiple deployment tags:", file=sys.stderr) for image_set, name, priority, by_tag in items: for line in format_deployment_tag_conflict(image_set, name, priority, by_tag): print(f" {line}", file=sys.stderr) def report_mutable_deployment_tags(items: Iterable[tuple[str, str, str, int, list[str]]]) -> None: items = list(items) if not items: return print("warning: BYOC-listed generated images resolved to mutable deployment tags:", file=sys.stderr) for image_set, name, tag, priority, files in items: print(f" - {image_set}:{name} tag {tag!r} at {PRIORITY_NAMES[priority]}:", file=sys.stderr) for file in files: print(f" - {file}", file=sys.stderr) def report_unresolved_byoc_images( byoc_names: dict[str, set[str]], required: Iterable[RequiredImage], skipped_names: set[tuple[str, str]] | None = None, ) -> None: resolved_names = {item.name for item in required} skipped_names = skipped_names or set() skipped_name_values = {name for _, name in skipped_names} missing = [] for image_set, names in byoc_names.items(): for name in names: if (image_set, name) in skipped_names: continue if name in skipped_name_values: continue if name not in resolved_names: missing.append((image_set, name)) if not missing: return print("warning: BYOC-listed generated images without resolved prod AWS BYOC tags:", file=sys.stderr) for image_set, name in sorted(missing): print(f" - {image_set}:{name}", file=sys.stderr) def normalize_byoc_public_entries( repos: CommentedSeq, item: RequiredImage, prod_config: Any, prod_index: set[tuple[str, str, str]], ) -> bool: changed = False desired_entry = find_entry(repos, item.name, item.source) for entry in list(repos): if str(entry.get("name")) != item.name: continue source = str(entry.get("source", "")) if not should_add_to_prod_sync(source, item.source): continue if ensure_prod_mirror(prod_config, prod_index, item.name, source, item.tag): changed = True if desired_entry is not None and desired_entry is not entry: if ensure_tag(desired_entry, item.tag): changed = True repos.remove(entry) changed = True continue if source != item.source: entry["source"] = item.source changed = True if set_exact_tag(entry, item.tag): changed = True desired_entry = entry return changed def update_sync_configs( required: Iterable[RequiredImage], prod_config: Any, byoc_config: Any, ) -> tuple[bool, bool]: prod_index = build_prod_index(prod_config) prod_changed = False byoc_changed = False for item in sorted(required, key=lambda image: (image.image_set, image.name, image.tag)): repos = byoc_entries_for_image_set(byoc_config, item.image_set) if item.upstream_source and ensure_prod_mirror(prod_config, prod_index, item.name, item.upstream_source, item.tag): prod_changed = True if normalize_byoc_public_entries(repos, item, prod_config, prod_index): prod_changed = True byoc_changed = True pingcap_entry = find_entry(repos, item.name, item.source) if pingcap_entry is not None and ensure_tag(pingcap_entry, item.tag): byoc_changed = True return prod_changed, byoc_changed def all_byoc_names(byoc_config: Any) -> set[str]: names = set() for repos in byoc_config.get("repositorySets", {}).values(): for entry in repos or []: if entry.get("name"): names.add(str(entry.get("name"))) return names def load_existing_kernel_images(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] data = pyyaml.safe_load(path.read_text()) or {} if "kernelImages" in data: return copy.deepcopy(data.get("kernelImages") or []) return [ copy.deepcopy(item) for item in data.get("images", []) or [] if isinstance(item, dict) and item.get("name") in KERNEL_IMAGE_NAMES ] def validate_pull_image_row(path: Path, set_name: str, index: int, image: Any, byoc_names: set[str] | None) -> None: if not isinstance(image, dict): raise ValueError(f"{path}: {set_name}[{index}] must be a map") expected_fields = ( {"name", "source", "tags"} if set_name == "kernelImages" else {"name", "source", "tag"} ) if set(image) != expected_fields: fields = ", ".join(sorted(expected_fields)) raise ValueError(f"{path}: {set_name}[{index}] must contain only {fields}") name = image.get("name") source = image.get("source") source_prefixes = tuple(f"{registry}/" for registry in IMAGE_SET_REGISTRY.values()) if not isinstance(name, str) or not name: raise ValueError(f"{path}: {set_name}[{index}].name must be a non-empty string") if not isinstance(source, str) or not source.startswith(source_prefixes): raise ValueError(f"{path}: {set_name}[{index}].source must be a PingCAP prod ECR source") tags = image.get("tags") if set_name == "kernelImages" else [image.get("tag")] if not isinstance(tags, list) or not tags: raise ValueError(f"{path}: {set_name}[{index}].tags must be a non-empty list") seen_tags: set[str] = set() for tag_index, tag in enumerate(tags): field = ( f"{set_name}[{index}].tags[{tag_index}]" if set_name == "kernelImages" else f"{set_name}[{index}].tag" ) if not isinstance(tag, str) or not LITERAL_TAG_RE.fullmatch(tag): raise ValueError(f"{path}: {field} must be one literal tag") if is_mutable_tag(tag): raise ValueError(f"{path}: {field} {tag!r} is a mutable tag and is not allowed") if tag in seen_tags: raise ValueError(f"{path}: {set_name}[{index}] has duplicate tag {tag!r}") seen_tags.add(tag) if set_name == "kernelImages": if name not in KERNEL_IMAGE_NAMES: raise ValueError(f"{path}: kernelImages[{index}].name {name!r} is not an allowed kernel image") if byoc_names is not None and name not in byoc_names: raise ValueError(f"{path}: kernelImages[{index}].name {name!r} is missing from image-sync-byoc-prod.yaml") def validate_pull_images(path: Path, byoc_config: Any | None = None) -> None: if not path.exists(): return data = pyyaml.safe_load(path.read_text()) or {} if "images" in data: raise ValueError(f"{path}: use kernelImages, dbaasImages, o11yImages, and o11yCharts instead of images") byoc_names = all_byoc_names(byoc_config) if byoc_config is not None else None seen: set[tuple[str, str, str]] = set() seen_names_by_set: dict[str, set[str]] = {set_name: set() for set_name in PULL_IMAGE_SETS} for set_name in PULL_IMAGE_SETS: images = data.get(set_name) if not isinstance(images, list): raise ValueError(f"{path}: {set_name} must be a list") for index, image in enumerate(images): validate_pull_image_row(path, set_name, index, image, byoc_names) name = str(image["name"]) source = str(image["source"]) if name in seen_names_by_set[set_name]: raise ValueError(f"{path}: duplicate {set_name} image name {name}") seen_names_by_set[set_name].add(name) if set_name == "dbaasImages" and source != desired_pingcap_source("dbaasImages", name): raise ValueError(f"{path}: dbaasImages[{index}].source must be {desired_pingcap_source('dbaasImages', name)}") if set_name == "o11yImages" and source != desired_pingcap_source("o11yImages", name): raise ValueError(f"{path}: o11yImages[{index}].source must be {desired_pingcap_source('o11yImages', name)}") if set_name == "o11yImages" and is_observability_chart_name(name): raise ValueError(f"{path}: o11yImages[{index}].name must not be an observability chart") if set_name == "o11yCharts" and source != desired_pingcap_source("o11yCharts", name): raise ValueError(f"{path}: o11yCharts[{index}].source must be {desired_pingcap_source('o11yCharts', name)}") if set_name == "o11yCharts" and not is_observability_chart_name(name): raise ValueError(f"{path}: o11yCharts[{index}].name must start with observability-charts/") tags = image["tags"] if set_name == "kernelImages" else [image["tag"]] for tag in tags: key = (name, source, str(tag)) if key in seen: raise ValueError(f"{path}: duplicate pull image {source}:{tag}") seen.add(key) def generate_pull_yaml(required: Iterable[RequiredImage], kernel_images: Iterable[dict[str, Any]]) -> str: by_set = {set_name: [] for set_name in PULL_IMAGE_SETS} by_set["kernelImages"] = [dict(item) for item in kernel_images] for item in required: by_set[item.image_set].append({"name": item.name, "source": item.source, "tag": item.tag}) output: dict[str, list[dict[str, Any]]] = {"kernelImages": by_set["kernelImages"]} for image_set in GENERATED_IMAGE_SETS: output[image_set] = sorted(by_set[image_set], key=lambda image: (image["name"], image["tag"])) return dump_generated_yaml(output) def validate_customers(path: Path) -> None: if not path.exists(): return data = pyyaml.safe_load(path.read_text()) or {} for customer in data.get("customers", []): account_id = str(customer.get("accountId", "")) if not re.fullmatch(r"\d{12}", account_id): raise ValueError(f"{path}: invalid accountId for customer {customer.get('name')!r}") principals = customer.get("principals", []) if not principals: raise ValueError(f"{path}: customer {customer.get('name')!r} has no principals") for principal in principals: match = PRINCIPAL_RE.match(str(principal)) if not match: raise ValueError(f"{path}: invalid principal {principal!r}") if match.group("account") != account_id: raise ValueError(f"{path}: principal {principal!r} does not match accountId {account_id}") def default_app_source_roots(args: argparse.Namespace) -> dict[str, Path]: if args.app_delivery_root is None: raise ValueError("--app-delivery-root is required") app_delivery = args.app_delivery_root return { "nextgen-regional-control-plane": app_delivery / "nextgen-regional-control-plane", "dedicated-cluster-ng-service": app_delivery / "dedicated-cluster-ng-service", } def require_observability_roots(args: argparse.Namespace) -> tuple[Path, Path]: if args.observability_root is None or args.auto_deploy_root is None: raise ValueError("--observability-root and --auto-deploy-root are required") return args.observability_root, args.auto_deploy_root def run(args: argparse.Namespace) -> int: if args.report_only and not args.check: raise ValueError("--report-only requires --check") repo_root = args.repo_root prod_path = repo_root / "image-sync-prod.yaml" byoc_path = repo_root / "image-sync-byoc-prod.yaml" pull_yaml_path = repo_root / "byoc-pull-images-aws-prod.yaml" customer_path = repo_root / "byoc-pull-customers-aws-prod.yaml" validate_customers(customer_path) byoc_config = load_roundtrip_yaml(byoc_path) if args.validate_local_only: validate_pull_images(pull_yaml_path, byoc_config) print("BYOC pull local config is valid.") return 0 prod_config = load_roundtrip_yaml(prod_path) original_prod = copy.deepcopy(prod_config) original_byoc = copy.deepcopy(byoc_config) observability_root, auto_deploy_root = require_observability_roots(args) refs = collect_deployment_images(default_app_source_roots(args), observability_root, auto_deploy_root) required = resolve_required_images(refs, byoc_config, fail_on_conflict=not args.report_only) prod_changed, byoc_changed = update_sync_configs(required, prod_config, byoc_config) kernel_images = load_existing_kernel_images(pull_yaml_path) changed: list[Path] = [] if prod_changed or dump_roundtrip_yaml(prod_config) != dump_roundtrip_yaml(original_prod): write_if_changed(prod_path, dump_roundtrip_yaml(prod_config), args.check, changed) if byoc_changed or dump_roundtrip_yaml(byoc_config) != dump_roundtrip_yaml(original_byoc): write_if_changed(byoc_path, dump_roundtrip_yaml(byoc_config), args.check, changed) pull_yaml = generate_pull_yaml(required, kernel_images) write_if_changed(pull_yaml_path, pull_yaml, args.check, changed) if not args.check: validate_pull_images(pull_yaml_path, byoc_config) if args.check and changed: print("Generated BYOC pull image artifacts are out of date.", file=sys.stderr) if args.report_only: print("warning: report-only mode is enabled; deployment-derived drift will not fail this run.", file=sys.stderr) else: return 1 if changed: print("Updated:") for path in changed: print(f" - {path}") else: print("BYOC pull image artifacts are up to date.") return 0 def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repo-root", type=Path, default=Path.cwd()) parser.add_argument("--app-delivery-root", type=Path) parser.add_argument("--observability-root", type=Path) parser.add_argument("--auto-deploy-root", type=Path) parser.add_argument("--check", action="store_true", help="fail if generated files are not current") parser.add_argument( "--report-only", action="store_true", help="with --check, report deployment-derived drift and tag conflicts without failing", ) parser.add_argument("--validate-local-only", action="store_true", help="validate local BYOC pull config without source repos") return parser.parse_args(argv) def main(argv: list[str]) -> int: try: return run(parse_args(argv)) except Exception as exc: print(f"ERROR: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))