05 - Kubernetes Client-Go

🎁 What if you could build your own kubectl, Helm, or ArgoCD? They all share one library, and once you learn it, you can build anything they can.

devctl image build pushes images to the registry. Now the tool needs devctl k8s deploy <name> <namespace> <image> — a subcommand that tells Kubernetes to run that image, watches the rollout, and reports exactly what failed.

Your team writes that deploy command. The first version shells out to kubectl:

cmd := exec.Command("kubectl", "apply", "-f", "deploy.yaml")
out, err := cmd.CombinedOutput()
if err != nil {
    log.Fatalf("kubectl apply failed: %s\n%s", err, out)
}

This works until it doesn't. The tool runs in a CI container. kubectl isn't installed. You add it to the image. A few weeks later the cluster is upgraded to 1.29 and the CI image still has kubectl 1.27 in it. The Kubernetes version skew policy allows one minor version of drift, but your internal tooling isn't that careful. You're now maintaining a version matrix and updating the CI image on every cluster upgrade.

Then someone needs to check rollout status before proceeding. A rollout is what happens when you update a Deployment — Kubernetes doesn't replace all pods at once. It creates a new ReplicaSet and gradually scales it up while scaling the old one down. kubectl rollout status watches that transition and blocks until all pods are on the new version.

kubectl rollout status deployment/api --timeout=120s

If it times out you get exit code 1 and a string. You can't tell whether the deployment is stuck on a failing liveness probe, a registry auth error, or a resource quota exhaustion. Parsing kubectl output is fragile and breaks when kubectl changes its wording.

The real problem: exec.Command("kubectl", ...) gives you a subprocess, not a client. You get stdout, stderr, and an exit code. Nothing else.

Imports in code snippets are trimmed for brevity. See the full example linked at the end for complete, compilable source.

Why Not Call the Kubernetes API with net/http?

The Kubernetes API is REST. Every kubectl command is an HTTP call. You could build a thin net/http client. Here's what you'd have to implement:

• Auth detection. In-cluster auth reads a service account token from a mounted file. Out-of-cluster auth parses kubeconfig, which supports multiple auth methods: client certificates, static tokens, and OIDC tokens (short-lived tokens issued by an identity provider like Google, Okta, or Azure AD — what GKE, EKS, and AKS use for human users). Each method has its own token refresh logic. The gcloud auth print-access-token and aws eks get-token commands you see in kubeconfig are OIDC flows under the hood.
• Typed structs. The API returns JSON. Without typed Go structs, you're unmarshaling into map[string]interface{} and navigating nested maps by hand. The Kubernetes API has hundreds of resource types with deeply nested specs.
• Informers. Efficient change detection means a long-lived HTTP watch connection, chunked event parsing, automatic reconnects, and a local cache safe for concurrent reads. This is the most complex part of client-go and took years to get right.

client-go is the library kubectl itself is built on. Using it directly gives you typed access with no installation dependency.

go get k8s.io/client-go@latest
go get k8s.io/apimachinery@latest

Connecting to the Cluster

Your tool needs to work in two environments: on a developer's laptop (reading ~/.kube/config) and inside a pod (reading a mounted service account token). The right pattern detects the environment automatically:

// getClient returns a Clientset that works both inside a pod and on a developer laptop.
func getClient() (*kubernetes.Clientset, error) {
    config, err := rest.InClusterConfig()
    if err != nil {
        kubeconfig := filepath.Join(homedir.HomeDir(), ".kube", "config")
        config, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
        if err != nil {
            return nil, fmt.Errorf("no valid cluster config found: %w", err)
        }
    }
    return kubernetes.NewForConfig(config)
}

rest.InClusterConfig() looks for /var/run/secrets/kubernetes.io/serviceaccount/token. If that file exists, you're in a pod. If not, it falls back to kubeconfig.

Gotcha: Never implement only one path. If you only support kubeconfig and later deploy your tool as a Kubernetes job, it fails with a confusing "no such file" error. If you only support in-cluster and a developer runs it locally, same problem. Three extra lines buys you both.

Patching a Deployment

A Deployment doesn't own pods directly. It owns ReplicaSets, and ReplicaSets own pods. When you change a Deployment, Kubernetes creates a new ReplicaSet for the new version and scales it up while scaling the old one down:

Deployment: api
├── ReplicaSet: api-v1  (replicas: 4 → 3 → 2 → 1 → 0)
└── ReplicaSet: api-v2  (replicas: 0 → 1 → 2 → 3 → 4)

That transition is the rollout. Kubernetes keeps old ReplicaSets around for rollback — kubectl rollout undo just scales the previous one back up.

kubectl get replicasets
# NAME         DESIRED   CURRENT   READY   AGE
# api-abc123   0         0         0       2d    ← old, kept for rollback
# api-def456   4         4         4       1h    ← current

To update the image, use a strategic merge patch. This sends only the fields you want to change — Kubernetes merges them into the existing spec:

