Platform Automation in Go and Python

1. 1

   Start the Go CLI

   A module with client-go. shopctl audit loads your kubeconfig exactly like kubectl does.

   terminalcopy cmd

   $ mkdir shopctl && cd shopctl && go mod init github.com/shoplite/shopctl

   go get k8s.io/client-go@v0.34.1 k8s.io/apimachinery@v0.34.1

   ── expected output ──

   go: creating new go.mod: module github.com/shoplite/shopctl

   go: added k8s.io/client-go v0.34.1

2. 2

   Audit deployments against platform standards

   The CLI lists Deployments in team namespaces and flags missing owner labels, missing resource requests, missing probes, and single replicas without a PDB. The same checks as Kyverno, but as a report developers can run locally before merging.

   package main
   
   import (
   	"context"
   	"fmt"
   	"os"
   	"path/filepath"
   
   	appsv1 "k8s.io/api/apps/v1"
   	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
   	"k8s.io/client-go/kubernetes"
   	"k8s.io/client-go/tools/clientcmd"
   )
   
   func main() {
   	if len(os.Args) < 2 || os.Args[1] != "audit" {
   		fmt.Println("usage: shopctl audit")
   		os.Exit(2)
   	}
   	cfg, err := clientcmd.BuildConfigFromFlags("", filepath.Join(os.Getenv("HOME"), ".kube", "config"))
   	check(err)
   	cs, err := kubernetes.NewForConfig(cfg)
   	check(err)
   	ctx := context.Background()
   
   	nss, err := cs.CoreV1().Namespaces().List(ctx, metav1.ListOptions{LabelSelector: "shoplite.dev/team"})
   	check(err)
   	problems := 0
   	for _, ns := range nss.Items {
   		deps, err := cs.AppsV1().Deployments(ns.Name).List(ctx, metav1.ListOptions{})
   		check(err)
   		pdbs, err := cs.PolicyV1().PodDisruptionBudgets(ns.Name).List(ctx, metav1.ListOptions{})
   		check(err)
   		for _, d := range deps.Items {
   			for _, issue := range audit(d, len(pdbs.Items) > 0) {
   				fmt.Printf("%-12s %-22s %s\n", ns.Name, d.Name, issue)
   				problems++
   			}
   		}
   	}
   	if problems > 0 {
   		os.Exit(1) // non-zero so CI can gate on it
   	}
   	fmt.Println("all deployments meet platform standards")
   }
   
   func audit(d appsv1.Deployment, hasPDB bool) []string {
   	var out []string
   	if d.Labels["app.kubernetes.io/owner"] == "" {
   		out = append(out, "missing label app.kubernetes.io/owner")
   	}
   	if d.Spec.Replicas != nil && *d.Spec.Replicas < 2 && !hasPDB {
   		out = append(out, "single replica and no PodDisruptionBudget")
   	}
   	for _, c := range d.Spec.Template.Spec.Containers {
   		if c.Resources.Requests.Cpu().IsZero() || c.Resources.Requests.Memory().IsZero() {
   			out = append(out, fmt.Sprintf("container %s: missing cpu/memory requests", c.Name))
   		}
   		if c.ReadinessProbe == nil {
   			out = append(out, fmt.Sprintf("container %s: no readiness probe", c.Name))
   		}
   	}
   	return out
   }
   
   func check(err error) {
   	if err != nil {
   		fmt.Fprintln(os.Stderr, "error:", err)
   		os.Exit(1)
   	}
   }

   terminalcopy cmd

   $ go build -o shopctl . && ./shopctl audit

   ── expected output ──

   catalog catalog missing label app.kubernetes.io/owner

   catalog catalog single replica and no PodDisruptionBudget

   fulfilment label-printer container printer: no readiness probe

3. 3

   Release it as a binary

   GoReleaser cross-compiles for macOS, Linux, and Windows and publishes GitHub releases (and a Homebrew tap) from a tag. Developers install one file, with no runtime.

   terminalcopy cmd

   $ git tag v0.1.0 && git push --tags

   goreleaser release --clean

   ── expected output ──

   • building binaries

   • building binary=dist/shopctl_linux_amd64_v1/shopctl

   • building binary=dist/shopctl_darwin_arm64/shopctl

   • building binary=dist/shopctl_windows_amd64_v1/shopctl.exe

   • release succeeded after 41s

