Kubernetes Has a Free API: Here's How to Use It for Cluster Automation

#kubernetes #devops #api #automation

Every Kubernetes cluster exposes a powerful REST API. If you're still using only kubectl, you're missing out on programmatic control that can automate deployments, scaling, monitoring, and incident response.

Why Use the Kubernetes API Directly?

Automate deployments without shell scripts wrapping kubectl
Build custom dashboards and monitoring tools
Create self-healing systems that respond to cluster events
Integrate K8s into your existing toolchain via HTTP

Getting Started

The API server runs on your cluster's control plane. Access it via kubectl proxy:

# Start a local proxy to the API server
kubectl proxy --port=8080 &

# List all pods in default namespace
curl -s http://localhost:8080/api/v1/namespaces/default/pods | jq '.items[] | {name: .metadata.name, status: .status.phase}'

List All Resources

# Get all API resources available
curl -s http://localhost:8080/api/v1 | jq '.resources[] | {name: .name, kind: .kind, verbs: .verbs}'

# Get nodes and their capacity
curl -s http://localhost:8080/api/v1/nodes | jq '.items[] | {name: .metadata.name, cpu: .status.capacity.cpu, memory: .status.capacity.memory}'

Python Client

The official Python client makes it even easier:

from kubernetes import client, config

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

# List all pods across all namespaces
pods = v1.list_pod_for_all_namespaces()
for pod in pods.items:
    print(f"{pod.metadata.namespace:20s} {pod.metadata.name:40s} {pod.status.phase}")

Watch for Events in Real-Time

from kubernetes import client, config, watch

config.load_kube_config()
v1 = client.CoreV1Api()
w = watch.Watch()

for event in w.stream(v1.list_pod_for_all_namespaces, timeout_seconds=60):
    pod = event['object']
    event_type = event['type']
    print(f"{event_type}: {pod.metadata.namespace}/{pod.metadata.name} -> {pod.status.phase}")

Auto-Scale Based on Custom Metrics

from kubernetes import client, config
import time

config.load_kube_config()
apps_v1 = client.AppsV1Api()
metrics_api = client.CustomObjectsApi()

def auto_scale(namespace, deployment, min_replicas=1, max_replicas=10, cpu_threshold=80):
    # Get current pod metrics
    pods = metrics_api.list_namespaced_custom_object(
        "metrics.k8s.io", "v1beta1", namespace, "pods"
    )

    total_cpu = 0
    pod_count = 0
    for pod in pods["items"]:
        for container in pod["containers"]:
            cpu_nano = int(container["usage"]["cpu"].rstrip("n"))
            total_cpu += cpu_nano / 1e9 * 100
            pod_count += 1

    avg_cpu = total_cpu / max(pod_count, 1)

    deploy = apps_v1.read_namespaced_deployment(deployment, namespace)
    current = deploy.spec.replicas

    if avg_cpu > cpu_threshold and current < max_replicas:
        new_replicas = min(current + 1, max_replicas)
        deploy.spec.replicas = new_replicas
        apps_v1.patch_namespaced_deployment(deployment, namespace, deploy)
        print(f"Scaled UP {deployment}: {current} -> {new_replicas} (CPU: {avg_cpu:.1f}%)")
    elif avg_cpu < cpu_threshold / 2 and current > min_replicas:
        new_replicas = max(current - 1, min_replicas)
        deploy.spec.replicas = new_replicas
        apps_v1.patch_namespaced_deployment(deployment, namespace, deploy)
        print(f"Scaled DOWN {deployment}: {current} -> {new_replicas} (CPU: {avg_cpu:.1f}%)")

Deploy an Application

from kubernetes import client, config

config.load_kube_config()
apps_v1 = client.AppsV1Api()

deployment = client.V1Deployment(
    metadata=client.V1ObjectMeta(name="my-app"),
    spec=client.V1DeploymentSpec(
        replicas=3,
        selector=client.V1LabelSelector(
            match_labels={"app": "my-app"}
        ),
        template=client.V1PodTemplateSpec(
            metadata=client.V1ObjectMeta(labels={"app": "my-app"}),
            spec=client.V1PodSpec(
                containers=[client.V1Container(
                    name="my-app",
                    image="my-app:latest",
                    ports=[client.V1ContainerPort(container_port=8080)]
                )]
            )
        )
    )
)

apps_v1.create_namespaced_deployment(namespace="default", body=deployment)
print("Deployment created!")

Real-World Use Case

An SRE team at an e-commerce company built a custom auto-scaler using the K8s API. During Black Friday, their system automatically scaled from 5 to 50 pods in under 2 minutes — handling 10x normal traffic without a single manual intervention.