Simple Controllers in Python

from kubernetes import client, config, watch

config.load_kube_config()
custom_api = client.CustomObjectsApi()
v1 = client.CoreV1Api()

GROUP = "mycompany.com"
VERSION = "v1"
PLURAL = "deployments"

def reconcile(spec: dict, name: str, namespace: str) -> str:
    """Compare desired state to actual state, take action."""
    desired_replicas = spec.get("replicas", 1)
    service_name = spec.get("service_name", name)

    # Check if the Deployment exists
    try:
        deployment = apps_v1.read_namespaced_deployment(name, namespace)
        current_replicas = deployment.spec.replicas
    except client.exceptions.ApiException as e:
        if e.status == 404:
            # Create the Deployment
            create_deployment(name, namespace, desired_replicas, service_name)
            return "created"
        raise

    # Scale if needed
    if current_replicas != desired_replicas:
        deployment.spec.replicas = desired_replicas
        apps_v1.patch_namespaced_deployment(name, namespace, deployment)
        return "scaled"

    return "unchanged"

def run_controller():
    """Main controller loop."""
    w = watch.Watch()
    for event in w.stream(custom_api.list_cluster_custom_object, group=GROUP, version=VERSION, plural=PLURAL):
        obj = event["object"]
        name = obj["metadata"]["name"]
        namespace = obj["metadata"].get("namespace", "default")
        spec = obj.get("spec", {})

        status = reconcile(spec, name, namespace)
        print(f"Reconciled {name}: {status}")

from kubernetes import client, config

config.load_kube_config()
api = client.ApiextensionsV1Api()

# Define a CRD
crd = {
    "apiVersion": "apiextensions.k8s.io/v1",
    "kind": "CustomResourceDefinition",
    "metadata": {"name": "webapps.mycompany.com"},
    "spec": {
        "group": "mycompany.com",
        "versions": [{"name": "v1", "served": True, "storage": True, "schema": {
            "openAPIV3Schema": {"type": "object", "properties": {
                "spec": {"type": "object", "properties": {
                    "replicas": {"type": "integer"},
                    "image": {"type": "string"},
                    "service_name": {"type": "string"},
                }},
            }},
        }}],
        "scope": "Namespaced",
        "names": {"plural": "webapps", "singular": "webapp", "kind": "WebApp"},
    },
}

# Create or update the CRD
try:
    api.create_custom_resource_definition(crd)
    print("CRD created")
except client.exceptions.ApiException as e:
    if e.status == 409:  # Already exists
        api.patch_custom_resource_definition("webapps.mycompany.com", crd)
        print("CRD updated")

# Create a custom resource
custom_api = client.CustomObjectsApi()
webapp = {
    "apiVersion": "mycompany.com/v1",
    "kind": "WebApp",
    "metadata": {"name": "my-webapp"},
    "spec": {
        "replicas": 3,
        "service_name": "my-webapp",
        "image": "nginx:latest",
    },
}

custom_api.create_namespaced_custom_object(
    group="mycompany.com",
    version="v1",
    namespace="default",
    plural="webapps",
    body=webapp,
)

from kubernetes import client, config, watch
import traceback
import time

def run_controller():
    config.load_kube_config()
    custom_api = client.CustomObjectsApi()
    w = watch.Watch()

    resource_version = ""
    while True:
        try:
            for event in w.stream(
                custom_api.list_cluster_custom_object,
                group="mycompany.com",
                version="v1",
                plural="webapps",
                resource_version=resource_version,
                timeout_seconds=300,
            ):
                obj = event["object"]
                resource_version = obj["metadata"]["resourceVersion"]
                event_type = event["type"]

                try:
                    handle_event(event_type, obj)
                except Exception as e:
                    print(f"Error handling {obj['metadata']['name']}: {e}")
                    traceback.print_exc()

        except client.exceptions.ApiException as e:
            if e.status == 410:  # Resource version too old
                resource_version = ""
                print("Resource version expired, restarting watch")
            else:
                print(f"API error: {e}")
                time.sleep(5)
        except Exception as e:
            print(f"Unexpected error: {e}")
            time.sleep(5)

def handle_event(event_type: str, obj: dict):
    name = obj["metadata"]["name"]
    print(f"Event: {event_type} on {name}")

    if event_type in ["ADDED", "MODIFIED"]:
        reconcile(obj["spec"], name, obj["metadata"].get("namespace", "default"))
    elif event_type == "DELETED":
        cleanup(name, obj["metadata"].get("namespace", "default"))

from kubernetes import client, config
from datetime import datetime

config.load_kube_config()
custom_api = client.CustomObjectsApi()

def update_status(name: str, namespace: str, status: dict):
    """Update the status subresource of a custom resource."""
    custom_api.patch_namespaced_custom_object_status(
        group="mycompany.com",
        version="v1",
        namespace=namespace,
        plural="webapps",
        name=name,
        body={"status": status},
    )

# Update status after reconciliation
update_status(
    name="my-webapp",
    namespace="default",
    status={
        "phase": "Running",
        "replicas": 3,
        "ready_replicas": 3,
        "last_reconciled": datetime.utcnow().isoformat(),
        "conditions": [
            {
                "type": "Ready",
                "status": "True",
                "last_transition_time": datetime.utcnow().isoformat(),
                "reason": "AllReplicasReady",
                "message": "All 3 replicas are ready",
            }
        ],
    },
)

from kubernetes import client, config
from kubernetes.leaderelection import election, leaderelection
from kubernetes.leaderelection.resourcelock import ResourceLock

config.load_kube_config()

# Only one instance of the controller runs at a time
def run():
    """Controller main loop."""
    w = watch.Watch()
    for event in w.stream(custom_api.list_cluster_custom_object, ...):
        reconcile(event)

def main():
    lock = ResourceLock(
        "leases",
        "kube-system",
        "my-controller-lock",
        client.CoordinationV1Api(),
        leader_election_id="my-controller",
    )

    with election.LeaderElector(
        lock,
        on_start_leading=lambda: run(),
        on_stop_leading=lambda: print("Lost leadership"),
    ).run():
        pass

import logging
import time
import traceback

logger = logging.getLogger(__name__)

def safe_reconcile(spec: dict, name: str, namespace: str) -> str:
    """Reconcile with error recovery and backoff."""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            return reconcile(spec, name, namespace)
        except client.exceptions.ApiException as e:
            if e.status in [409, 422]:  # Conflict, validation error
                logger.warning(f"Cannot reconcile {name}: {e.reason}")
                return "error"
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                logger.warning(f"Retry {attempt + 1}/{max_retries} for {name} in {wait}s")
                time.sleep(wait)
        except Exception as e:
            logger.error(f"Unexpected error reconciling {name}: {e}")
            traceback.print_exc()
            return "error"

    logger.error(f"Failed to reconcile {name} after {max_retries} attempts")
    return "error"