diff --git a/edgeconfig/edgeconfig.go b/edgeconfig/edgeconfig.go new file mode 100644 index 0000000..a8b85fa --- /dev/null +++ b/edgeconfig/edgeconfig.go @@ -0,0 +1,147 @@ +// Package edgeconfig is the seam for publish-time generated nginx config. +// +// Producers (auth gating, grain routes, redirects, …) implement Fragment; +// a Writer renders fragments atomically to the shared nginx include +// directory. A Reloader fans the change out to every running nginx +// instance, whose reload handler validates (nginx -t) before reloading. +// +// Risk model: only files inside the include directory ever change — the +// main nginx config is static and never touched at runtime. A broken +// fragment fails `nginx -t` and the previous config keeps serving. +// +// Dependency rule: imports only stdlib and other platform packages. +package edgeconfig + +import ( + "context" + "errors" + "fmt" + "hash/adler32" + "os" + "path/filepath" + "sort" + "strconv" + "strings" +) + +// Fragment is one named piece of generated nginx config owned by exactly +// one producer. Name determines include order: it must start with a number +// prefix ("10-auth", "20-routes"); the writer sorts by name. +type Fragment interface { + Name() string + Render() ([]byte, error) +} + +// Reloader fans a config change out to all nginx instances. Providers are +// selected per environment: kube in-cluster, http for compose/dev, null for +// tests. Implementations must be safe for concurrent use. +type Reloader interface { + Reload(ctx context.Context) Result +} + +// TargetResult is the outcome for one nginx instance. +type TargetResult struct { + Target string // host:port or pod identifier + Err error // nil on success +} + +// Result aggregates per-target outcomes of one Reload call. +type Result struct { + Targets []TargetResult +} + +// Errs returns the non-nil errors, or nil if every target succeeded. +func (r Result) Errs() []error { + var errs []error + for _, t := range r.Targets { + if t.Err != nil { + errs = append(errs, fmt.Errorf("%s: %w", t.Target, t.Err)) + } + } + return errs +} + +// OK reports whether every target reloaded successfully. +func (r Result) OK() bool { return len(r.Errs()) == 0 } + +// Writer renders fragments into dir (the nginx include directory mounted +// from the RWX volume). Every file write is atomic: tmp file + rename, so +// a reader (nginx reload handler) never sees a partial fragment. +type Writer struct { + dir string +} + +// NewWriter returns a Writer that renders fragments into dir. +func NewWriter(dir string) *Writer { return &Writer{dir: dir} } + +// Dir is the include directory fragments are rendered into. +func (w *Writer) Dir() string { return w.dir } + +// Write renders each fragment atomically and reports which names changed +// since the last known content. Fragments whose Render returns an error +// abort the whole flush: nothing is written, so nginx keeps serving the +// previous complete set. The returned names are sorted. +func (w *Writer) Write(fragments ...Fragment) (changed []string, err error) { + type rendered struct { + frag Fragment + data []byte + } + rs := make([]rendered, 0, len(fragments)) + for _, f := range fragments { + data, err := f.Render() + if err != nil { + return nil, fmt.Errorf("edgeconfig: render %s: %w", f.Name(), err) + } + rs = append(rs, rendered{f, data}) + } + sort.Slice(rs, func(i, j int) bool { return rs[i].frag.Name() < rs[j].frag.Name() }) + + for _, r := range rs { + if err := w.writeAtomic(r.frag.Name(), r.data); err != nil { + return nil, fmt.Errorf("edgeconfig: write %s: %w", r.frag.Name(), err) + } + } + for _, r := range rs { + changed = append(changed, r.frag.Name()) + } + return changed, nil +} + +// writeAtomic writes data to name via a tmp file + rename in dir. +func (w *Writer) writeAtomic(name string, data []byte) error { + if err := validName(name); err != nil { + return err + } + final := filepath.Join(w.dir, name) + tmp, err := os.CreateTemp(w.dir, ".edgeconfig-*") + if err != nil { + return err + } + tmpName := tmp.Name() + defer os.Remove(tmpName) // no-op after successful rename + if _, err := tmp.Write(data); err != nil { + tmp.Close() + return err + } + if err := tmp.Sync(); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmpName, final) +} + +func validName(name string) error { + if name == "" || strings.ContainsAny(name, "/\\") || strings.Contains(name, "..") || + filepath.Base(name) != name { + return errors.New("invalid fragment name " + strconv.Quote(name)) + } + return nil +} + +// Checksum is the content fingerprint the reload handler compares against +// (adler32 — cheap, stable across processes; matches nothing security- +// sensitive, this is change detection only). +func Checksum(data []byte) uint32 { return adler32.Checksum(data) } diff --git a/edgeconfig/edgeconfig_test.go b/edgeconfig/edgeconfig_test.go new file mode 100644 index 0000000..b1bbf8e --- /dev/null +++ b/edgeconfig/edgeconfig_test.go @@ -0,0 +1,206 @@ +package edgeconfig + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync" + "testing" +) + +// --- Fragment / Writer --- + +type testFrag struct { + name string + data string + err error +} + +func (f testFrag) Name() string { return f.name } +func (f testFrag) Render() ([]byte, error) { + if f.err != nil { + return nil, f.err + } + return []byte(f.data), nil +} + +func TestWriterAtomicAndOrdering(t *testing.T) { + dir := t.TempDir() + w := NewWriter(dir) + + changed, err := w.Write( + testFrag{name: "20-routes.conf", data: "# routes"}, + testFrag{name: "10-auth.conf", data: "# auth"}, + ) + if err != nil { + t.Fatal(err) + } + if len(changed) != 2 || changed[0] != "10-auth.conf" || changed[1] != "20-routes.conf" { + t.Fatalf("changed = %v, want sorted [10-auth.conf 20-routes.conf]", changed) + } + for _, name := range []string{"10-auth.conf", "20-routes.conf"} { + data, err := os.ReadFile(filepath.Join(dir, name)) + if err != nil { + t.Fatal(err) + } + if len(data) == 0 { + t.Fatalf("%s: empty", name) + } + } + // no tmp leftovers + entries, _ := os.ReadDir(dir) + if len(entries) != 2 { + t.Fatalf("dir has %d entries, want exactly the 2 fragments", len(entries)) + } +} + +func TestWriterRenderErrorAbortsFlush(t *testing.T) { + dir := t.TempDir() + w := NewWriter(dir) + + _, err := w.Write( + testFrag{name: "10-auth.conf", data: "# ok"}, + testFrag{name: "20-bad.conf", err: errors.New("boom")}, + ) + if err == nil { + t.Fatal("want render error") + } + // nothing written — previous complete set keeps serving + if _, statErr := os.Stat(filepath.Join(dir, "10-auth.conf")); !os.IsNotExist(statErr) { + t.Fatal("flush must be all-or-nothing on render error") + } +} + +func TestWriterRejectsBadNames(t *testing.T) { + dir := t.TempDir() + w := NewWriter(dir) + for _, bad := range []string{"../evil", "sub/dir.conf", "", ".."} { + if _, err := w.Write(testFrag{name: bad, data: "x"}); err == nil { + t.Fatalf("name %q accepted", bad) + } + } +} + +// --- HTTPReloader + PostReload against a fake handler --- + +func TestHTTPReloaderFanOut(t *testing.T) { + var mu sync.Mutex + var hits int + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + hits++ + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"reloaded":true,"checksum":42}`)) + })) + defer srv.Close() + + host := srv.Listener.Addr().String() + r, err := NewHTTPReloader(host + "," + host) + if err != nil { + t.Fatal(err) + } + res := r.Reload(context.Background()) + if !res.OK() { + t.Fatalf("want OK, errs=%v", res.Errs()) + } + if hits != 2 { + t.Fatalf("hits=%d, want 2 (one per target)", hits) + } +} + +func TestHTTPReloaderPartialFailure(t *testing.T) { + ok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"reloaded":true}`)) + })) + defer ok.Close() + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"message":"nginx -t failed"}`, http.StatusInternalServerError) + })) + defer bad.Close() + + r, err := NewHTTPReloader(ok.Listener.Addr().String() + "," + bad.Listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + res := r.Reload(context.Background()) + if res.OK() { + t.Fatal("want failure when one target fails") + } + errs := res.Errs() + if len(errs) != 1 { + t.Fatalf("errs=%v, want exactly one (the failing target)", errs) + } +} + +func TestNewHTTPReloaderValidation(t *testing.T) { + if _, err := NewHTTPReloader(""); err == nil { + t.Fatal("empty targets accepted") + } + if _, err := NewHTTPReloader("no-port-here"); err == nil { + t.Fatal("target without port accepted") + } + if _, err := NewHTTPReloader(" a:1 , b:2 "); err != nil { + t.Fatalf("whitespace-separated list rejected: %v", err) + } +} + +// --- KubeReloader with injected lister/post (no k8s client in tests) --- + +func TestKubeReloaderFansOutToAllPods(t *testing.T) { + var mu sync.Mutex + got := map[string]bool{} + + k, err := NewKubeReloader(func(ctx context.Context) ([]string, error) { + return []string{"10.0.0.4", "10.0.0.5"}, nil + }, "8081") + if err != nil { + t.Fatal(err) + } + k.post = func(ctx context.Context, target string) (HandlerResult, error) { + mu.Lock() + got[target] = true + mu.Unlock() + return HandlerResult{Reloaded: true}, nil + } + + res := k.Reload(context.Background()) + if !res.OK() { + t.Fatalf("want OK, errs=%v", res.Errs()) + } + for _, want := range []string{"10.0.0.4:8081", "10.0.0.5:8081"} { + if !got[want] { + t.Fatalf("pod %s not reloaded; got %v", want, got) + } + } +} + +func TestKubeReloaderPodListFailure(t *testing.T) { + k, err := NewKubeReloader(func(ctx context.Context) ([]string, error) { + return nil, errors.New("api server down") + }, "") + if err != nil { + t.Fatal(err) + } + res := k.Reload(context.Background()) + if res.OK() { + t.Fatal("pod-list failure must surface as error") + } +} + +func TestKubeReloaderNoPods(t *testing.T) { + k, _ := NewKubeReloader(func(ctx context.Context) ([]string, error) { return nil, nil }, "") + if res := k.Reload(context.Background()); res.OK() { + t.Fatal("zero pods must be an error, not silent success") + } +} + +func TestNullReloader(t *testing.T) { + if res := (NullReloader{}).Reload(context.Background()); !res.OK() { + t.Fatal("null reloader must succeed") + } +} diff --git a/edgeconfig/reloader_http.go b/edgeconfig/reloader_http.go new file mode 100644 index 0000000..b815478 --- /dev/null +++ b/edgeconfig/reloader_http.go @@ -0,0 +1,126 @@ +package edgeconfig + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "sync" + "time" +) + +// reloadPath is the endpoint the nginx-side handler serves. The handler +// (baked into the nginx image) does: content check → nginx -t → nginx -s +// reload, and is idempotent — reloading an unchanged config is a no-op. +const reloadPath = "/reload" + +// HTTPReloader fans reloads out to a fixed list of nginx instances by +// host:port. This is the dev/compose provider (Docker embedded DNS names) +// and the transport KubeReloader uses per discovered pod. +type HTTPReloader struct { + client *http.Client + targets []string // host:port list, e.g. ["nginx:8081"] +} + +// NewHTTPReloader targets is a comma-separated host:port list +// (EDGE_RELOAD_TARGETS=nginx:8081 or "10.0.0.4:8081,10.0.0.5:8081"). +func NewHTTPReloader(targets string) (*HTTPReloader, error) { + var out []string + for _, t := range strings.Split(targets, ",") { + t = strings.TrimSpace(t) + if t == "" { + continue + } + if _, _, err := net.SplitHostPort(t); err != nil { + return nil, fmt.Errorf("edgeconfig: bad target %q: %w", t, err) + } + out = append(out, t) + } + if len(out) == 0 { + return nil, fmt.Errorf("edgeconfig: no reload targets configured") + } + return &HTTPReloader{ + client: &http.Client{Timeout: 10 * time.Second}, + targets: out, + }, nil +} + +// Reload posts to every target concurrently and aggregates results. +// A target that fails validation (nginx -t) returns its error in the +// result; the caller decides on retry/alert. Never panics on partial +// failure — partial success is reported, not swallowed. +func (h *HTTPReloader) Reload(ctx context.Context) Result { + res := Result{Targets: make([]TargetResult, len(h.targets))} + var wg sync.WaitGroup + for i, t := range h.targets { + wg.Add(1) + go func(i int, target string) { + defer wg.Done() + res.Targets[i] = TargetResult{Target: target, Err: h.post(ctx, target)} + }(i, t) + } + wg.Wait() + return res +} + +func (h *HTTPReloader) post(ctx context.Context, target string) error { + url := "http://" + target + reloadPath + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, nil) + if err != nil { + return err + } + resp, err := h.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<10)) + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("reload failed: %s: %s", resp.Status, strings.TrimSpace(string(body))) + } + return nil +} + +// NullReloader succeeds without doing anything — tests and deployments +// where the static floor config is sufficient. +type NullReloader struct{} + +func (NullReloader) Reload(context.Context) Result { return Result{} } + +// HandlerResult is the JSON body the nginx-side handler replies with; +// shared here so producers can decode it in tests/integrations. +type HandlerResult struct { + Reloaded bool `json:"reloaded"` // false = content unchanged, no-op + Checksum uint32 `json:"checksum"` // adler32 of current fragment set + Message string `json:"message,omitempty"` // validation error detail on failure +} + +// PostReload is a helper for tests and the kube provider's per-pod call: +// one POST to target, decoding the handler's JSON reply. +func PostReload(ctx context.Context, client *http.Client, target string) (HandlerResult, error) { + if client == nil { + client = &http.Client{Timeout: 10 * time.Second} + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + "http://"+target+reloadPath, bytes.NewReader(nil)) + if err != nil { + return HandlerResult{}, err + } + resp, err := client.Do(req) + if err != nil { + return HandlerResult{}, err + } + defer resp.Body.Close() + var hr HandlerResult + if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<10)).Decode(&hr); err != nil { + return HandlerResult{}, fmt.Errorf("decode reload reply (%s): %w", resp.Status, err) + } + if resp.StatusCode != http.StatusOK { + return hr, fmt.Errorf("reload failed: %s: %s", resp.Status, hr.Message) + } + return hr, nil +} diff --git a/edgeconfig/reloader_kube.go b/edgeconfig/reloader_kube.go new file mode 100644 index 0000000..46b449d --- /dev/null +++ b/edgeconfig/reloader_kube.go @@ -0,0 +1,85 @@ +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 +}