mode_pipeline.go

// mode_pipeline.go — the full realistic controller pipeline:
//
//	k6/curl ──▶ POST /trigger/{name} ──▶ real kube-apiserver write (envtest)
//	      ──▶ etcd ──▶ Cacher fan-out ──▶ SharedInformer handler
//	      ──▶ queue.Add(namespace/name) ──▶ worker pool
//
// The informer + handlers below are the TEXTBOOK controller pattern; the
// /trigger endpoint exists so an external load tool (k6) can turn HTTP
// concurrency into real apiserver writes. The server half of this path
// (write → etcd → Cacher → watch stream) is dissected frame-by-frame in
// the sibling project, informer-lab.
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"
	"sigs.k8s.io/controller-runtime/pkg/envtest"
)

func runPipeline(ctx context.Context, q *observableQueue, st *stats) {
	// --- 1. envtest: real etcd + kube-apiserver (same setup as informer-lab)
	assets := os.Getenv("KUBEBUILDER_ASSETS")
	if assets == "" {
		if found := findEnvtestAssets(); found != "" {
			assets = found
			os.Setenv("KUBEBUILDER_ASSETS", assets)
		}
	}
	if assets == "" {
		printSetupHelp()
		os.Exit(1)
	}
	env := &envtest.Environment{}
	cfg, err := env.Start()
	if err != nil {
		log.Printf("[SETUP] failed to start envtest: %v", err)
		printSetupHelp()
		os.Exit(1)
	}
	defer func() {
		log.Printf("[SETUP] stopping kube-apiserver + etcd ...")
		_ = env.Stop()
	}()
	log.Printf("[SETUP] control plane up: apiserver=%s", cfg.Host)

	cs := kubernetes.NewForConfigOrDie(cfg)
	ensureDefaultNamespace(ctx, cs)

	// --- 2. ONE SharedInformer on core/v1 ConfigMaps; handlers feed the queue.
	// This is the standard controller wiring: informer -> workqueue.
	factory := informers.NewSharedInformerFactory(cs, 0)
	inf := factory.Core().V1().ConfigMaps().Informer()
	enqueue := func(obj any, event string) {
		key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj) // "default/obj-7"
		if err != nil {
			log.Printf("[ENQUEUE] could not compute key: %v", err)
			return
		}
		q.Add(key, "informer: "+event)
	}
	if _, err := inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    func(obj any) { enqueue(obj, "ADDED") },
		UpdateFunc: func(_, newObj any) { enqueue(newObj, "MODIFIED") },
		DeleteFunc: func(obj any) { enqueue(obj, "DELETED") },
	}); err != nil {
		log.Fatalf("[SETUP] AddEventHandler: %v", err)
	}
	factory.Start(ctx.Done())
	if !cache.WaitForCacheSync(ctx.Done(), inf.HasSynced) {
		log.Fatalf("[SETUP] informer cache failed to sync")
	}

	// --- 3. HTTP trigger endpoint: k6 concurrency -> real apiserver writes
	mux := http.NewServeMux()
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("ok\n")) })
	mux.HandleFunc("POST /trigger/{name}", triggerHandler(cs, st))
	srv := &http.Server{Addr: fmt.Sprintf(":%d", *port), Handler: mux}
	go func() {
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Printf("[SETUP] http server error: %v", err)
		}
	}()
	go func() { <-ctx.Done(); _ = srv.Shutdown(context.Background()) }()

	log.Printf("[SETUP] informer synced; trigger endpoint live:  curl -X POST localhost:%d/trigger/obj-3   (valid names: obj-0 .. obj-%d)", *port, *numKeys-1)
	log.Printf("[SETUP] full path per request: HTTP -> apiserver write -> etcd -> Cacher -> informer -> [ENQUEUE] -> workqueue -> [WORKER-N]")
	go runStatsPrinter(ctx, q, st, true)

	<-ctx.Done()
}

// triggerHandler turns one HTTP POST into one apiserver write (create if
// absent, else update). Every request latency is recorded for the p50/p95/p99
// summary — this number is the apiserver+etcd write path, end to end.
func triggerHandler(cs *kubernetes.Clientset, st *stats) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		name := r.PathValue("name")
		if !validPoolName(name) {
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(http.StatusBadRequest)
			fmt.Fprintf(w, `{"error":"name must be obj-N with 0 <= N < %d"}`+"\n", *numKeys)
			return
		}

		start := time.Now()
		action, rv, err := upsertConfigMap(r.Context(), cs, name)
		elapsed := time.Since(start)
		st.recordLatency(elapsed)
		if n := st.triggers.Add(1); n%int64(*logSample) == 0 {
			log.Printf("[TRIGGER] POST %s action=%s rv=%s took=%s", r.URL.Path, action, rv, elapsed.Round(time.Microsecond))
		}

		w.Header().Set("Content-Type", "application/json")
		if err != nil {
			w.WriteHeader(http.StatusInternalServerError)
			fmt.Fprintf(w, `{"error":%q}`+"\n", err.Error())
			return
		}
		_ = json.NewEncoder(w).Encode(map[string]string{
			"name": name, "action": action, "resourceVersion": rv,
		})
	}
}

