#!/usr/bin/env python3 """Find impacted OCI images from changed paths and stage NativeAOT contexts.""" from __future__ import annotations import argparse import json import os import shutil import subprocess from pathlib import Path from xml.etree import ElementTree ROOT = Path(__file__).resolve().parents[1] BACKEND = ROOT / "backend" IMAGE_ROOTS = { "backend": [BACKEND / "src/Monitor.Backend/Monitor.Backend.csproj"], "agent": [ BACKEND / "agent/src/Monitor.Agent/Monitor.Agent.csproj", BACKEND / "fleet/tools/fleet-enroll/fleet-enroll.csproj", ], "probe": [BACKEND / "probe/src/Monitor.Probe/Monitor.Probe.csproj"], } GLOBAL_BACKEND_FILES = { "Directory.Build.props", "Directory.Build.targets", "Directory.Packages.props", "global.json", "NuGet.Config", "NuGet.config", } IGNORED_PARTS = {".git", ".release", "artifacts", "bin", "obj", "TestResults", "tests"} def read_changed_files() -> list[str]: raw = os.environ.get("CI_PIPELINE_FILES") if os.environ.get("FASTLANE_FILES_FROM_PR_API") == "true": value = json.loads(raw or "[]") return [str(path).replace("\\", "/") for path in value] if raw: try: value = json.loads(raw) if isinstance(value, list): changed = [str(path).replace("\\", "/") for path in value] if changed: return changed except json.JSONDecodeError: return [path.strip().replace("\\", "/") for path in raw.split(",") if path.strip()] previous = os.environ.get("CI_PREV_COMMIT_SHA") try: base = previous if previous else "HEAD^" return subprocess.check_output( ["git", "diff", "--name-only", base, "HEAD"], cwd=ROOT, text=True, ).splitlines() except (OSError, subprocess.CalledProcessError): return [] def project_graph() -> tuple[dict[Path, set[Path]], dict[Path, Path]]: projects = [path.resolve() for path in BACKEND.rglob("*.csproj") if not (IGNORED_PARTS & set(path.parts))] graph: dict[Path, set[Path]] = {} directories: dict[Path, Path] = {} for project in projects: graph[project] = set() directories[project.parent] = project try: document = ElementTree.parse(project) except ElementTree.ParseError: continue for reference in document.iter(): if reference.tag.rsplit("}", 1)[-1] == "ProjectReference": include = reference.attrib.get("Include") if include: graph[project].add((project.parent / include.replace("\\", os.sep)).resolve()) return graph, directories def reachable_projects(graph: dict[Path, set[Path]], roots: list[Path]) -> set[Path]: found: set[Path] = set() pending = [path.resolve() for path in roots] while pending: project = pending.pop() if project in found: continue found.add(project) pending.extend(graph.get(project, ())) return found def is_within(path: Path, directory: Path) -> bool: try: path.relative_to(directory) return True except ValueError: return False def affected_images(changed: list[str]) -> list[str]: graph, project_dirs = project_graph() image_dependencies = { name: reachable_projects(graph, roots) for name, roots in IMAGE_ROOTS.items() } selected: set[str] = set() def select_for_project(project: Path) -> None: for image, dependencies in image_dependencies.items(): if project in dependencies: selected.add(image) project_directories = sorted(project_dirs, key=lambda path: len(path.parts), reverse=True) container_map = { "packaging/container/Backend.Containerfile": "backend", "packaging/container/Agent.Containerfile": "agent", "packaging/container/Probe.Containerfile": "probe", "packaging/container/Agent.Fastlane.Containerfile": "agent", "packaging/container/Probe.Fastlane.Containerfile": "probe", } for filename in changed: normalized = filename.replace("\\", "/") path = (ROOT / normalized).resolve() if normalized in container_map: selected.add(container_map[normalized]) continue if normalized.startswith("ci/") or normalized.startswith(".woodpecker/"): if not normalized.endswith(".md"): selected.update(IMAGE_ROOTS) continue if any(part in IGNORED_PARTS for part in Path(normalized).parts): continue if not normalized.startswith("backend/"): continue if "/README" in normalized or normalized.endswith(".md"): continue if path.name in GLOBAL_BACKEND_FILES and path.parent == BACKEND: selected.update(IMAGE_ROOTS) continue matching = [directory for directory in project_directories if is_within(path, directory)] if matching: select_for_project(project_dirs[matching[0]]) continue # A props/targets file above one or more projects affects each project # below it and every image that consumes those projects. descendants = [project for directory, project in project_dirs.items() if is_within(directory, path.parent)] if path.suffix in {".props", ".targets"} and descendants: for project in descendants: select_for_project(project) elif normalized.startswith("backend/agent/"): selected.add("agent") elif normalized.startswith("backend/probe/"): selected.add("probe") elif path.is_file() and path.suffix in {".cs", ".csproj", ".props", ".targets", ".json"}: selected.update(IMAGE_ROOTS) return [name for name in ("backend", "agent", "probe") if name in selected] def stage_context(component: str, destination: Path) -> None: containerfiles = { "agent": ROOT / "packaging/container/Agent.Fastlane.Containerfile", "probe": ROOT / "packaging/container/Probe.Fastlane.Containerfile", } if component not in containerfiles: raise SystemExit(f"NativeAOT context is not available for {component}") if destination.exists(): shutil.rmtree(destination) restore = destination / "restore/backend" source = destination / "source/backend" restore.mkdir(parents=True) source.mkdir(parents=True) shutil.copy2(containerfiles[component], destination / "Containerfile") for path in BACKEND.rglob("*"): if not path.is_file() or IGNORED_PARTS.intersection(path.parts): continue relative = path.relative_to(BACKEND) if path.suffix in {".csproj", ".props", ".targets", ".slnx"} or path.name in GLOBAL_BACKEND_FILES or path.name == "packages.lock.json": target = restore / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(path, target) target = source / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(path, target) (destination / "config").mkdir() (destination / "config/.keep").write_text("", encoding="utf-8") def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--stage", choices=("agent", "probe")) parser.add_argument("--destination", type=Path) args = parser.parse_args() if args.stage: if args.destination is None: parser.error("--destination is required with --stage") stage_context(args.stage, args.destination) return for image in affected_images(read_changed_files()): print(image) if __name__ == "__main__": main()