prototype of an operator to continuously patch registry credentials into all k8s namespaces and eliminate imagePullPolicy: Always

main.go

package main

// run in the vm
// * continuously set imagePullPolicy -> IfNotPresent because 'Always' makes docker hub
//   hate us
// * continuously distribute regcreds secret to new serviceaccounts

// TODO: rewrite as k8s operator & admission controller

import (
	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"os/exec"
	"strings"
	"time"
)

func deleteEmpty(s []string) []string {
	var r []string
	for _, str := range s {
		if str != "" {
			r = append(r, str)
		}
	}
	return r
}

func kubectlGetJson(kind, namespace, name string) ([]byte, error) {
	cmd := []string{
		"get", kind, "--namespace", namespace, name, "-o", "json",
	}
	out, err := exec.Command("kubectl", cmd...).CombinedOutput()
	if err != nil {
		return []byte{}, fmt.Errorf("%s: %s", err, out)
	}
	return out, nil
}

type NamespacedObject struct {
	Namespace string
	Name      string
}

func kubectlGetAll(kind string) ([]NamespacedObject, error) {
	cmd := []string{
		"get", kind, "--no-headers", "-A", "-o", "custom-columns=namespace:.metadata.namespace,name:.metadata.name",
	}
	out, err := exec.Command("kubectl", cmd...).CombinedOutput()
	if err != nil {
		return []NamespacedObject{}, fmt.Errorf("%s: %s", err, out)
	}
	ret := []NamespacedObject{}
	list := deleteEmpty(strings.Split(string(out), "\n"))
	for _, item := range list {
		parts := strings.Fields(item)
		ret = append(ret, NamespacedObject{
			Namespace: parts[0],
			Name:      parts[1],
		})
	}
	return ret, nil
}

type ImagePullPolicy struct {
	Spec struct {
		Template struct {
			Spec struct {
				Containers []struct {
					ImagePullPolicy string
				}
			}
		}
	}
}

func patchImagePullPolicy(kind, ns, object string, i int, targetPolicy string) error {
	cmd := []string{
		"patch", kind, object, "-n", ns, "--type", "json", "-p",
		fmt.Sprintf(`[{
				"op":"replace",
				"path":"/spec/template/spec/containers/%d/imagePullPolicy",
				"value": "IfNotPresent"
			}]`, i,
		),
	}
	out, err := exec.Command("kubectl", cmd...).CombinedOutput()
	if err != nil {
		return fmt.Errorf("%s: %s", err, out)
	}
	return nil
}

func eliminate(kind, ns, object string) error {
	// object is a kind of thing that has /spec/template/spec/containers/*/imagePullPolicy
	// our mission is to set them all to 'IfNotPresent'
	// let's start by searching for them.

	js, err := kubectlGetJson(kind, ns, object)
	if err != nil {
		return err
	}

	var imagePullPolicy ImagePullPolicy
	json.Unmarshal(js, &imagePullPolicy)

	//log.Printf("found %d containers on %s %s/%s", len(imagePullPolicy.Spec.Template.Spec.Containers), kind, ns, object)

	for i, imp := range imagePullPolicy.Spec.Template.Spec.Containers {
		if imp.ImagePullPolicy == "Always" {
			targetPolicy := "IfNotPresent"
			err = patchImagePullPolicy(kind, ns, object, i, targetPolicy)
			if err != nil {
				log.Printf("error patching %s/%s/%s: %s; continuing", kind, ns, object, err)
			} else {
				log.Printf(
					"successfully patched imagePullPolicy on %s/%s/%s/%d to %s",
					kind, ns, object, i, targetPolicy,
				)
			}
		}
	}
	return nil
}

func eliminateImagePullPolicyAlways(cache map[string]bool) error {
	for _, kind := range []string{"deployment", "statefulset", "replicaset"} {
		objects, err := kubectlGetAll(kind)
		if err != nil {
			return err
		}
		for _, object := range objects {
			tag := fmt.Sprintf("%s/%s/%s", kind, object.Namespace, object.Name)
			if !cache[tag] {
				err = eliminate(kind, object.Namespace, object.Name)
				if err != nil {
					log.Printf("error eliminating %s/%s/%s: %s; continuing", kind, object.Namespace, object.Name, err)
				}
				cache[tag] = true
			}
		}
	}
	return nil
}

func patchRegcreds(cache map[string]bool) error {
	serviceaccounts, err := kubectlGetAll("serviceaccount")
	if err != nil {
		return fmt.Errorf("error listing serviceaccounts: %s; continuing", err)
	}
	for _, sa := range serviceaccounts {
		tag := fmt.Sprintf("%s/%s", sa.Namespace, sa.Name)
		if !cache[tag] {
			regcred, err := exec.Command(
				"kubectl", "get", "secret", "regcred",
				"--namespace", "default", "--export", "-o", "yaml",
			).Output()
			if err != nil {
				return fmt.Errorf("Error getting regcred secret in default namespace: %s", err)
			}
			cmd := exec.Command(
				"kubectl", "apply", "--namespace", sa.Namespace,
				"-f", "-",
			)
			var b bytes.Buffer
			b.Write([]byte(regcred))
			cmd.Stdin = &b
			out, err := cmd.CombinedOutput()
			if err != nil {
				return fmt.Errorf(
					"error running kubectl apply to install secret in %s: %s: %s",
					sa.Namespace, err, out,
				)
			}
			out, err = exec.Command(
				"kubectl", "patch", "serviceaccount", "-n", sa.Namespace, sa.Name,
				"-p", `{"imagePullSecrets": [{"name": "regcred"}]}`,
			).CombinedOutput()
			if err != nil {
				return fmt.Errorf(
					"error running kubectl apply to patch sa %s in %s: %s: %s",
					sa.Name, sa.Namespace, err, out,
				)
			} else {
				if !strings.Contains(string(out), "no change") {
					log.Printf("patched sa %s/%s: %s", sa.Namespace, sa.Name, string(out))
				}
			}
			cache[tag] = true
		}
	}
	return nil
}

func always(f func(map[string]bool) error, cache map[string]bool, name string) {
	for {
		start := time.Now()
		err := f(cache)
		duration := time.Now().Sub(start)
		log.Printf("%s took %s", name, duration)
		if err != nil {
			log.Printf("Error running %s: %s, retrying in 10s.", name, err)
		}
		time.Sleep(10 * time.Second)
	}
}

func main() {
	log.Println("Starting testfaster-agent v0.0.5")
	imagePullCache := map[string]bool{}
	go always(eliminateImagePullPolicyAlways, imagePullCache, "eliminateImagePullPolicyAlways")

	regcredsCache := map[string]bool{}
	always(patchRegcreds, regcredsCache, "patchRegcreds")
}