// validPoolName restricts triggers to the fixed pool obj-0 .. obj-(keys-1),
// so k6 load concentrates on few objects and the informer enqueues repeat
// keys — which is what exercises dedup downstream.
func validPoolName(name string) bool {
	n, err := strconv.Atoi(strings.TrimPrefix(name, "obj-"))
	return strings.HasPrefix(name, "obj-") && err == nil && n >= 0 && n < *numKeys
}

// upsertConfigMap: create if absent, else increment data.hits. Classic
// get-modify-update with conflict retry (two concurrent triggers on the
// same name collide on resourceVersion — the loser retries).
func upsertConfigMap(ctx context.Context, cs *kubernetes.Clientset, name string) (action, rv string, err error) {
	client := cs.CoreV1().ConfigMaps("default")
	for attempt := 0; attempt < 3; attempt++ {
		cm, err := client.Get(ctx, name, metav1.GetOptions{})
		if apierrors.IsNotFound(err) {
			cm, err = client.Create(ctx, &corev1.ConfigMap{
				ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{"app": "workqueue-lab"}},
				Data:       map[string]string{"hits": "1", "fired": time.Now().Format(time.RFC3339Nano)},
			}, metav1.CreateOptions{})
			if err == nil {
				return "created", cm.ResourceVersion, nil
			}
			if apierrors.IsAlreadyExists(err) {
				continue // someone created it first — loop around and update
			}
			return "", "", err
		}
		if err != nil {
			return "", "", err
		}
		hits, _ := strconv.Atoi(cm.Data["hits"])
		cm.Data["hits"] = strconv.Itoa(hits + 1)
		cm.Data["fired"] = time.Now().Format(time.RFC3339Nano)
		cm, err = client.Update(ctx, cm, metav1.UpdateOptions{})
		if err == nil {
			return "updated", cm.ResourceVersion, nil
		}
		if apierrors.IsConflict(err) {
			continue
		}
		return "", "", err
	}
	return "", "", fmt.Errorf("upsert kept conflicting")
}

// --- envtest helpers (same as informer-lab) ---------------------------------

func ensureDefaultNamespace(ctx context.Context, cs *kubernetes.Clientset) {
	for i := 0; i < 20; i++ {
		_, err := cs.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
			ObjectMeta: metav1.ObjectMeta{Name: "default"},
		}, metav1.CreateOptions{})
		if err == nil || apierrors.IsAlreadyExists(err) {
			log.Printf("[SETUP] namespace \"default\" ready")
			return
		}
		time.Sleep(500 * time.Millisecond)
	}
	log.Fatalf("[SETUP] namespace \"default\" never became available")
}

func findEnvtestAssets() string {
	home, _ := os.UserHomeDir()
	var dirs []string
	for _, pattern := range []string{
		filepath.Join(home, "Library/Application Support/io.kubebuilder.envtest/k8s/*"),
		filepath.Join(home, ".local/share/kubebuilder-envtest/k8s/*"),
	} {
		matches, _ := filepath.Glob(pattern)
		dirs = append(dirs, matches...)
	}
	dirs = append(dirs, "/usr/local/kubebuilder/bin")
	sort.Strings(dirs)
	for i := len(dirs) - 1; i >= 0; i-- {
		if isFile(filepath.Join(dirs[i], "kube-apiserver")) && isFile(filepath.Join(dirs[i], "etcd")) {
			return dirs[i]
		}
	}
	return ""
}

func isFile(p string) bool {
	st, err := os.Stat(p)
	return err == nil && !st.IsDir()
}

func printSetupHelp() {
	log.Printf(`[SETUP] envtest binaries (etcd, kube-apiserver, kubectl) not found.

One-time download:
  1. go install sigs.k8s.io/controller-runtime/tools/setup-envtest@latest
  2. setup-envtest use 1.31.x -p path
  3. either export KUBEBUILDER_ASSETS="<that path>", or just re-run —
     this program auto-detects setup-envtest's default store.
(Already done if you ran informer-lab — the same binaries are reused.)`)
}