401 lines
13 KiB
Python
401 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import difflib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
try:
|
|
import yaml
|
|
except Exception as exc: # noqa: BLE001
|
|
raise SystemExit(
|
|
f"需要 PyYAML。请使用: uvx --with pyyaml python scripts/gitops_v2/render_shadow_diff.py\n{exc}"
|
|
)
|
|
|
|
from profile_matrix import (
|
|
get_profile_payload,
|
|
get_profile_registry_host,
|
|
get_profile_service_map_path,
|
|
load_profile_matrix,
|
|
resolve_target_profile,
|
|
)
|
|
|
|
|
|
JsonObject = dict[str, object]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ComponentUpdate:
|
|
service_name: str
|
|
image_ref: str
|
|
digest: str
|
|
overlay_rel: str
|
|
image_name: str
|
|
deploy_image: str
|
|
|
|
|
|
def ensure_mapping(value: object, label: str) -> JsonObject:
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{label} 必须是对象")
|
|
return cast(JsonObject, value)
|
|
|
|
|
|
def ensure_list(value: object, label: str) -> list[object]:
|
|
if not isinstance(value, list):
|
|
raise ValueError(f"{label} 必须是数组")
|
|
return value
|
|
|
|
|
|
def get_required_str(payload: JsonObject, key: str) -> str:
|
|
value = payload.get(key)
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ValueError(f"{key} 必须是非空字符串")
|
|
return value.strip()
|
|
|
|
|
|
def read_json_file(path: Path) -> JsonObject:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
return ensure_mapping(payload, str(path))
|
|
|
|
|
|
def read_yaml_file(path: Path) -> JsonObject:
|
|
payload = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
return ensure_mapping(payload, str(path))
|
|
|
|
|
|
def render_yaml(payload: JsonObject) -> str:
|
|
return yaml.safe_dump(payload, sort_keys=False, allow_unicode=True)
|
|
|
|
|
|
def repo_slug(repo_name: str) -> str:
|
|
return repo_name.replace("/", "__")
|
|
|
|
|
|
def digest_from_image(image_ref: str) -> str:
|
|
if "@sha256:" not in image_ref:
|
|
raise ValueError(f"image 缺少 digest: {image_ref}")
|
|
return image_ref.rsplit("@", 1)[1]
|
|
|
|
|
|
def image_repo(image_ref: str) -> str:
|
|
return image_ref.split("@", 1)[0]
|
|
|
|
|
|
def image_registry_host(image_repository: str) -> str:
|
|
first_segment = image_repository.split("/", 1)[0].strip().lower()
|
|
if not first_segment:
|
|
return ""
|
|
if "." in first_segment or ":" in first_segment or first_segment == "localhost":
|
|
return first_segment.rstrip("/")
|
|
return ""
|
|
|
|
|
|
def preferred_deploy_image(
|
|
target_payload: JsonObject, image_ref: str, profile_payload: JsonObject
|
|
) -> str:
|
|
source_image_repo = image_repo(image_ref)
|
|
deploy_image = str(target_payload.get("deployImage", "")).strip()
|
|
harbor_image = str(target_payload.get("harborImage", "")).strip()
|
|
image_name = str(target_payload.get("kustomizeImageName", "")).strip()
|
|
deploy_registry_host = get_profile_registry_host(profile_payload, "deploy")
|
|
|
|
if (
|
|
not deploy_image
|
|
and deploy_registry_host
|
|
and image_registry_host(source_image_repo) == deploy_registry_host
|
|
):
|
|
deploy_image = source_image_repo
|
|
if not deploy_image and harbor_image:
|
|
deploy_image = harbor_image
|
|
if not deploy_image and source_image_repo:
|
|
deploy_image = source_image_repo
|
|
if not deploy_image and image_name:
|
|
deploy_image = image_name
|
|
if not deploy_image:
|
|
raise ValueError("缺少 deploy image，无法解析目标仓库")
|
|
return deploy_image
|
|
|
|
|
|
def update_overlay(
|
|
overlay_payload: JsonObject,
|
|
image_name: str,
|
|
deploy_image: str,
|
|
digest: str,
|
|
) -> JsonObject:
|
|
updated = copy.deepcopy(overlay_payload)
|
|
images_value = updated.get("images", [])
|
|
images = ensure_list(images_value, "images")
|
|
|
|
target_index: int | None = None
|
|
for index, raw_item in enumerate(images):
|
|
item = ensure_mapping(raw_item, f"images[{index}]")
|
|
existing_name = str(item.get("name", "")).strip()
|
|
existing_new_name = str(item.get("newName", "")).strip()
|
|
if image_name in {existing_name, existing_new_name} or deploy_image in {
|
|
existing_name,
|
|
existing_new_name,
|
|
}:
|
|
target_index = index
|
|
break
|
|
|
|
target_item: JsonObject
|
|
if target_index is None:
|
|
target_item = {}
|
|
else:
|
|
target_item = ensure_mapping(
|
|
images[target_index], f"images[{target_index}]"
|
|
).copy()
|
|
|
|
target_item["name"] = image_name
|
|
target_item["newName"] = deploy_image
|
|
target_item["digest"] = digest
|
|
if "newTag" in target_item:
|
|
del target_item["newTag"]
|
|
|
|
if target_index is None:
|
|
images.append(target_item)
|
|
else:
|
|
images[target_index] = target_item
|
|
updated["images"] = images
|
|
return updated
|
|
|
|
|
|
def resolve_target(
|
|
service_map: JsonObject, service_name: str, env_name: str
|
|
) -> JsonObject:
|
|
services = ensure_mapping(service_map.get("services", {}), "services")
|
|
service_payload = ensure_mapping(
|
|
services.get(service_name, {}), f"services.{service_name}"
|
|
)
|
|
envs = ensure_mapping(
|
|
service_payload.get("envs", {}), f"services.{service_name}.envs"
|
|
)
|
|
return ensure_mapping(
|
|
envs.get(env_name, {}), f"services.{service_name}.envs.{env_name}"
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Render shadow diff from DeployResolution"
|
|
)
|
|
parser.add_argument(
|
|
"--repo-root", type=Path, default=Path("."), help="ljwx-deploy root"
|
|
)
|
|
parser.add_argument(
|
|
"--state-root",
|
|
type=Path,
|
|
default=Path("/root/codes/ljwx-release-state"),
|
|
help="ljwx-release-state root",
|
|
)
|
|
parser.add_argument(
|
|
"--service-map",
|
|
type=Path,
|
|
default=Path("release/services.yaml"),
|
|
help="legacy fallback service map path relative to deploy repo root",
|
|
)
|
|
parser.add_argument(
|
|
"--profile-matrix",
|
|
type=Path,
|
|
default=Path("platform/assembly/env-matrix.yaml"),
|
|
help="profile matrix path relative to deploy repo root",
|
|
)
|
|
parser.add_argument(
|
|
"--resolution-file",
|
|
type=Path,
|
|
default=Path(""),
|
|
help="single resolution file; defaults to all state resolutions",
|
|
)
|
|
parser.add_argument(
|
|
"--out-dir",
|
|
type=Path,
|
|
default=Path(""),
|
|
help="optional output directory for *.diff files",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def load_service_map_for_target(
|
|
repo_root: Path,
|
|
matrix_payload: JsonObject,
|
|
cache: dict[str, JsonObject],
|
|
target_payload: JsonObject,
|
|
fallback_service_map: JsonObject,
|
|
) -> tuple[str, JsonObject, JsonObject]:
|
|
env_name = get_required_str(target_payload, "env")
|
|
requested_profile = str(target_payload.get("profile", "")).strip()
|
|
profile_name = resolve_target_profile(matrix_payload, env_name, requested_profile)
|
|
profile_payload = get_profile_payload(matrix_payload, profile_name)
|
|
service_map_rel = get_profile_service_map_path(profile_payload)
|
|
service_map = cache.get(service_map_rel)
|
|
if service_map is None:
|
|
service_map_path = repo_root / service_map_rel
|
|
if service_map_path.exists():
|
|
service_map = read_yaml_file(service_map_path)
|
|
cache[service_map_rel] = service_map
|
|
else:
|
|
raise ValueError(
|
|
f"profile `{profile_name}` 对应的 service map 不存在: {service_map_rel}"
|
|
)
|
|
return profile_name, profile_payload, service_map
|
|
|
|
|
|
def load_resolution_files(state_root: Path, explicit_path: Path) -> list[Path]:
|
|
if explicit_path != Path(""):
|
|
return [explicit_path]
|
|
return sorted(
|
|
path
|
|
for path in (state_root / "resolutions").rglob("*.json")
|
|
if path.name != ".gitkeep"
|
|
)
|
|
|
|
|
|
def bundle_components(bundle_payload: JsonObject) -> list[JsonObject]:
|
|
components = ensure_list(bundle_payload.get("components", []), "components")
|
|
if not components:
|
|
raise ValueError("bundle.components 不能为空")
|
|
return [
|
|
ensure_mapping(item, f"components[{index}]")
|
|
for index, item in enumerate(components)
|
|
]
|
|
|
|
|
|
def build_component_updates(
|
|
service_map: JsonObject,
|
|
profile_payload: JsonObject,
|
|
bundle_payload: JsonObject,
|
|
env_name: str,
|
|
) -> list[ComponentUpdate]:
|
|
updates: list[ComponentUpdate] = []
|
|
for component in bundle_components(bundle_payload):
|
|
service_name = get_required_str(component, "name")
|
|
image_ref = get_required_str(component, "image")
|
|
target_payload = resolve_target(service_map, service_name, env_name)
|
|
updates.append(
|
|
ComponentUpdate(
|
|
service_name=service_name,
|
|
image_ref=image_ref,
|
|
digest=digest_from_image(image_ref),
|
|
overlay_rel=get_required_str(target_payload, "overlayPath"),
|
|
image_name=get_required_str(target_payload, "kustomizeImageName"),
|
|
deploy_image=preferred_deploy_image(
|
|
target_payload, image_ref, profile_payload
|
|
),
|
|
)
|
|
)
|
|
return updates
|
|
|
|
|
|
def render_single_diff(
|
|
repo_root: Path,
|
|
state_root: Path,
|
|
matrix_payload: JsonObject,
|
|
service_map_cache: dict[str, JsonObject],
|
|
fallback_service_map: JsonObject,
|
|
resolution_path: Path,
|
|
out_dir: Path,
|
|
) -> str:
|
|
resolution = read_json_file(resolution_path)
|
|
decision = ensure_mapping(resolution.get("decision", {}), "decision")
|
|
if get_required_str(decision, "status") != "resolved":
|
|
return f"跳过 {resolution_path.name}: 决议状态不是 resolved"
|
|
|
|
repo_name = get_required_str(resolution, "repo")
|
|
bundle_id = get_required_str(resolution, "bundleId")
|
|
target = ensure_mapping(resolution.get("target", {}), "target")
|
|
env_name = get_required_str(target, "env")
|
|
profile_name, profile_payload, service_map = load_service_map_for_target(
|
|
repo_root,
|
|
matrix_payload,
|
|
service_map_cache,
|
|
target,
|
|
fallback_service_map,
|
|
)
|
|
bundle_path = state_root / "bundles" / repo_slug(repo_name) / f"{bundle_id}.json"
|
|
bundle = read_json_file(bundle_path)
|
|
component_updates = build_component_updates(
|
|
service_map, profile_payload, bundle, env_name
|
|
)
|
|
|
|
diffs: list[str] = []
|
|
updates_by_overlay: dict[str, list[ComponentUpdate]] = {}
|
|
for update in component_updates:
|
|
updates_by_overlay.setdefault(update.overlay_rel, []).append(update)
|
|
|
|
resolution_id = get_required_str(resolution, "resolutionId")
|
|
if out_dir != Path(""):
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
for overlay_rel, overlay_updates in sorted(updates_by_overlay.items()):
|
|
overlay_path = repo_root / overlay_rel
|
|
current_overlay = read_yaml_file(overlay_path)
|
|
rendered_overlay = copy.deepcopy(current_overlay)
|
|
for update in overlay_updates:
|
|
rendered_overlay = update_overlay(
|
|
rendered_overlay,
|
|
update.image_name,
|
|
update.deploy_image,
|
|
update.digest,
|
|
)
|
|
|
|
current_text = render_yaml(current_overlay).splitlines(keepends=True)
|
|
rendered_text = render_yaml(rendered_overlay).splitlines(keepends=True)
|
|
diff = "".join(
|
|
difflib.unified_diff(
|
|
current_text,
|
|
rendered_text,
|
|
fromfile=str(overlay_path),
|
|
tofile=f"{overlay_path} (shadow)",
|
|
)
|
|
)
|
|
if diff:
|
|
diffs.append(diff)
|
|
if out_dir != Path(""):
|
|
diff_name = overlay_rel.replace("/", "__")
|
|
(out_dir / f"{resolution_id}__{diff_name}.diff").write_text(
|
|
diff, encoding="utf-8"
|
|
)
|
|
|
|
if not diffs:
|
|
return f"{resolution_id} ({profile_name}): 无差异"
|
|
return f"{resolution_id} ({profile_name}):\n" + "\n".join(diffs)
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
repo_root = args.repo_root.resolve()
|
|
state_root = args.state_root.resolve()
|
|
matrix_payload = load_profile_matrix(repo_root, args.profile_matrix)
|
|
fallback_service_map = read_yaml_file(repo_root / args.service_map)
|
|
service_map_cache: dict[str, JsonObject] = {}
|
|
resolution_files = load_resolution_files(state_root, args.resolution_file)
|
|
|
|
if not resolution_files:
|
|
print("未找到 resolution 文件")
|
|
return 0
|
|
|
|
rendered_outputs: list[str] = []
|
|
for resolution_path in resolution_files:
|
|
rendered_outputs.append(
|
|
render_single_diff(
|
|
repo_root=repo_root,
|
|
state_root=state_root,
|
|
matrix_payload=matrix_payload,
|
|
service_map_cache=service_map_cache,
|
|
fallback_service_map=fallback_service_map,
|
|
resolution_path=resolution_path,
|
|
out_dir=args.out_dir,
|
|
)
|
|
)
|
|
|
|
print("\n\n".join(rendered_outputs))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|