package edgeconfig import ( "context" "fmt" "sync" ) // KubeReloader fans reloads out to every pod of the nginx Deployment by // talking to the Kubernetes API in-cluster. It exists so a change reaches // ALL replicas (no single-pod-per-Service drift); per-pod results are // returned for retry/alerting. // // The actual API calls (pod list + per-pod POST) are behind PodLister and // reloadFunc so this file stays stdlib-only per the platform dependency // rule — the k8s client wiring lives in the consuming module (backoffice), // which injects them at startup. type KubeReloader struct { mu sync.Mutex port string lister PodLister post func(ctx context.Context, target string) (HandlerResult, error) } // PodLister returns the addresses (ip:port or host:port) of all nginx pods // matching the selector right now. Implemented by the consumer with // client-go: corev1 Pods(ns).List(labelSelector) → podIP:port. type PodLister func(ctx context.Context) ([]string, error) // NewKubeReloader builds a KubeReloader. lister must be non-nil; port is // the reload-handler port on each pod (default "8081"). func NewKubeReloader(lister PodLister, port string) (*KubeReloader, error) { if lister == nil { return nil, fmt.Errorf("edgeconfig: kube reloader needs a pod lister") } if port == "" { port = "8081" } return &KubeReloader{ lister: lister, port: port, post: func(ctx context.Context, target string) (HandlerResult, error) { return PostReload(ctx, nil, target) }, }, nil } // Reload lists current nginx pods and POSTs /reload to each concurrently. // A pod that appears after listing misses this round — bounded by the // slow poll inside the nginx image (it loads fresh config on start anyway). func (k *KubeReloader) Reload(ctx context.Context) Result { k.mu.Lock() defer k.mu.Unlock() targets, err := k.lister(ctx) if err != nil { return Result{Targets: []TargetResult{{Target: "(pod-list)", Err: err}}} } if len(targets) == 0 { return Result{Targets: []TargetResult{{ Target: "(pod-list)", Err: fmt.Errorf("no nginx pods found"), }}} } res := Result{Targets: make([]TargetResult, len(targets))} var wg sync.WaitGroup for i, ip := range targets { wg.Add(1) go func(i int, ip string) { defer wg.Done() res.Targets[i] = TargetResult{ Target: ip, Err: firstErr(k.post(ctx, joinHostPort(ip, k.port))), } }(i, ip) } wg.Wait() return res } func firstErr(_ HandlerResult, err error) error { return err } func joinHostPort(host, port string) string { return host + ":" + port }