#!/usr/bin/env python3 """ request_upgrades.py: Kubernetes image upgrade approval workflow (GitOps). Finds outdated images in the cluster, maps them to YAML files in the ArgoCD gitops repo, requests Mattermost approval, then on approval commits and pushes the tag updates. ArgoCD auto-syncs from there — no direct kubectl patching. """ import os import re import shutil import subprocess import tempfile import time import logging from datetime import datetime, timezone from pathlib import Path import requests from kubernetes import client, config from packaging.version import Version, InvalidVersion from check_versions import ( ImageInfo, get_cluster_images, get_latest_version, ) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger(__name__) APPROVAL_MIDDLEWARE_URL = os.environ.get("APPROVAL_MIDDLEWARE_URL", "http://192.168.87.14:8085") MATTERMOST_WEBHOOK_URL = os.environ["MATTERMOST_WEBHOOK_URL"] APPROVAL_TIMEOUT_MINUTES = int(os.environ.get("APPROVAL_TIMEOUT_MINUTES", "60")) APPROVAL_POLL_INTERVAL = int(os.environ.get("APPROVAL_POLL_INTERVAL", "30")) GITEA_TOKEN = os.environ["GITEA_TOKEN"] GITEA_USERNAME = os.environ.get("GITEA_USERNAME", "x-token") GITEA_REPO_URL = os.environ.get( "GITEA_REPO_URL", "https://repo.adservio.us/ai_approver/argocd-gitops.git" ) GITEA_COMMITTER_NAME = os.environ.get("GITEA_COMMITTER_NAME", "k8s-version-tracker") GITEA_COMMITTER_EMAIL = os.environ.get( "GITEA_COMMITTER_EMAIL", "k8s-version-tracker@themosers.club" ) # ───────────────────────────────────────────────────────────────────────────── # Version check # ───────────────────────────────────────────────────────────────────────────── def collect_outdated(all_images: dict[str, ImageInfo]) -> list[dict]: results = [] for ref, img in all_images.items(): if img.is_local or img.skip_check or img.is_floating: continue latest = get_latest_version(img) if not latest: continue try: is_outdated = Version(latest.lstrip("v")) > Version(img.tag.lstrip("v")) except InvalidVersion: is_outdated = img.tag != latest if is_outdated: results.append({ "image": ref, "image_short": img.display_name, "current_tag": img.tag, "latest_tag": latest, "new_image": ref.rsplit(":", 1)[0] + ":" + latest, "pods": img.pods, }) return results # ───────────────────────────────────────────────────────────────────────────── # Git operations # ───────────────────────────────────────────────────────────────────────────── def _git(args: list[str], cwd: str | None = None) -> subprocess.CompletedProcess: result = subprocess.run( ["git"] + args, cwd=cwd, capture_output=True, text=True, ) if result.returncode != 0: raise RuntimeError(f"git {' '.join(args)} failed:\n{result.stderr}") return result def clone_repo(dest: str) -> None: # Embed token in URL for authentication (never logged) parsed = GITEA_REPO_URL.split("://", 1) auth_url = f"{parsed[0]}://{GITEA_USERNAME}:{GITEA_TOKEN}@{parsed[1]}" _git(["clone", "--depth=1", auth_url, dest]) _git(["config", "user.name", GITEA_COMMITTER_NAME], cwd=dest) _git(["config", "user.email", GITEA_COMMITTER_EMAIL], cwd=dest) logger.info("Cloned %s to %s", GITEA_REPO_URL, dest) def find_yaml_files_with_image(repo_dir: str, image_ref: str) -> list[Path]: """Return all YAML files under repo_dir that contain the exact image_ref string.""" matches = [] for path in Path(repo_dir).rglob("*.yaml"): if image_ref in path.read_text(): matches.append(path) return matches def build_upgrade_plan(repo_dir: str, outdated: list[dict]) -> list[dict]: plan = [] for entry in outdated: files = find_yaml_files_with_image(repo_dir, entry["image"]) if not files: logger.debug("No YAML files found for %s — skipping", entry["image"]) continue for f in files: plan.append({ "file": str(f.relative_to(repo_dir)), "abs_path": str(f), "old_image": entry["image"], "new_image": entry["new_image"], "image_short": entry["image_short"], "current_tag": entry["current_tag"], "latest_tag": entry["latest_tag"], }) return plan def apply_updates(plan: list[dict]) -> list[dict]: """Replace old image refs in YAML files. Returns entries that were actually changed.""" changed = [] for item in plan: path = Path(item["abs_path"]) original = path.read_text() updated = original.replace(item["old_image"], item["new_image"]) if updated != original: path.write_text(updated) changed.append(item) logger.info("Updated %s: %s -> %s", item["file"], item["current_tag"], item["latest_tag"]) else: logger.warning("No replacement made in %s for %s", item["file"], item["old_image"]) return changed def commit_and_push(repo_dir: str, changed: list[dict]) -> str: """Stage changed files, commit, push. Returns the commit SHA.""" for item in changed: _git(["add", item["file"]], cwd=repo_dir) summary_lines = [f"- {item['image_short']}: {item['current_tag']} -> {item['latest_tag']}" for item in changed] now = datetime.now(timezone.utc).strftime("%Y-%m-%d") commit_msg = ( f"chore: upgrade {len(changed)} image(s) — {now}\n\n" + "\n".join(summary_lines) ) _git(["commit", "-m", commit_msg], cwd=repo_dir) _git(["push", "origin", "main"], cwd=repo_dir) sha = _git(["rev-parse", "--short", "HEAD"], cwd=repo_dir).stdout.strip() logger.info("Pushed commit %s with %d image update(s)", sha, len(changed)) return sha # ───────────────────────────────────────────────────────────────────────────── # Approval workflow # ───────────────────────────────────────────────────────────────────────────── def request_approval(upgrade_plan: list[dict]) -> str: lines = ["Proposed ArgoCD image upgrades (GitOps — will commit to argocd-gitops):\n"] for item in upgrade_plan: lines.append( f"- `{item['file']}` — `{item['image_short']}` " f"`{item['current_tag']}` -> `{item['latest_tag']}`" ) reason = "\n".join(lines) resp = requests.post( f"{APPROVAL_MIDDLEWARE_URL}/request-approval", json={ "secret_path": "secret/approval-required/k8s-upgrades", "requester": "k8s-version-tracker", "reason": reason, "ttl_minutes": APPROVAL_TIMEOUT_MINUTES, }, timeout=15, ) resp.raise_for_status() approval_id = resp.json()["approval_id"] logger.info("Approval request created: %s (TTL %dm)", approval_id, APPROVAL_TIMEOUT_MINUTES) return approval_id def poll_for_decision(approval_id: str) -> str: deadline = time.monotonic() + APPROVAL_TIMEOUT_MINUTES * 60 while time.monotonic() < deadline: time.sleep(APPROVAL_POLL_INTERVAL) try: resp = requests.get( f"{APPROVAL_MIDDLEWARE_URL}/approval/{approval_id}/status", timeout=10, ) if resp.status_code == 404: return "expired" resp.raise_for_status() status = resp.json()["status"] if status != "pending": return status except Exception as e: logger.warning("Poll error (will retry): %s", e) return "timeout" # ───────────────────────────────────────────────────────────────────────────── # Mattermost notifications # ───────────────────────────────────────────────────────────────────────────── def send_mattermost(text: str) -> None: requests.post( MATTERMOST_WEBHOOK_URL, json={"text": text, "username": "k8s-upgrade-bot", "icon_emoji": ":kubernetes:"}, timeout=15, ).raise_for_status() def build_applied_report(changed: list[dict], sha: str) -> str: now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") lines = [ f"## :white_check_mark: K8s Image Upgrades Committed -- {now}\n", f"Commit `{sha}` pushed to `argocd-gitops` — ArgoCD is syncing.\n", ] for item in sorted(changed, key=lambda x: x["file"]): lines.append( f"- `{item['file']}` **{item['image_short']}** " f"`{item['current_tag']}` -> `{item['latest_tag']}`" ) return "\n".join(lines) # ───────────────────────────────────────────────────────────────────────────── # Main # ───────────────────────────────────────────────────────────────────────────── def main() -> None: try: config.load_incluster_config() except config.ConfigException: config.load_kube_config() logger.info("Gathering cluster image inventory ...") all_images = get_cluster_images() logger.info("Found %d unique image references", len(all_images)) logger.info("Checking for outdated images ...") outdated = collect_outdated(all_images) logger.info("%d outdated image(s) found", len(outdated)) if not outdated: logger.info("Nothing to upgrade.") send_mattermost( ":white_check_mark: **K8s Upgrade Check**: All images are current -- no upgrades needed." ) return # Clone repo early so we can build an accurate plan (which files will change) repo_dir = tempfile.mkdtemp(prefix="argocd-gitops-") try: logger.info("Cloning gitops repo to build upgrade plan ...") clone_repo(repo_dir) upgrade_plan = build_upgrade_plan(repo_dir, outdated) if not upgrade_plan: msg_lines = [ f":warning: **K8s Upgrade Check**: {len(outdated)} outdated image(s) found " "but none matched any YAML file in the gitops repo.\n", "These may be images from Helm charts or external sources:", ] for entry in outdated: msg_lines.append( f"- `{entry['image_short']}` `{entry['current_tag']}` -> `{entry['latest_tag']}`" ) send_mattermost("\n".join(msg_lines)) return logger.info( "%d file update(s) planned across %d image(s) -- requesting approval ...", len(upgrade_plan), len(outdated), ) approval_id = request_approval(upgrade_plan) logger.info( "Polling for decision (timeout: %dm, interval: %ds) ...", APPROVAL_TIMEOUT_MINUTES, APPROVAL_POLL_INTERVAL, ) decision = poll_for_decision(approval_id) logger.info("Decision: %s", decision) if decision == "approved": changed = apply_updates(upgrade_plan) if changed: sha = commit_and_push(repo_dir, changed) report = build_applied_report(changed, sha) send_mattermost(report) logger.info("Done -- %d file(s) committed", len(changed)) else: send_mattermost( ":warning: **K8s Upgrades**: Approval received but no file changes were " "produced (images may already be updated in the repo)." ) else: status_label = {"denied": "Denied", "expired": "Expired", "timeout": "Timed Out"} emoji_map = {"denied": ":no_entry:", "expired": ":timer_clock:", "timeout": ":hourglass:"} label = status_label.get(decision, decision.title()) icon = emoji_map.get(decision, ":warning:") send_mattermost( f"{icon} **K8s Upgrades {label}** -- no changes committed.\n" f"{len(upgrade_plan)} file update(s) were proposed but not applied." ) logger.info("No changes committed.") finally: shutil.rmtree(repo_dir, ignore_errors=True) if __name__ == "__main__": main()