ci: add dependency-aware main fastlane [CI SKIP]

ci/main-fastlane.py

@@ -0,0 +1,209 @@
+#!/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 raw:
+        try:
+            value = json.loads(raw)
+            if isinstance(value, list):
+                return [str(path).replace("\\", "/") for path in value]
+        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()