#!/usr/bin/env python3 from __future__ import annotations import json import re from collections import Counter, defaultdict from copy import deepcopy from datetime import date from pathlib import Path from typing import Any import yaml ROOT = Path(__file__).resolve().parents[1] META_DIR = ROOT / "knowledge" / "meta" DEPLOYMENT_DIR = ROOT / "knowledge" / "deployment" APP_DELIVERY_ROOT = Path("/Users/lc/coding/app-delivery") EXTERNAL_REPO_ROOTS = { "account": Path("/Users/lc/coding/account"), "infra-cd": Path("/Users/lc/coding/infra-cd"), } TOP_LEVEL_MAPPING_EXCLUSIONS = {"argocd-target-clusters-config"} REPO_CATALOGS = [ META_DIR / "repos-private.json", META_DIR / "repos-public.json", DEPLOYMENT_DIR / "deployment-repos.json", ] MANIFEST_RULES = { "kubernetes_service_addr": "Extracted from manifest or config values using kubernetes:/// or in-cluster service DNS targets.", "http_service_url": "Extracted from manifest or config values using in-cluster HTTP service URLs.", "dns_service_addr": "Extracted from manifest or config values using raw service DNS addresses.", "external_mapped_object_url": "Extracted from manifests that point at known public or load-balanced endpoints and normalized back to a mapped deployment object.", "api_group_crd": "Extracted from controller manifests that require an API group or CRD installed by another deployment object.", "cluster_secret_store_ref": "Extracted from ExternalSecret or SecretStore references that require the cluster secret store bundle.", } SPECIAL_MAPPING_EVIDENCE = { "cert-manager-regional-objects": [ { "path": APP_DELIVERY_ROOT / "dedicated-regional-server" / "app-registry" / "cert-manager.yaml", "needle": "name: cert-manager", "type": "addon_applicationset", "detail": "Dedicated regional server bundle deploys cert-manager as a regional addon object.", }, { "path": APP_DELIVERY_ROOT / "dev" / "nextgen-regional-control-plane" / "app-registry" / "cert-manager.yaml", "needle": "name: cert-manager", "type": "addon_applicationset", "detail": "Next-gen regional control-plane bundle also deploys cert-manager in regional environments.", }, ], "clinic": [ { "path": APP_DELIVERY_ROOT / "clinic" / "app-registry" / "clinic-api.yaml", "needle": "name: clinic-api", "type": "bundle_internal_applicationset", "detail": "Clinic delivery family includes the clinic-api ApplicationSet.", }, { "path": APP_DELIVERY_ROOT / "clinic" / "cloud-server" / "base" / "deployment.yaml", "needle": "image: cloud_server:", "type": "bundle_internal_workload", "detail": "Clinic delivery family also packages the cloud-server workload image.", }, { "path": APP_DELIVERY_ROOT / "clinic" / "ngm" / "base" / "statefulset.yaml", "needle": "image: ngm_api:", "type": "bundle_internal_workload", "detail": "Clinic delivery family packages the ngm monitoring workload.", }, ], "infra-etcd-backup": [ { "path": APP_DELIVERY_ROOT / "infra-etcd-backup" / "README.md", "needle": "This application is for taking snapshot of infra apiserver's etcd cluster.", "type": "bundle_runtime_note", "detail": "README states the package snapshots the infra apiserver etcd cluster.", }, { "path": APP_DELIVERY_ROOT / "infra-etcd-backup" / "base" / "etcd-backup.yaml", "needle": "--endpoints=https://etcd.infra.svc.cluster.local:2379", "type": "manifest_runtime_reference", "detail": "CronJob snapshots the infra etcd service and uploads the result to the backup bucket.", }, ], "lake-svc": [ { "path": APP_DELIVERY_ROOT / "lake-svc" / "envs" / "prod" / "lake-svc-service.yaml", "needle": "externalName:", "type": "externalname_proxy", "detail": "The prod package is only an ExternalName service pointing at a VPCE hostname.", } ], "rbac-management-service": [ { "path": EXTERNAL_REPO_ROOTS["account"] / ".pingcap" / "Dockerfile_rbac_tools", "needle": "rbac_management.py", "type": "source_build_definition", "detail": "The account repo builds the RBAC management tooling image from Dockerfile_rbac_tools.", }, { "path": EXTERNAL_REPO_ROOTS["account"] / ".github" / "workflows" / "update-rbac-prod.yaml", "needle": "dbaas/prod/rbac-management", "type": "source_release_workflow", "detail": "The prod update-rbac workflow pushes the rbac-management image consumed by app-delivery.", }, ], "dedicated-infra-apiserver": [ { "path": APP_DELIVERY_ROOT / "dedicated-infra-apiserver" / "base" / "deployment.yaml", "needle": "third-party/kube-apiserver", "type": "deployment_bundle_manifest", "detail": "The deployment bundle runs third-party kube-apiserver images rather than a first-party business service image.", }, { "path": EXTERNAL_REPO_ROOTS["infra-cd"] / "projects" / "apiserver" / "Pulumi.yaml", "needle": "name: apiserver", "type": "infra_stack_definition", "detail": "The infra-cd apiserver project owns the surrounding infra stack and wiring.", }, ], } EXTRA_MANIFEST_EDGES = [ { "from_object": "clinic", "to_object": "external-secrets-operator", "relation": "manifest_requires_addon_crd", "reference_type": "api_group_crd", "confidence": "high", "environments": ["dev", "prod-cloud", "staging"], "notes": "Clinic bundle workloads define ExternalSecret resources, normalized back to the generic external-secrets-operator addon object.", "evidence_sources": [ { "path": APP_DELIVERY_ROOT / "clinic" / "chaos-api" / "base" / "externalsecret.yaml", "needle": "apiVersion: external-secrets.io/v1", "snippet": "apiVersion: external-secrets.io/v1", "environment": "prod-cloud", }, { "path": APP_DELIVERY_ROOT / "dev" / "clinic" / "cloud-server" / "base" / "externalsecret.yaml", "needle": "apiVersion: external-secrets.io/v1", "snippet": "apiVersion: external-secrets.io/v1", "environment": "dev", }, { "path": APP_DELIVERY_ROOT / "staging" / "clinic" / "cloud-server" / "base" / "externalsecret.yaml", "needle": "apiVersion: external-secrets.io/v1", "snippet": "apiVersion: external-secrets.io/v1", "environment": "staging", }, ], }, { "from_object": "clinic", "to_object": "cluster-secret-store", "relation": "manifest_references_secret_store", "reference_type": "cluster_secret_store_ref", "confidence": "high", "environments": ["dev", "prod-cloud", "staging"], "notes": "Clinic bundle workloads reference ClusterSecretStore-backed ExternalSecret resources.", "evidence_sources": [ { "path": APP_DELIVERY_ROOT / "clinic" / "chaos-api" / "base" / "externalsecret.yaml", "needle": "kind: ClusterSecretStore", "snippet": "kind: ClusterSecretStore", "environment": "prod-cloud", }, { "path": APP_DELIVERY_ROOT / "dev" / "clinic" / "cloud-server" / "base" / "externalsecret.yaml", "needle": "kind: ClusterSecretStore", "snippet": "kind: ClusterSecretStore", "environment": "dev", }, { "path": APP_DELIVERY_ROOT / "staging" / "clinic" / "cloud-server" / "base" / "externalsecret.yaml", "needle": "kind: ClusterSecretStore", "snippet": "kind: ClusterSecretStore", "environment": "staging", }, ], }, { "from_object": "clinic", "to_object": "o11y-svc", "relation": "manifest_references_mapped_endpoint", "reference_type": "external_mapped_object_url", "confidence": "medium-high", "environments": ["dev-cloud", "prod-cloud", "staging"], "notes": "Clinic data-service configs point at observability global-server endpoints, normalized back to o11y-svc.", "evidence_sources": [ { "path": APP_DELIVERY_ROOT / "dev" / "clinic" / "clinic-data-service" / "envs" / "dev-cloud" / "configs" / "config.yaml", "needle": 'o11yGlobalServerUrl: "http://www.gs.us-west-2.aws.observability.tidbcloud.com"', "snippet": 'o11yGlobalServerUrl: "http://www.gs.us-west-2.aws.observability.tidbcloud.com"', "environment": "dev-cloud", }, { "path": APP_DELIVERY_ROOT / "clinic" / "clinic-data-service" / "envs" / "prod-cloud" / "configs" / "config.yaml", "needle": 'o11yGlobalServerUrl: "http://www.gs.us-west-2.aws.observability.tidbcloud.com"', "snippet": 'o11yGlobalServerUrl: "http://www.gs.us-west-2.aws.observability.tidbcloud.com"', "environment": "prod-cloud", }, { "path": APP_DELIVERY_ROOT / "staging" / "clinic" / "clinic-data-service" / "envs" / "staging" / "configs" / "config.yaml", "needle": 'o11yGlobalServerUrl: "http://www.gs.us-west-2.aws.observability.tidbcloud.com"', "snippet": 'o11yGlobalServerUrl: "http://www.gs.us-west-2.aws.observability.tidbcloud.com"', "environment": "staging", }, ], }, { "from_object": "nextgen-regional-control-plane", "to_object": "o11y-svc", "relation": "manifest_references_mapped_endpoint", "reference_type": "external_mapped_object_url", "confidence": "medium-high", "environments": ["dev", "prod", "staging"], "notes": "Next-gen o11y-agent-control-server configs point at observability global-server endpoints, normalized back to o11y-svc.", "evidence_sources": [ { "path": APP_DELIVERY_ROOT / "dev" / "nextgen-regional-control-plane" / "o11y-agent-control-server" / "envs" / "dev-aws-us-east-1-ng" / "config.yaml", "needle": 'openAPIAddress: "https://www.gs.us-west-2.aws.observability.tidbcloud.com"', "snippet": 'openAPIAddress: "https://www.gs.us-west-2.aws.observability.tidbcloud.com"', "environment": "dev", }, { "path": APP_DELIVERY_ROOT / "nextgen-regional-control-plane" / "o11y-agent-control-server" / "envs" / "prod-aws-us-west-2-ng" / "config.yaml", "needle": 'openAPIAddress: "https://www.gs.us-west-2.aws.observability.tidbcloud.com"', "snippet": 'openAPIAddress: "https://www.gs.us-west-2.aws.observability.tidbcloud.com"', "environment": "prod", }, { "path": APP_DELIVERY_ROOT / "staging" / "nextgen-regional-control-plane" / "o11y-agent-control-server" / "envs" / "staging-aws-us-west-2-ng-0711" / "config.yaml", "needle": 'openAPIAddress: "https://www.gs.us-west-2.aws.observability.tidbcloud.com"', "snippet": 'openAPIAddress: "https://www.gs.us-west-2.aws.observability.tidbcloud.com"', "environment": "staging", }, ], }, { "from_object": "devops-service", "to_object": "o11y-svc", "relation": "manifest_references_mapped_endpoint", "reference_type": "external_mapped_object_url", "confidence": "medium-high", "environments": ["dev", "prod", "staging"], "notes": "Devops-service tenant configs point at observability endpoints, normalized back to o11y-svc.", "evidence_sources": [ { "path": APP_DELIVERY_ROOT / "dev" / "devops-service" / "devops-server" / "envs" / "dev" / "configs" / "config.yaml", "needle": "o11yEndpoint: http://www.ds.us-west-2.aws.observability.tidbcloud.com", "snippet": "o11yEndpoint: http://www.ds.us-west-2.aws.observability.tidbcloud.com", "environment": "dev", }, { "path": APP_DELIVERY_ROOT / "devops-service" / "devops-server" / "envs" / "prod" / "configs" / "config.yaml", "needle": "o11yEndpoint: http://www.ds.us-west-2.aws.observability.tidbcloud.com", "snippet": "o11yEndpoint: http://www.ds.us-west-2.aws.observability.tidbcloud.com", "environment": "prod", }, { "path": APP_DELIVERY_ROOT / "staging" / "devops-service" / "devops-server" / "envs" / "staging" / "configs" / "config.yaml", "needle": "o11yEndpoint: http://www.ds.us-west-2.aws.observability.tidbcloud.com", "snippet": "o11yEndpoint: http://www.ds.us-west-2.aws.observability.tidbcloud.com", "environment": "staging", }, ], }, ] HIGH_LEVEL_LAYER_MEMBERSHIP = [ { "layer_id": "layer_1_platform_and_infra", "members": [ "obj:infra-api-crd", "obj:infra-api-providerconfig", "obj:infra-api-webhook", "obj:infra-webhook", "obj:external-secrets-operator", "obj:eks-provider", "obj:gke-provider", "obj:gardener-provider", ], }, { "layer_id": "layer_2_configuration_and_identity", "members": [ "obj:config-center", "obj:config-center-regional", "obj:config-manager", "obj:config-service", "obj:account-svc", "obj:authserver", "obj:billing-svc", "obj:core-svc", ], }, { "layer_id": "layer_3_global_control_plane", "members": [ "obj:tidb-mgmt-service", "obj:dedicated-global-service", "obj:nextgen-global-service", "obj:serverless-global-service", "obj:serverless-global-regional-proxy", "obj:dedicated-metering-server", "obj:dataservice-central-svc", "obj:dataflow-svc", ], }, { "layer_id": "layer_4_regional_execution_plane", "members": [ "obj:dedicated-regional-server", "obj:serverless-regional-service", "obj:config-center-regional", "obj:dedicated-ticdc-service", "obj:tcidm-service", "obj:dataservice-regional-svc", "obj:nextgen-regional-control-plane", "obj:o11y-agent-control-server", ], }, { "layer_id": "layer_5_async_dataflow_observability_ops", "members": [ "obj:dedicated-ticdc-service", "obj:dataflow-service-ng", "obj:serverless-metering", "obj:metrics-integration", "obj:o11y-svc", "obj:devops-service", "obj:infra-exporter", "obj:quota-exporter", ], }, ] HIGH_LEVEL_DEPLOYMENT_PHASES = [ { "phase": 0, "name": "cluster_and_platform_registration", "members": [ "obj:infra-api-crd", "obj:infra-api-providerconfig", "obj:infra-api-webhook", "obj:external-secrets-operator", "obj:eks-provider", "obj:gke-provider", "obj:gardener-provider", ], }, { "phase": 1, "name": "configuration_and_identity_base", "members": [ "obj:config-center", "obj:config-center-regional", "obj:config-manager", "obj:config-service", "obj:account-svc", "obj:authserver", "obj:core-svc", "obj:billing-svc", ], }, { "phase": 2, "name": "global_control_plane", "members": [ "obj:tidb-mgmt-service", "obj:dedicated-global-service", "obj:nextgen-global-service", "obj:serverless-global-service", "obj:serverless-global-regional-proxy", "obj:dataservice-central-svc", ], }, { "phase": 3, "name": "regional_execution_plane", "members": [ "obj:dedicated-regional-server", "obj:serverless-regional-service", "obj:tcidm-service", "obj:dedicated-ticdc-service", "obj:dataservice-regional-svc", "obj:o11y-agent-control-server", ], }, { "phase": 4, "name": "dataflow_metering_and_ops", "members": [ "obj:dataflow-svc", "obj:dataflow-service-ng", "obj:serverless-metering", "obj:metrics-integration", "obj:devops-service", "obj:infra-exporter", "obj:quota-exporter", ], }, ] HIGH_LEVEL_DEPENDENCY_EDGES = [ {"from": "obj:account-svc", "to": "obj:tidb-mgmt-service", "relation": "foundation_for", "confidence": "high", "evidence_chain": "chain_a"}, {"from": "obj:authserver", "to": "obj:tidb-mgmt-service", "relation": "foundation_for", "confidence": "high", "evidence_chain": "chain_a"}, {"from": "obj:core-svc", "to": "obj:tidb-mgmt-service", "relation": "foundation_for", "confidence": "high", "evidence_chain": "chain_a"}, {"from": "obj:tidb-mgmt-service", "to": "obj:dedicated-global-service", "relation": "delegates_to", "confidence": "high", "evidence_chain": "chain_a"}, {"from": "obj:dedicated-global-service", "to": "obj:dedicated-regional-server", "relation": "depends_on", "confidence": "high", "evidence_chain": "chain_a"}, {"from": "obj:tidb-mgmt-service", "to": "obj:serverless-global-service", "relation": "delegates_to", "confidence": "high", "evidence_chain": "chain_b"}, {"from": "obj:serverless-global-service", "to": "obj:serverless-regional-service", "relation": "delegates_to", "confidence": "high", "evidence_chain": "chain_b"}, {"from": "obj:serverless-global-service", "to": "obj:serverless-svc", "relation": "depends_on", "confidence": "high", "evidence_chain": "chain_b"}, {"from": "obj:serverless-global-service", "to": "obj:billing-svc", "relation": "depends_on", "confidence": "high", "evidence_chain": "chain_b"}, {"from": "obj:serverless-global-service", "to": "obj:account-svc", "relation": "depends_on", "confidence": "medium", "evidence_chain": "chain_b"}, {"from": "obj:nextgen-global-service", "to": "obj:cluster-ng-service", "relation": "delegates_to", "confidence": "medium", "evidence_chain": "chain_c"}, {"from": "obj:cluster-ng-service", "to": "obj:config-center", "relation": "depends_on", "confidence": "high", "evidence_chain": "chain_c"}, {"from": "obj:cluster-ng-service", "to": "obj:dedicated-metering-server", "relation": "depends_on", "confidence": "medium", "evidence_chain": "chain_c"}, {"from": "obj:config-center", "to": "obj:config-center-regional", "relation": "distributes_to", "confidence": "high", "evidence_chain": "chain_d"}, {"from": "obj:config-center", "to": "obj:dataflow-svc", "relation": "configures", "confidence": "high", "evidence_chain": "chain_d"}, {"from": "obj:config-center", "to": "obj:cluster-ng-service", "relation": "configures", "confidence": "high", "evidence_chain": "chain_d"}, {"from": "obj:dataservice-central-svc", "to": "obj:dataservice-regional-svc", "relation": "delegates_to", "confidence": "high", "evidence_chain": "chain_e"}, {"from": "obj:dataservice-central-svc", "to": "obj:core-svc", "relation": "depends_on", "confidence": "high", "evidence_chain": "chain_e"}, {"from": "obj:dataservice-central-svc", "to": "obj:account-svc", "relation": "depends_on", "confidence": "medium", "evidence_chain": "chain_e"}, {"from": "obj:dedicated-ticdc-service", "to": "obj:dataflow-svc", "relation": "adjacent_to", "confidence": "medium", "evidence_chain": "async_layer"}, ] HIGH_LEVEL_NOTES = [ "This graph captures high-signal topology anchors and dependency chains, not a full per-manifest DAG.", "Mapping edges come from app-delivery object mapping conclusions.", "Dependency edges come from the topology chains and recommended deployment ordering.", "Confidence values are normalized to high, medium-high, medium, medium-low, and low.", ] def load_json(path: Path) -> Any: return json.loads(path.read_text()) def write_json(path: Path, data: Any) -> None: path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") def path_to_label(path: Path) -> str: if path.is_relative_to(ROOT): return str(path.relative_to(ROOT)) if path.is_relative_to(APP_DELIVERY_ROOT): return f"app-delivery/{path.relative_to(APP_DELIVERY_ROOT)}" for repo_root in EXTERNAL_REPO_ROOTS.values(): if path.is_relative_to(repo_root.parent): return str(path.relative_to(repo_root.parent)) return str(path) def find_line(path: Path, needle: str) -> int | None: try: for idx, line in enumerate(path.read_text().splitlines(), start=1): if needle in line: return idx except FileNotFoundError: return None return None def slug(value: str) -> str: return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") def app_delivery_relative(label: str) -> str: return label.removeprefix("app-delivery/") def load_filtered_repos() -> list[str]: filtered = ROOT / ".local" / "review-history" / "repos-filtered.txt" result = [] for line in filtered.read_text().splitlines(): if not line.strip(): continue result.append(line.split()[0]) return sorted(set(result)) def build_manifest_edge_from_template(template: dict[str, Any]) -> dict[str, Any]: evidence_samples = [] for sample in template["evidence_sources"]: line = find_line(sample["path"], sample["needle"]) evidence_samples.append( { "file": path_to_label(sample["path"]), "line": line, "snippet": sample["snippet"], "environment": sample["environment"], } ) return { "from_object": template["from_object"], "to_object": template["to_object"], "relation": template["relation"], "reference_type": template["reference_type"], "confidence": template["confidence"], "environments": template["environments"], "evidence_samples": evidence_samples, "notes": template["notes"], } def recalculate_manifest_summary(manifest_data: dict[str, Any]) -> None: edges = manifest_data["dependency_edges"] manifest_data["summary"]["dependency_edge_count"] = len(edges) manifest_data["summary"]["unique_from_object_count"] = len({edge["from_object"] for edge in edges}) manifest_data["summary"]["unique_to_object_count"] = len({edge["to_object"] for edge in edges}) manifest_data["summary"]["reference_type_counts"] = dict(Counter(edge["reference_type"] for edge in edges)) manifest_data["summary"]["confidence_counts"] = dict(Counter(edge["confidence"] for edge in edges)) manifest_data["summary"]["relation_counts"] = dict(Counter(edge["relation"] for edge in edges)) def scan_applicationsets() -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] files = sorted( path for path in APP_DELIVERY_ROOT.rglob("app-registry/*.yaml") if not path.name.endswith(".bak") ) for path in files: rel = path.relative_to(APP_DELIVERY_ROOT) parts = rel.parts if parts[0] in {"dev", "staging"}: scope = parts[0] remainder = parts[1:] else: scope = "root" remainder = parts bundle = None if remainder[0] == "app-registry" else remainder[0] data = yaml.safe_load(path.read_text()) template = data.get("spec", {}).get("template", {}) source = template.get("spec", {}).get("source", {}) destination = template.get("spec", {}).get("destination", {}) placements: list[dict[str, Any]] = [] for generator in data.get("spec", {}).get("generators", []): list_generator = generator.get("list") if not list_generator: continue for element in list_generator.get("elements", []): env = str(element.get("env")) cluster = str(element.get("cluster")) namespace_template = destination.get("namespace", "") namespace = element.get("namespace") or namespace_template.replace("{{env}}", env) if "{{namespace}}" in namespace_template: namespace = namespace_template.replace("{{namespace}}", str(element.get("namespace", ""))) source_path = source.get("path", "").replace("{{env}}", env) placements.append( { "environment": env, "cluster": cluster, "namespace": namespace, "source_path": source_path, } ) records.append( { "kind": data.get("kind"), "scope": scope, "bundle": bundle, "registry_file": f"app-delivery/{rel}", "registry_object": path.stem, "applicationset_name": data.get("metadata", {}).get("name", path.stem), "metadata_name_line": find_line(path, f"name: {data.get('metadata', {}).get('name', path.stem)}"), "generated_app_name_template": template.get("metadata", {}).get("name"), "source_path_template": source.get("path"), "destination_name_template": destination.get("name"), "destination_namespace_template": destination.get("namespace"), "placements": placements, } ) return records def find_repo_catalog_evidence(repo_full_name: str) -> dict[str, Any] | None: repo_name = repo_full_name.split("/", 1)[1] repo_meta = ROOT / "knowledge" / "repos" / repo_name / "meta.json" if repo_meta.exists(): line = find_line(repo_meta, repo_full_name) or find_line(repo_meta, f'"name": "{repo_name}"') return { "type": "kb_repo_meta", "file": path_to_label(repo_meta), "line": line, "detail": "KB repo metadata confirms the mapped source repository exists locally in the knowledge base.", } for catalog in REPO_CATALOGS: line = find_line(catalog, f'"full_name": "{repo_full_name}"') or find_line(catalog, f'"name": "{repo_name}"') if line: return { "type": "repo_catalog_entry", "file": path_to_label(catalog), "line": line, "detail": "Repo catalog entry confirms the mapped repository or deployment repo exists in the KB inventory.", } return None def repo_from_source_family(source_family: str | None) -> str | None: if not source_family: return None match = re.search(r"(pingcap|tidbcloud|tikv)/([A-Za-z0-9._-]+)", source_family) if not match: return None return f"{match.group(1)}/{match.group(2)}" def build_mapping_evidence( mapping: dict[str, Any], applicationsets: list[dict[str, Any]], ) -> list[dict[str, Any]]: evidence: list[dict[str, Any]] = [] deployment_object = mapping["deployment_object"] top_level_matches = [ app for app in applicationsets if app["bundle"] is None and app["registry_object"] == deployment_object ] for app in top_level_matches: evidence.append( { "type": "app_registry_object", "file": app["registry_file"], "line": app["metadata_name_line"], "detail": f"{app['scope']} scope ApplicationSet for the deployment object.", } ) source_repo = mapping.get("source_repo") if source_repo: repo_evidence = find_repo_catalog_evidence(source_repo) if repo_evidence: evidence.append(repo_evidence) source_family_repo = repo_from_source_family(mapping.get("source_family")) if source_family_repo and not source_repo: repo_evidence = find_repo_catalog_evidence(source_family_repo) if repo_evidence: evidence.append(repo_evidence) for item in SPECIAL_MAPPING_EVIDENCE.get(deployment_object, []): line = find_line(item["path"], item["needle"]) evidence.append( { "type": item["type"], "file": path_to_label(item["path"]), "line": line, "detail": item["detail"], } ) deduped = [] seen = set() for item in evidence: key = (item["type"], item["file"], item.get("line"), item["detail"]) if key in seen: continue seen.add(key) deduped.append(item) return deduped def enrich_mapping_json( mapping_data: dict[str, Any], applicationsets: list[dict[str, Any]], ) -> dict[str, Any]: enriched = deepcopy(mapping_data) enriched.setdefault("scope", {}).setdefault( "notes", [], ) if "Each object includes a generated mapping_id plus evidence references that point back to registry files, repo metadata, or special-case manifests." not in enriched["scope"]["notes"]: enriched["scope"]["notes"].append( "Each object includes a generated mapping_id plus evidence references that point back to registry files, repo metadata, or special-case manifests." ) for mapping in enriched["objects"]: mapping["mapping_id"] = f"mapping:{slug(mapping['deployment_object'])}" mapping["evidence"] = build_mapping_evidence(mapping, applicationsets) confidence_counts = Counter(item["confidence"] for item in enriched["objects"]) mapping_type_counts = Counter(item["mapping_type"] for item in enriched["objects"]) enriched["summary"] = { "object_count": len(enriched["objects"]), "residual_ambiguity_count": len(enriched["residual_ambiguity"]), "mapping_type_counts": dict(mapping_type_counts), "confidence_counts": dict(confidence_counts), "objects_with_evidence": sum(1 for item in enriched["objects"] if item["evidence"]), } return enriched def enrich_manifest_graph(manifest_data: dict[str, Any]) -> dict[str, Any]: enriched = deepcopy(manifest_data) existing_keys = { (edge["from_object"], edge["to_object"], edge["relation"], edge["reference_type"]) for edge in enriched["dependency_edges"] } for template in EXTRA_MANIFEST_EDGES: key = ( template["from_object"], template["to_object"], template["relation"], template["reference_type"], ) if key not in existing_keys: enriched["dependency_edges"].append(build_manifest_edge_from_template(template)) existing_keys.add(key) recalculate_manifest_summary(enriched) if "Each dependency edge includes a stable edge_id, a derivation_rule, and normalized evidence location lists." not in enriched["notes"]: enriched["notes"].append( "Each dependency edge includes a stable edge_id, a derivation_rule, and normalized evidence location lists." ) for edge in enriched["dependency_edges"]: edge["edge_id"] = "edge:{from_object}:{relation}:{to_object}".format( from_object=slug(edge["from_object"]), relation=slug(edge["relation"]), to_object=slug(edge["to_object"]), ) edge["derivation_rule"] = MANIFEST_RULES.get( edge["reference_type"], "Derived from explicit manifest or config references normalized into deployment-object dependency edges.", ) edge["evidence_files"] = sorted({sample["file"] for sample in edge.get("evidence_samples", [])}) edge["evidence_locations"] = [ { "file": sample["file"], "line": sample["line"], "environment": sample.get("environment"), } for sample in edge.get("evidence_samples", []) ] enriched["summary"]["edge_id_count"] = len({edge["edge_id"] for edge in enriched["dependency_edges"]}) enriched["summary"]["edges_with_evidence_locations"] = sum( 1 for edge in enriched["dependency_edges"] if edge["evidence_locations"] ) return enriched def build_environment_matrix( applicationsets: list[dict[str, Any]], mapping_index: dict[str, dict[str, Any]], ) -> dict[str, Any]: records: list[dict[str, Any]] = [] bundle_counter = Counter() for app in applicationsets: top_level_object = app["bundle"] or app["registry_object"] top_level_mapping = mapping_index.get(top_level_object) if app["bundle"]: bundle_counter[app["bundle"]] += 1 for placement in app["placements"]: placement_id = "placement:{scope}:{top}:{component}:{env}:{cluster}:{namespace}".format( scope=slug(app["scope"]), top=slug(top_level_object), component=slug(app["registry_object"]), env=slug(placement["environment"]), cluster=slug(placement["cluster"]), namespace=slug(placement["namespace"]), ) records.append( { "placement_id": placement_id, "scope": app["scope"], "bundle": app["bundle"], "top_level_object": top_level_object, "registry_object": app["registry_object"], "applicationset_name": app["applicationset_name"], "registry_file": app["registry_file"], "environment": placement["environment"], "cluster": placement["cluster"], "namespace": placement["namespace"], "source_path": placement["source_path"], "mapping": { "top_level_deployment_object": top_level_object, "mapping_type": top_level_mapping.get("mapping_type") if top_level_mapping else None, "source_repo": top_level_mapping.get("source_repo") if top_level_mapping else None, "source_family": top_level_mapping.get("source_family") if top_level_mapping else None, "confidence": top_level_mapping.get("confidence") if top_level_mapping else None, "component_mapping_available": app["registry_object"] in mapping_index, }, } ) top_level_objects = sorted({app["registry_object"] for app in applicationsets if app["bundle"] is None}) mapped_top_level = sorted( obj for obj in top_level_objects if obj in mapping_index or obj in TOP_LEVEL_MAPPING_EXCLUSIONS ) missing_top_level = sorted( obj for obj in top_level_objects if obj not in mapping_index and obj not in TOP_LEVEL_MAPPING_EXCLUSIONS ) nested_components = sorted({app["registry_object"] for app in applicationsets if app["bundle"] is not None}) environments = sorted({record["environment"] for record in records}) clusters = sorted({record["cluster"] for record in records}) return { "date": str(date.today()), "generator": "construction/generate-deployment-derived-artifacts.py", "source_documents": [ "knowledge/deployment/app-delivery-object-mapping.json", "knowledge/deployment/deployment-environment-graph.json", ], "source_paths": [ "app-delivery/**/app-registry/*.yaml", ], "summary": { "applicationset_file_count": len(applicationsets), "placement_count": len(records), "top_level_object_count": len(top_level_objects), "mapped_top_level_object_count": len(mapped_top_level), "top_level_objects_without_mapping": missing_top_level, "nested_component_count": len(nested_components), "bundle_count": len(bundle_counter), "environment_count": len(environments), "cluster_count": len(clusters), "largest_bundles_by_component_files": [ {"bundle": bundle, "applicationset_files": count} for bundle, count in bundle_counter.most_common(10) ], }, "records": records, } def build_bundle_decomposition( applicationsets: list[dict[str, Any]], mapping_index: dict[str, dict[str, Any]], ) -> dict[str, Any]: bundle_records: dict[str, list[dict[str, Any]]] = defaultdict(list) for app in applicationsets: if app["bundle"]: bundle_records[app["bundle"]].append(app) bundles = [] for bundle_name in sorted(bundle_records): components: dict[str, dict[str, Any]] = {} placements_total = 0 for app in bundle_records[bundle_name]: component = components.setdefault( app["registry_object"], { "component": app["registry_object"], "scopes": set(), "registry_files": set(), "source_path_templates": set(), "environments": set(), "clusters": set(), "namespaces": set(), "placement_count": 0, }, ) component["scopes"].add(app["scope"]) component["registry_files"].add(app["registry_file"]) component["source_path_templates"].add(app["source_path_template"]) for placement in app["placements"]: component["environments"].add(placement["environment"]) component["clusters"].add(placement["cluster"]) component["namespaces"].add(placement["namespace"]) component["placement_count"] += 1 placements_total += 1 mapping = mapping_index.get(bundle_name) bundles.append( { "bundle": bundle_name, "mapping": { "mapping_type": mapping.get("mapping_type") if mapping else None, "source_repo": mapping.get("source_repo") if mapping else None, "source_family": mapping.get("source_family") if mapping else None, "confidence": mapping.get("confidence") if mapping else None, }, "component_count": len(components), "placement_count": placements_total, "scopes": sorted({scope for component in components.values() for scope in component["scopes"]}), "components": [ { "component": component["component"], "scopes": sorted(component["scopes"]), "registry_files": sorted(component["registry_files"]), "source_path_templates": sorted(filter(None, component["source_path_templates"])), "environments": sorted(component["environments"]), "clusters": sorted(component["clusters"]), "namespaces": sorted(component["namespaces"]), "placement_count": component["placement_count"], "component_has_direct_mapping": component["component"] in mapping_index, } for component in sorted(components.values(), key=lambda item: item["component"]) ], } ) return { "date": str(date.today()), "generator": "construction/generate-deployment-derived-artifacts.py", "source_paths": [ "app-delivery/**/app-registry/*.yaml", "knowledge/deployment/app-delivery-object-mapping.json", ], "summary": { "bundle_count": len(bundles), "component_count": sum(bundle["component_count"] for bundle in bundles), "placement_count": sum(bundle["placement_count"] for bundle in bundles), "bundles_with_top_level_mapping": sum( 1 for bundle in bundles if bundle["mapping"]["mapping_type"] is not None ), }, "bundles": bundles, } def build_top_level_environment_graph( applicationsets: list[dict[str, Any]], mapping_index: dict[str, dict[str, Any]], ) -> dict[str, Any]: top_level_apps = [ app for app in applicationsets if app["bundle"] is None and app["kind"] == "ApplicationSet" ] graph_apps = [] cluster_counter = Counter() envs = set() for app in top_level_apps: placements = [] for placement in app["placements"]: placements.append( { "env": placement["environment"], "cluster": placement["cluster"], } ) cluster_counter[placement["cluster"]] += 1 envs.add(placement["environment"]) graph_apps.append( { "scope": app["scope"], "registry_file": app_delivery_relative(app["registry_file"]), "registry_object": app["registry_object"], "applicationset_name": app["applicationset_name"], "generated_app_name_template": app["generated_app_name_template"], "source_path_template": app["source_path_template"], "destination": { "name_template": app["destination_name_template"], "namespace": app["destination_namespace_template"], }, "placement_mode": "list_generator", "placement_count": len(placements), "placements": placements, "mapping": { "deployment_object": app["registry_object"], "mapping_class": mapping_index.get(app["registry_object"], {}).get("mapping_class"), "mapping_type": mapping_index.get(app["registry_object"], {}).get("mapping_type"), "source_repo": mapping_index.get(app["registry_object"], {}).get("source_repo"), "source_family": mapping_index.get(app["registry_object"], {}).get("source_family"), "confidence": mapping_index.get(app["registry_object"], {}).get("confidence"), }, } ) unmapped = sorted( app["registry_object"] for app in top_level_apps if app["registry_object"] not in mapping_index and app["registry_object"] not in TOP_LEVEL_MAPPING_EXCLUSIONS ) return { "date": str(date.today()), "graph_version": "0.2", "source_document": "knowledge/deployment/app-delivery-object-mapping.json", "source_paths": ["app-delivery/**/app-registry/*.yaml"], "summary": { "applicationset_count": len(graph_apps), "placement_count": sum(app["placement_count"] for app in graph_apps), "unique_cluster_count": len(cluster_counter), "unique_env_count": len(envs), "scope_counts": dict(Counter(app["scope"] for app in graph_apps)), "mapped_applicationset_count": sum(1 for app in graph_apps if app["mapping"]["mapping_type"]), "unmapped_registry_objects": unmapped, }, "top_clusters_by_placement": [ {"cluster": cluster, "placement_count": count} for cluster, count in cluster_counter.most_common(20) ], "applicationsets": graph_apps, } def build_high_level_deployment_graph(mapping_index: dict[str, dict[str, Any]]) -> dict[str, Any]: anchor_object_ids = { member for layer in HIGH_LEVEL_LAYER_MEMBERSHIP for member in layer["members"] } anchor_object_ids.update( member for phase in HIGH_LEVEL_DEPLOYMENT_PHASES for member in phase["members"] ) anchor_object_ids.update(edge["from"] for edge in HIGH_LEVEL_DEPENDENCY_EDGES) anchor_object_ids.update(edge["to"] for edge in HIGH_LEVEL_DEPENDENCY_EDGES) anchor_objects = sorted(obj_id.removeprefix("obj:") for obj_id in anchor_object_ids if obj_id.startswith("obj:")) repo_nodes = [] mapping_edges = [] for deployment_object in anchor_objects: mapping = mapping_index.get(deployment_object) if not mapping or not mapping.get("source_repo"): continue repo_id = f"repo:{mapping['source_repo']}" repo_nodes.append( { "id": repo_id, "kind": "source_repo", "label": mapping["source_repo"], } ) mapping_edges.append( { "from": f"obj:{deployment_object}", "to": repo_id, "relation": "implemented_by", "confidence": mapping["confidence"], } ) object_nodes = [ { "id": f"obj:{deployment_object}", "kind": "deployment_object", "label": deployment_object, } for deployment_object in anchor_objects ] nodes = object_nodes + sorted({json.dumps(node, sort_keys=True): node for node in repo_nodes}.values(), key=lambda node: node["id"]) return { "date": str(date.today()), "graph_version": "0.2", "scope": "high_level_deployment_topology", "source_documents": [ "knowledge/deployment/app-delivery-object-mapping.json", "knowledge/deployment/deployment-manifest-graph.json", "knowledge/deployment/service-deployment-topology.md", ], "notes": HIGH_LEVEL_NOTES, "nodes": nodes, "mapping_edges": mapping_edges, "dependency_edges": HIGH_LEVEL_DEPENDENCY_EDGES, "layer_membership": HIGH_LEVEL_LAYER_MEMBERSHIP, "deployment_phases": HIGH_LEVEL_DEPLOYMENT_PHASES, "summary": { "node_count": len(nodes), "mapping_edge_count": len(mapping_edges), "dependency_edge_count": len(HIGH_LEVEL_DEPENDENCY_EDGES), "layer_count": len(HIGH_LEVEL_LAYER_MEMBERSHIP), "deployment_phase_count": len(HIGH_LEVEL_DEPLOYMENT_PHASES), }, } def build_validation_report( mapping_data: dict[str, Any], manifest_data: dict[str, Any], top_level_environment_graph: dict[str, Any], high_level_graph: dict[str, Any], environment_matrix: dict[str, Any], bundle_decomposition: dict[str, Any], filtered_repos: list[str], ) -> dict[str, Any]: mapping_index = {item["deployment_object"]: item for item in mapping_data["objects"]} top_level_objects = { record["registry_object"] for record in environment_matrix["records"] if record["bundle"] is None } top_level_missing = sorted( obj for obj in top_level_objects if obj not in mapping_index and obj not in TOP_LEVEL_MAPPING_EXCLUSIONS ) known_objects = set(mapping_index) unknown_from = sorted( {edge["from_object"] for edge in manifest_data["dependency_edges"] if edge["from_object"] not in known_objects} ) unknown_to = sorted( {edge["to_object"] for edge in manifest_data["dependency_edges"] if edge["to_object"] not in known_objects} ) leaked_repos = [] filtered_strings = set(filtered_repos) for item in mapping_data["objects"]: source_repo = item.get("source_repo") source_family = item.get("source_family", "") or "" if source_repo and source_repo.split("/", 1)[1] in filtered_strings: leaked_repos.append(source_repo) for filtered in filtered_strings: if f"/{filtered}" in source_family or source_family == filtered: leaked_repos.append(source_family) mapping_ids = [item["mapping_id"] for item in mapping_data["objects"]] edge_ids = [edge["edge_id"] for edge in manifest_data["dependency_edges"]] expected_bundle_names = { record["bundle"] for record in environment_matrix["records"] if record["bundle"] is not None } actual_bundle_names = {bundle["bundle"] for bundle in bundle_decomposition["bundles"]} top_level_matrix_records = [record for record in environment_matrix["records"] if record["bundle"] is None] graph_objects = { node["label"] for node in high_level_graph["nodes"] if node["kind"] == "deployment_object" } unknown_high_level_objects = sorted( obj for obj in graph_objects if obj not in mapping_index and obj not in {"cluster-ng-service", "infra-webhook"} ) checks = [ { "name": "top_level_registry_objects_have_mapping", "status": "pass" if not top_level_missing else "warn", "details": { "excluded_objects": sorted(TOP_LEVEL_MAPPING_EXCLUSIONS), "missing_objects": top_level_missing, }, }, { "name": "manifest_edges_reference_known_objects", "status": "pass" if not unknown_from and not unknown_to else "warn", "details": { "unknown_from_objects": unknown_from, "unknown_to_objects": unknown_to, }, }, { "name": "filtered_repos_do_not_leak_into_mapping_outputs", "status": "pass" if not leaked_repos else "fail", "details": { "filtered_repos": filtered_repos, "leaked_references": sorted(set(leaked_repos)), }, }, { "name": "mapping_ids_are_unique", "status": "pass" if len(mapping_ids) == len(set(mapping_ids)) else "fail", "details": { "mapping_id_count": len(mapping_ids), "unique_mapping_id_count": len(set(mapping_ids)), }, }, { "name": "edge_ids_are_unique", "status": "pass" if len(edge_ids) == len(set(edge_ids)) else "fail", "details": { "edge_id_count": len(edge_ids), "unique_edge_id_count": len(set(edge_ids)), }, }, { "name": "bundle_decomposition_covers_nested_applicationsets", "status": "pass" if expected_bundle_names == actual_bundle_names else "warn", "details": { "expected_bundles": sorted(expected_bundle_names), "actual_bundles": sorted(actual_bundle_names), }, }, { "name": "top_level_environment_graph_matches_matrix_projection", "status": "pass" if top_level_environment_graph["summary"]["applicationset_count"] == len(top_level_environment_graph["applicationsets"]) and top_level_environment_graph["summary"]["placement_count"] == len(top_level_matrix_records) else "warn", "details": { "environment_graph_applicationsets": top_level_environment_graph["summary"]["applicationset_count"], "top_level_environment_graph_entries": len(top_level_environment_graph["applicationsets"]), "environment_graph_placements": top_level_environment_graph["summary"]["placement_count"], "top_level_matrix_placements": len(top_level_matrix_records), }, }, { "name": "high_level_graph_objects_are_expected", "status": "pass" if not unknown_high_level_objects else "warn", "details": { "allowed_unmapped_objects": ["cluster-ng-service", "infra-webhook"], "unexpected_objects": unknown_high_level_objects, }, }, ] return { "date": str(date.today()), "generator": "construction/generate-deployment-derived-artifacts.py", "summary": { "check_count": len(checks), "pass_count": sum(1 for check in checks if check["status"] == "pass"), "warn_count": sum(1 for check in checks if check["status"] == "warn"), "fail_count": sum(1 for check in checks if check["status"] == "fail"), }, "checks": checks, } def main() -> None: applicationsets = scan_applicationsets() mapping_path = DEPLOYMENT_DIR / "app-delivery-object-mapping.json" manifest_path = DEPLOYMENT_DIR / "deployment-manifest-graph.json" mapping_data = enrich_mapping_json(load_json(mapping_path), applicationsets) manifest_data = enrich_manifest_graph(load_json(manifest_path)) mapping_index = {item["deployment_object"]: item for item in mapping_data["objects"]} top_level_environment_graph = build_top_level_environment_graph(applicationsets, mapping_index) high_level_graph = build_high_level_deployment_graph(mapping_index) environment_matrix = build_environment_matrix(applicationsets, mapping_index) bundle_decomposition = build_bundle_decomposition(applicationsets, mapping_index) validation_report = build_validation_report( mapping_data, manifest_data, top_level_environment_graph, high_level_graph, environment_matrix, bundle_decomposition, load_filtered_repos(), ) write_json(mapping_path, mapping_data) write_json(manifest_path, manifest_data) write_json(DEPLOYMENT_DIR / "deployment-environment-graph.json", top_level_environment_graph) write_json(DEPLOYMENT_DIR / "deployment-graph.json", high_level_graph) write_json(DEPLOYMENT_DIR / "deployment-object-environment-matrix.json", environment_matrix) write_json(DEPLOYMENT_DIR / "deployment-bundle-decomposition.json", bundle_decomposition) write_json(DEPLOYMENT_DIR / "deployment-validation-report.json", validation_report) if __name__ == "__main__": main()