argocd-revision-repo-url-patcher.md

Argo CD Application history repoURL patch helper

Small uv run Python utility to inspect Argo CD Application resources and patch stale status.history[*].source.repoURL entries.

It shows a unified diff first, asks for confirmation, then applies the patch via kubectl.

Why this exists

After migrating a repo URL, Argo CD can still hold history entries that reference the old repo/revisions. This can lead to sync errors like not our ref / unadvertised object while Argo still tries to resolve old commits.

Related discussion/use case: argoproj/argo-cd#25455 (comment)

Example usage

Replace old URL in matching history entries:

uv run argocd-revision-repo-url-patcher.py \
  --context fracture \
  --application cloudnativepg \
  --old-url 'git@github.com:myorg/gitops.git' \
  --new-url 'git@gitlab.com:myorg/infrastructure.git' \
  --color always

Drop matching revisions entirely instead of rewriting repoURL:

uv run argocd-revision-repo-url-patcher.py \
  --application cloudnativepg \
  --old-url 'git@github.com:myorg/gitops.git' \
  --drop-revisions

--context is optional; if omitted, current kubectl context is used.

argocd-revision-repo-url-patcher.py

#!/usr/bin/env python3
"""
Codex-generated utility from a prompt to systematically update Argo CD
Application `status.history[*].source.repoURL` values in a cluster context.

Workflow:
1. Discover all Applications (or a single one via `--application`).
2. Build and show unified diffs for matching history repo URLs.
3. Ask for confirmation, then apply JSON patches to the Application resource.
"""

import argparse
import difflib
import json
import os
import subprocess
import sys
from typing import Any


def run(cmd: list[str]) -> str:
    p = subprocess.run(cmd, text=True, capture_output=True)
    if p.returncode != 0:
        print(f"ERROR: command failed ({p.returncode}): {' '.join(cmd)}", file=sys.stderr)
        if p.stdout:
            print(p.stdout, file=sys.stderr)
        if p.stderr:
            print(p.stderr, file=sys.stderr)
        sys.exit(p.returncode)
    return p.stdout


def json_dump(value: Any) -> str:
    return json.dumps(value, indent=2, sort_keys=True) + "\n"


def use_color(mode: str) -> bool:
    if mode == "always":
        return True
    if mode == "never":
        return False
    return sys.stdout.isatty() and "NO_COLOR" not in os.environ