4. 4

   Python: clean up expired preview environments

   Preview environments (one namespace per open PR, created by an ApplicationSet PR generator) must be removed when stale, or they cost money forever. This job deletes preview-* namespaces whose expires-at annotation has passed, and reports unattached EBS volumes. It defaults to a DRY RUN, since automation that deletes things should always start by printing what it would do.

   """Delete expired preview namespaces and report orphaned EBS volumes."""
   import datetime as dt
   import logging
   import os
   
   import boto3
   from kubernetes import client, config
   
   DRY_RUN = os.environ.get("DRY_RUN", "true") == "true"
   logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
   log = logging.getLogger("preview-cleanup")
   
   
   def expired_preview_namespaces(v1: client.CoreV1Api):
       now = dt.datetime.now(dt.timezone.utc)
       for ns in v1.list_namespace(label_selector="shoplite.dev/preview=true").items:
           expires = (ns.metadata.annotations or {}).get("shoplite.dev/expires-at")
           if expires and dt.datetime.fromisoformat(expires) < now:
               yield ns.metadata.name
   
   
   def orphaned_volumes(ec2):
       pages = ec2.get_paginator("describe_volumes").paginate(
           Filters=[{"Name": "status", "Values": ["available"]},
                    {"Name": "tag:kubernetes.io/created-for/pvc/namespace", "Values": ["preview-*"]}])
       for page in pages:
           for vol in page["Volumes"]:
               yield vol["VolumeId"], vol["Size"]
   
   
   def main():
       config.load_incluster_config() if os.environ.get("KUBERNETES_SERVICE_HOST") else config.load_kube_config()
       v1 = client.CoreV1Api()
       for name in expired_preview_namespaces(v1):
           log.info("%s namespace %s", "would delete" if DRY_RUN else "deleting", name)
           if not DRY_RUN:
               v1.delete_namespace(name)
   
       ec2 = boto3.client("ec2")
       total = 0
       for vol_id, size in orphaned_volumes(ec2):
           total += size
           log.info("orphaned volume %s (%s GiB)", vol_id, size)
       log.info("orphaned preview volumes: %s GiB", total)
   
   
   if __name__ == "__main__":
       main()

   terminalcopy cmd

   $ pip install kubernetes boto3 && DRY_RUN=true python platform-jobs/preview_cleanup.py

   ── expected output ──

   2026-09-27 12:10:02 INFO would delete namespace preview-pr-412

   2026-09-27 12:10:02 INFO would delete namespace preview-pr-398

   2026-09-27 12:10:03 INFO orphaned volume vol-0a1b2c3d4e5f (20 GiB)

   2026-09-27 12:10:03 INFO orphaned preview volumes: 20 GiB

5. 5

   Run it as a CronJob with least privilege

   A ServiceAccount allowed only to list and delete namespaces (bound via ClusterRole), and an IAM role (Pod Identity) allowed only ec2:DescribeVolumes. Switch DRY_RUN to false after a week of reviewing its logs.

   apiVersion: batch/v1
   kind: CronJob
   metadata: { name: preview-cleanup, namespace: platform-jobs }
   spec:
     schedule: "0 * * * *"
     concurrencyPolicy: Forbid
     jobTemplate:
       spec:
         backoffLimit: 1
         template:
           spec:
             serviceAccountName: preview-cleanup
             restartPolicy: Never
             containers:
               - name: cleanup
                 image: 123456789012.dkr.ecr.ap-south-1.amazonaws.com/platform-jobs:1.4.0
                 command: ["python", "preview_cleanup.py"]
                 env: [{ name: DRY_RUN, value: "true" }]
                 resources: { requests: { cpu: 50m, memory: 128Mi }, limits: { memory: 256Mi } }

Make each change, run the command, and read the error before revealing the diagnosis. Recognising these messages on sight is what makes you fast on a real team. Undo the change afterwards.