// deploy patches the deployment's container image and waits for the rollout.
func deploy(clientset *kubernetes.Clientset, name, namespace, image string) error {
    patch := map[string]interface{}{
        "spec": map[string]interface{}{
            "template": map[string]interface{}{
                "spec": map[string]interface{}{
                    "containers": []interface{}{
                        map[string]interface{}{
                            "name":  name,
                            "image": image,
                        },
                    },
                },
            },
        },
    }

    patchBytes, _ := json.Marshal(patch)

    _, err := clientset.AppsV1().Deployments(namespace).Patch(
        context.Background(),
        name,
        types.StrategicMergePatchType,
        patchBytes,
        metav1.PatchOptions{},
    )
    if err != nil {
        return fmt.Errorf("patch deployment: %w", err)
    }

    fmt.Printf("deployment %s updated to %s, waiting for rollout...\n", name, image)
    return waitForRollout(clientset, name, namespace, 2*time.Minute)
}

Gotcha: If you use types.MergePatchType instead of types.StrategicMergePatchType on a resource with list fields, the patch replaces the entire list. Patching a deployment's container list with MergePatchType removes every container except the one you specified. StrategicMergePatchType merges intelligently. Always use strategic merge for built-in Kubernetes resources.

Waiting for the Rollout

This is what replaces kubectl rollout status. Instead of a string and an exit code, you get the exact replica counts and the failure reason:

// waitForRollout polls until all replicas are updated and ready, or timeout elapses.
func waitForRollout(clientset *kubernetes.Clientset, name, namespace string, timeout time.Duration) error {
    deadline := time.Now().Add(timeout)
    for time.Now().Before(deadline) {
        d, err := clientset.AppsV1().Deployments(namespace).Get(
            context.Background(), name, metav1.GetOptions{},
        )
        if err != nil {
            return err
        }

        updated := d.Status.UpdatedReplicas
        ready := d.Status.ReadyReplicas
        desired := *d.Spec.Replicas

        if updated == desired && ready == desired && d.Status.AvailableReplicas == desired {
            fmt.Printf("rollout complete: %d/%d replicas ready\n", ready, desired)
            return nil
        }

        // Check for a stuck deployment before waiting
        for _, condition := range d.Status.Conditions {
            if condition.Type == "Progressing" && condition.Reason == "ProgressDeadlineExceeded" {
                return fmt.Errorf("deployment stuck: %s", condition.Message)
            }
        }

        fmt.Printf("waiting: %d/%d updated, %d/%d ready...\n", updated, desired, ready, desired)
        time.Sleep(2 * time.Second)
    }
    return fmt.Errorf("rollout timed out after %s", timeout)
}

When this fails, you know whether it's a deadline exceeded condition, a specific count of stuck replicas, or a network error — not just "exit code 1."

Putting It Together: devctl k8s

Here's how the deploy and watch functions wire into devctl as the k8s subcommand tree. The raw os.Args switch from the intro is gone — Cobra handles routing, arg validation, and the help text:

var k8sCmd = &cobra.Command{
    Use:   "k8s",
    Short: "Interact with a Kubernetes cluster",
}

var k8sDeployCmd = &cobra.Command{
    Use:   "deploy <name> <namespace> <image>",
    Short: "Update a deployment's image and wait for rollout",
    Args:  cobra.ExactArgs(3),
    RunE: func(cmd *cobra.Command, args []string) error {
        clientset, err := getClient()
        if err != nil {
            return fmt.Errorf("connect to cluster: %w", err)
        }
        return deploy(clientset, args[0], args[1], args[2])
    },
}

var k8sWatchCmd = &cobra.Command{
    Use:   "watch [namespace]",
    Short: "Stream pod add/delete events in real time",
    Args:  cobra.MaximumNArgs(1),
    RunE: func(cmd *cobra.Command, args []string) error {
        clientset, err := getClient()
        if err != nil {
            return fmt.Errorf("connect to cluster: %w", err)
        }
        namespace := "default"
        if len(args) == 1 {
            namespace = args[0]
        }
        watchPods(clientset, namespace)
        return nil
    },
}

func init() {
    k8sCmd.AddCommand(k8sDeployCmd)
    k8sCmd.AddCommand(k8sWatchCmd)
    rootCmd.AddCommand(k8sCmd)
}

