ingress-patcher.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "argparse",
#     "kubernetes<=31.0",
#     "loguru",
# ]
# ///

import re
import argparse
from typing import Optional, List
from dataclasses import dataclass
from kubernetes import client, config
from loguru import logger

@dataclass
class IngressConfig:
    """Configuration for ingress updates."""
    context: str
    name_regex: Optional[str] = None
    dry_run: bool = False

class IngressPatcher:
    """Handles Kubernetes ingress annotation updates."""

    def __init__(self, config: IngressConfig):
        self.config = config
        self.api: Optional[client.NetworkingV1Api] = None

    def initialize(self) -> None:
        """Initialize Kubernetes client."""
        try:
            config.load_kube_config(context=self.config.context)
            self.api = client.NetworkingV1Api()
            logger.info(f"Connected to cluster using context: {self.config.context}")
        except config.config_exception.ConfigException as e:
            logger.error(f"Failed to load kubeconfig: {e}")
            raise

    def should_process_ingress(self, ingress: client.V1Ingress) -> bool:
        """Check if ingress should be processed based on regex."""
        if not self.config.name_regex:
            return True
        return bool(re.match(self.config.name_regex, ingress.metadata.name))

    def update_annotations(self) -> None:
        """Update ingress annotations."""
        try:
            ingresses = self.api.list_ingress_for_all_namespaces()

            for ingress in ingresses.items:
                if not self.should_process_ingress(ingress):
                    continue

                self._process_ingress(ingress)

        except client.exceptions.ApiException as e:
            logger.error(f"Kubernetes API error: {e}")
            raise

    def _process_ingress(self, ingress: client.V1Ingress) -> None:
        """Process single ingress resource."""
        name = ingress.metadata.name
        namespace = ingress.metadata.namespace
        annotations = ingress.metadata.annotations or {}

        if 'nginx.ingress.kubernetes.io/proxy-body-size' not in annotations:
            logger.info(f"Skipping {namespace}/{name}: no proxy-body-size annotation")
            return

        if annotations['nginx.ingress.kubernetes.io/proxy-body-size'] != '1Gi':
            logger.info(f"Skipping {namespace}/{name}: value not '1Gi'")
            return

        if self.config.dry_run:
            logger.info(f"[DRY RUN] Would update {namespace}/{name}")
            return

        self._update_ingress(ingress, namespace, name)

    def _update_ingress(self, ingress: client.V1Ingress, namespace: str, name: str) -> None:
        """Update ingress with new annotation value."""
        try:
            ingress.metadata.annotations['nginx.ingress.kubernetes.io/proxy-body-size'] = '1G'
            self.api.patch_namespaced_ingress(name, namespace, ingress)
            logger.success(f"Updated {namespace}/{name}")
        except client.exceptions.ApiException as e:
            logger.error(f"Failed to update {namespace}/{name}: {e}")

def main() -> None:
    """Main entry point."""
    parser = argparse.ArgumentParser(description="Update Kubernetes Ingress annotations")
    parser.add_argument("context", help="Kubernetes context to use")
    parser.add_argument("--name-regex", help="Regex pattern to match ingress names")
    parser.add_argument("--dry-run", action="store_true", help="Perform a dry run")

    args = parser.parse_args()
    logger.add("ingress_update.log", rotation="10 MB")

    config = IngressConfig(
        context=args.context,
        name_regex=args.name_regex,
        dry_run=args.dry_run
    )

    try:
        patcher = IngressPatcher(config)
        patcher.initialize()
        patcher.update_annotations()
    except Exception as e:
        logger.error(f"Script failed: {e}")
        raise SystemExit(1)

if __name__ == "__main__":
    main()