claude-code
528da1e167
- Add request_upgrades.py and trigger-upgrade.sh - Add upgrade-cronjob.yaml and manual-trigger.yaml (WorkflowTemplate) - Fix git auth: use GITEA_USERNAME env var instead of hardcoded x-token (Gitea requires username:token format, not x-token:token)
330 lines
14 KiB
Python
330 lines
14 KiB
Python
#!/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()
|