// watchPods streams pod add/delete events using a shared informer until Ctrl+C.
func watchPods(clientset *kubernetes.Clientset, namespace string) {
    stopCh := make(chan struct{})

    factory := informers.NewSharedInformerFactory(clientset, 30*time.Second)
    podInformer := factory.Core().V1().Pods().Informer()

    podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            pod := obj.(*corev1.Pod)
            fmt.Printf("[ADD]    %s/%s\n", pod.Namespace, pod.Name)
        },
        DeleteFunc: func(obj interface{}) {
            // Unwrap DeletedFinalStateUnknown — objects deleted during connectivity
            // gaps arrive in this wrapper, not as a raw pod.
            if d, ok := obj.(cache.DeletedFinalStateUnknown); ok {
                obj = d.Obj
            }
            pod, ok := obj.(*corev1.Pod)
            if !ok {
                return
            }
            fmt.Printf("[DELETE] %s/%s\n", pod.Namespace, pod.Name)
        },
    })

    factory.Start(stopCh)
    factory.WaitForCacheSync(stopCh)
    fmt.Printf("watching pods in %q — Ctrl+C to stop\n", namespace)

    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, os.Interrupt)
    <-sigCh
    close(stopCh)
}

Running the subcommands:

# deploy and watch the rollout
devctl k8s deploy api default myregistry/api:v1.3.0
# deployment api updated to myregistry/api:v1.3.0, waiting for rollout...
# waiting: 2/4 updated, 2/4 ready...
# waiting: 3/4 updated, 3/4 ready...
# rollout complete: 4/4 replicas ready

# stream pod events in real time
devctl k8s watch default
# watching pods in "default" — Ctrl+C to stop
# [ADD]    default/api-def456-xk9p2
# [DELETE] default/api-abc123-m7t8q

That's the full replacement for the bash script from the intro. The devctl k8s deploy command knows exactly what failed and why. The watch command stays alive through reconnects.

Beyond the Deploy Tool

The patterns above cover the core use case. client-go gives you the same API surface for everything else you might need.

Listing any resource:

// Equivalent to: kubectl get pods -l app=web -n default
pods, err := clientset.CoreV1().Pods("default").List(ctx, metav1.ListOptions{
    LabelSelector: "app=web",
})
for _, pod := range pods.Items {
    fmt.Printf("%s  phase=%s  ip=%s\n", pod.Name, pod.Status.Phase, pod.Status.PodIP)
    // Output: web-abc123  phase=Running  ip=10.244.0.12
}

pod.Status.Phase is a typed PodPhase constant. Typos are compile errors, not runtime bugs.

Creating resources from code, no YAML:

cm := &corev1.ConfigMap{
    ObjectMeta: metav1.ObjectMeta{Name: "app-config", Namespace: "default"},
    Data:       map[string]string{"log_level": "info", "region": "us-west-2"},
}
created, err := clientset.CoreV1().ConfigMaps("default").Create(ctx, cm, metav1.CreateOptions{})
fmt.Printf("created: %s\n", created.UID)

CRDs and custom resources — the dynamic client:

The typed clientset only knows built-in resources. For Custom Resource Definitions, use the dynamic client with a group/version/resource triplet (schema.GroupVersionResource from k8s.io/apimachinery/pkg/runtime/schema):

dynClient, _ := dynamic.NewForConfig(config)

gvr := schema.GroupVersionResource{Group: "apps.example.com", Version: "v1", Resource: "webapps"}
list, err := dynClient.Resource(gvr).Namespace("default").List(ctx, metav1.ListOptions{})

for _, item := range list.Items {
    name := item.GetName()
    image, _, _ := unstructured.NestedString(item.Object, "spec", "image")
    fmt.Printf("WebApp: %s  image: %s\n", name, image)
    // Output: WebApp: my-service  image: nginx:1.25
}

The dynamic client returns unstructured.Unstructured objects. Navigate them with unstructured.NestedString, unstructured.NestedMap, and unstructured.NestedSlice.

Cluster Resource Watcher

A Kubernetes client tool demonstrating three core client-go patterns: listing pods across all namespaces, creating a ConfigMap, and watching for real-time pod changes using a shared informer. It auto-detects in-cluster vs local kubeconfig and handles graceful shutdown on Ctrl+C.

Input: A valid kubeconfig at ~/.kube/config (or in-cluster service account when running inside a pod)

Output: A list of all pods with their status, ConfigMap creation confirmation, and a live stream of pod add/delete events

Full source: examples/cluster-resource-watcher

Key Takeaways

• exec.Command("kubectl", ...) gives you a subprocess with an exit code. client-go gives you typed structs, structured errors, and streaming events
• Implement both in-cluster and kubeconfig auth paths. Three lines handles both environments
• A Deployment owns ReplicaSets, which own pods. Updating a Deployment creates a new ReplicaSet — that transition is the rollout
• Use StrategicMergePatchType for built-in resources. MergePatchType replaces list fields entirely instead of merging, removing containers you didn't mention
• Poll waitForRollout for one-shot CLI tools. Use informers for long-running controllers that need to react to every cluster event
• In the informer DeleteFunc, always unwrap cache.DeletedFinalStateUnknown. Objects deleted during connectivity gaps arrive in this wrapper, not as a raw pod
• The dynamic client handles CRDs using group/version/resource triplets. Navigate the response with unstructured.NestedString, NestedMap, and NestedSlice

🎁 What if your code could watch for changes and automatically reconcile the cluster to match a desired state, no human intervention needed?