def colorize_diff_line(line: str, color: bool) -> str:
    if not color:
        return line
    if line.startswith("+++ ") or line.startswith("--- "):
        return f"\033[1m{line}\033[0m"
    if line.startswith("@@"):
        return f"\033[36m{line}\033[0m"
    if line.startswith("+") and not line.startswith("+++ "):
        return f"\033[32m{line}\033[0m"
    if line.startswith("-") and not line.startswith("--- "):
        return f"\033[31m{line}\033[0m"
    return line


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Patch ArgoCD Application status.history[*].source.repoURL values"
    )
    ap.add_argument("--context", help="Kube context (defaults to current kubectl context)")
    ap.add_argument("--old-url", required=True)
    mode = ap.add_mutually_exclusive_group(required=True)
    mode.add_argument("--new-url", help="Replace matching repoURL values with this URL")
    mode.add_argument(
        "--drop-revisions",
        action="store_true",
        help="Delete matching status.history entries instead of replacing repoURL",
    )
    ap.add_argument(
        "--application",
        help="Only process the Application with this metadata.name",
    )
    ap.add_argument(
        "--color",
        choices=["auto", "always", "never"],
        default="auto",
        help="Colorize unified diff output",
    )
    ap.add_argument("--yes", action="store_true", help="Apply without confirmation")
    args = ap.parse_args()
    color = use_color(args.color)

    context_args = ["--context", args.context] if args.context else []
    raw = run(["kubectl", *context_args, "get", "applications.argoproj.io", "-A", "-o", "json"])
    items = json.loads(raw).get("items", [])
    if args.application:
        items = [app for app in items if app.get("metadata", {}).get("name") == args.application]
        if not items:
            print(
                "No Application named "
                f"{args.application!r} found in context "
                f"{(args.context if args.context else 'current kubectl context')!r}.",
                file=sys.stderr,
            )
            return 1

    plans: list[dict[str, Any]] = []
    for app in items:
        ns = app.get("metadata", {}).get("namespace")
        name = app.get("metadata", {}).get("name")
        history = app.get("status", {}).get("history") or []

        ops = []
        before = []
        after = []

        matched: list[tuple[int, dict[str, Any]]] = []
        for i, entry in enumerate(history):
            source = entry.get("source") or {}
            url = source.get("repoURL")
            if url == args.old_url:
                matched.append((i, entry))
                before.append(
                    {
                        "index": i,
                        "id": entry.get("id"),
                        "revision": entry.get("revision"),
                        "repoURL": args.old_url,
                    }
                )

        if args.drop_revisions:
            for i, _entry in reversed(matched):
                ops.append({"op": "remove", "path": f"/status/history/{i}"})
        else:
            for i, entry in matched:
                ops.append(
                    {
                        "op": "replace",
                        "path": f"/status/history/{i}/source/repoURL",
                        "value": args.new_url,
                    }
                )
                after.append(
                    {
                        "index": i,
                        "id": entry.get("id"),
                        "revision": entry.get("revision"),
                        "repoURL": args.new_url,
                    }
                )

        if ops:
            plans.append(
                {
                    "namespace": ns,
                    "name": name,
                    "ops": ops,
                    "before": before,
                    "after": after,
                }
            )

    if not plans:
        print("No Application status.history entries matched the old URL.")
        return 0

    total_ops = sum(len(p["ops"]) for p in plans)
    action = "removed" if args.drop_revisions else "updated"
    print(f"Found {len(plans)} Applications with {total_ops} matching history entries to be {action}.\n")

    for plan in plans:
        before_doc = {
            "namespace": plan["namespace"],
            "name": plan["name"],
            "changedHistory": plan["before"],
        }
        after_doc = {
            "namespace": plan["namespace"],
            "name": plan["name"],
            "changedHistory": plan["after"],
        }

        obj_id = f"{plan['namespace']}/{plan['name']}"
        diff = difflib.unified_diff(
            json_dump(before_doc).splitlines(),
            json_dump(after_doc).splitlines(),
            fromfile=f"a/{obj_id}.json",
            tofile=f"b/{obj_id}.json",
            lineterm="",
        )
        for line in diff:
            print(colorize_diff_line(line, color))
        print()

    if not args.yes:
        answer = input("Do you want to apply these changes? [y/N]: ").strip().lower()
        if answer not in {"y", "yes"}:
            print("Aborted. No changes were applied.")
            return 0

    failed = 0
    for plan in plans:
        ns = plan["namespace"]
        name = plan["name"]
        patch = json.dumps(plan["ops"], separators=(",", ":"))
        cmd = [
            "kubectl",
            *context_args,
            "patch",
            "applications.argoproj.io",
            name,
            "--type=json",
            "-p",
            patch,
        ]
        if ns:
            cmd = cmd[:4] + ["-n", ns] + cmd[4:]
        p = subprocess.run(cmd, text=True, capture_output=True)
        if p.returncode != 0 and "NotFound" in (p.stderr or "") and ns:
            # Retry once without namespace in case the resource is cluster-scoped.
            retry_cmd = [
                "kubectl",
                *context_args,
                "patch",
                "applications.argoproj.io",
                name,
                "--type=json",
                "-p",
                patch,
            ]
            p = subprocess.run(retry_cmd, text=True, capture_output=True)
        if p.returncode == 0:
            print(f"APPLIED {ns}/{name}")
        else:
            failed += 1
            print(f"FAILED  {ns}/{name}", file=sys.stderr)
            if p.stderr:
                print(p.stderr.strip(), file=sys.stderr)

    if failed:
        print(f"Done with failures: {failed} app(s) failed.", file=sys.stderr)
        return 1

    print("Done. All patches applied successfully.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())