diff --git a/docs/architecture.md b/docs/architecture.md index ae490c3..6da6eea 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,12 +1,10 @@ # Architecture -> **Status:** the operator is built through Step 6 (lease store) of +> **Status:** the operator is built through Step 7 (discovery API) of > [docs/plans/2026-08-07-1747-proxy-operator.md](plans/2026-08-07-1747-proxy-operator.md). -> This document currently covers the event/reconcile flow; the components -> table and the Decisions section arrive with Step 10. The lease store -> (`internal/lease/`) is HTTP-driven, not cluster-event-driven, so its -> diagram lands together with the discovery API in Step 7; the orphan-GC -> flow lands with Step 9. +> This document currently covers the event/reconcile flow and the +> HTTP-driven lease/discovery path; the components table and the Decisions +> section arrive with Step 10, and the orphan-GC flow lands with Step 9. ## Event flow: cluster events → reconciler functions @@ -185,3 +183,48 @@ Consequence worth knowing: `status.lastHealthCheckTime` is the time of the last *status-affecting* probe, not the most recent probe — suppressed probes deliberately never write status. True probe recency will live in metrics (Step 9). + +### 7. Discovery + lease API (`internal/discovery/`, `internal/lease/`) + +HTTP-driven, not cluster-event-driven: crawler clients call in; the only +Kubernetes interaction is reading Proxies from the manager's cache. The +server is a non-leader-elected Runnable (all replicas would serve, but the +deployment ships `replicas: 1` because lease state is per-process — an +operator restart drops all leases and cooldowns, a documented caveat). + +```text +crawler client + │ Authorization: Bearer $DISCOVERY_TOKEN (empty token = auth disabled, loud startup warning) + ▼ +Server.handler() middleware, outermost first (server.go) + recover → request-log → MaxBytesReader(64KiB) → bearer auth (constant-time; /healthz exempt) + │ + ├─ GET /healthz ──► 200 ok (unauthenticated) + │ + ├─ GET /v1/proxies?attr.k=v&healthy=true (handlers.go) + │ Reader.List(Proxies) ── manager cache + │ filter: attributes equality + Healthy condition + │ + Store.Counts() for activeLeases + │ ──► 200 {"proxies":[...], "count":N} (empty list is 200, not 404) + │ + ├─ POST /v1/leases {"selector":{...},"ttlSeconds":300,"target":"..."} + │ Reader.List → filter selector; unhealthy matches counted, not offered + │ Store.Acquire(healthy candidates, target, ttl) ── one lock: select+insert + │ │ selection: fewest active leases, then latency, then name + │ ├─ granted ──► 201 {leaseID, proxy:{...}, expiresAt, ttlSeconds} + │ └─ ErrNoMatch ──► 409 {"error":"no_match", considered, atCapacity, + │ inCooldown, unhealthy} + │ + ├─ DELETE /v1/leases/{id} ──► Store.Release ──► always 204 (idempotent) + │ + └─ POST /v1/leases/{id}/report {"result":"ok|rate_limited|banned","target":"..."} + Store.Report ── rate_limited/banned ⇒ cooldown[{proxy,target}] for + │ CooldownWindow (target falls back: report → lease → global) + ├─ 204 │ 400 invalid_result │ 404 unknown_lease + └─ an expired lease still resolves for CooldownWindow past its TTL — + a late report lands exactly when the proxy is being rate-limited + +Store.Start(ctx) ── manager Runnable, NOT leader-elected: sweeps expired + leases + cooldowns; correctness never depends on the + sweep (every read checks ExpiresAt against the clock) +``` diff --git a/docs/plans-executions/2026-08-07-1747-proxy-operator.md b/docs/plans-executions/2026-08-07-1747-proxy-operator.md index 7795d1a..607c5bf 100644 --- a/docs/plans-executions/2026-08-07-1747-proxy-operator.md +++ b/docs/plans-executions/2026-08-07-1747-proxy-operator.md @@ -11,7 +11,7 @@ Pairs with [docs/plans/2026-08-07-1747-proxy-operator.md](../plans/2026-08-07-17 - [x] Step 4 — Reconciler (`internal/controller/`) - [x] Step 5 — Health engine (`internal/health/`) - [x] Step 6 — Lease store (`internal/lease/`) -- [ ] Step 7 — Discovery API (`internal/discovery/`) +- [x] Step 7 — Discovery API (`internal/discovery/`) - [ ] Step 8 — GCP provider (`internal/provider/gcp/`) - [ ] Step 9 — Orphan GC + metrics - [ ] Step 10 — Wiring, config, docs @@ -758,3 +758,56 @@ Worth noting: `docs/architecture.md` was not extended this step — the lease store is HTTP-driven, not cluster-event-driven, so its diagram belongs with the discovery API and lands in Step 7 (banner updated to say so). + +## Step 7 — Discovery API (`internal/discovery/`) + +Implemented `server.go` (Runnable + middleware chain) and `handlers.go` +(the four endpoints + `proxyView` wire shape) per the plan: stdlib +`http.ServeMux` method+wildcard routing (no third-party router — see the +plan clarification commit: this is a stdlib feature since Go 1.22, the +project stays on the pinned Go 1.26), middleware outermost-first recover → +request-log → `MaxBytesReader(64KiB)` → constant-time bearer auth with +`/healthz` exempt, empty `DISCOVERY_TOKEN` serving unauthenticated with a +loud startup warning, `NeedLeaderElection() = false` with the plan's +runnable-ordering rationale in the doc comment, and graceful `Shutdown` +with a 10 s grace on ctx cancel. + +The `LeaseStore` interface landed consumer-side in this package (spec §8 +wants handlers swappable to a CRD/Redis store); `internal/lease.*Store` +satisfies it without modification. + +Judgment calls the plan/spec left open: + +- **409 arithmetic:** the store only ever sees healthy candidates, so its + `Considered` excludes unhealthy matches. The handler counts unhealthy + selector-matches itself and reports `considered = healthy + unhealthy`, + keeping the plan's example arithmetic (7 = 2+2+3) consistent. +- **TTL handling:** omitted/zero `ttlSeconds` → 300 s default; negative or + above `MaxLeaseTTL` (default 1h, flag in Step 10) → 400 `invalid_ttl` + rather than silent clamping — a client asking for a week-long lease + should find out, not get an hour quietly. +- **Grant response includes the fresh `activeLeases`** (the just-granted + lease counted), read back via `Store.Counts()` after the acquire. +- Proxies with a deletionTimestamp are filtered out of both list and + candidate selection — a proxy mid-teardown shouldn't be advertised. + +Tests (87.3% coverage, `-race -count=2` clean, green on first run): +httptest over the real handler chain with a fake cache reader and a real +`lease.Store` — auth on/off/wrong-token/healthz-exempt, list filtering +(attributes, healthy, combined, empty-is-200), grant shape (201, default +TTL, lowest-latency pick, RFC3339 expiresAt, activeLeases=1), the full +409 body arithmetic, invalid TTL/body/result, idempotent 204 release, +report→cooldown→409 round-trip, 404 on unknown lease, and a real +`Start` on `127.0.0.1:0` (via the new `BoundAddr()` accessor) serving +healthz then shutting down cleanly on cancel. + +```bash +go test -race -count=2 ./internal/discovery/ +make test # whole repo green +``` + +Worth noting: `go mod tidy` promoted `github.com/go-logr/logr` from +indirect to direct (the server holds a `logr.Logger` field). The +`--discovery-addr`, `--max-lease-ttl` flags and the `DISCOVERY_TOKEN` +Secret mount arrive with `cmd/main.go` in Step 10. `docs/architecture.md` +gained §7 covering the whole HTTP path and the store's sweep Runnable. diff --git a/go.mod b/go.mod index 443b831..0eb0c93 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator go 1.26.0 require ( + github.com/go-logr/logr v1.4.3 github.com/onsi/ginkgo/v2 v2.27.4 github.com/onsi/gomega v1.39.0 k8s.io/api v0.36.0 @@ -26,7 +27,6 @@ require ( github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect - github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect github.com/go-openapi/jsonpointer v0.21.0 // indirect diff --git a/internal/discovery/handlers.go b/internal/discovery/handlers.go new file mode 100644 index 0000000..8975d0b --- /dev/null +++ b/internal/discovery/handlers.go @@ -0,0 +1,235 @@ +package discovery + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "slices" + "strings" + "time" + + apimeta "k8s.io/apimachinery/pkg/api/meta" + "sigs.k8s.io/controller-runtime/pkg/client" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease" +) + +// proxyView is the wire shape of one proxy in list and lease responses. +type proxyView struct { + ID string `json:"id"` // namespace/name + IP string `json:"ip"` + Port int32 `json:"port"` + Attributes map[string]string `json:"attributes,omitempty"` + Phase string `json:"phase"` + Healthy bool `json:"healthy"` + LatencyMillis int64 `json:"latencyMillis"` + ActiveLeases int `json:"activeLeases"` + MaxLeases int32 `json:"maxLeases"` +} + +func viewOf(p *crawlv1alpha1.Proxy, activeLeases int) proxyView { + return proxyView{ + ID: client.ObjectKeyFromObject(p).String(), + IP: p.EffectiveHost(), + Port: p.EffectivePort(), + Attributes: p.Spec.Attributes, + Phase: string(p.Status.Phase), + Healthy: isHealthy(p), + LatencyMillis: p.Status.LatencyMillis, + ActiveLeases: activeLeases, + MaxLeases: p.MaxLeasesOrDefault(), + } +} + +func isHealthy(p *crawlv1alpha1.Proxy) bool { + return apimeta.IsStatusConditionTrue(p.Status.Conditions, crawlv1alpha1.ConditionHealthy) +} + +// matchesAttributes is spec.attributes equality: every selector pair must +// be present verbatim. +func matchesAttributes(p *crawlv1alpha1.Proxy, selector map[string]string) bool { + for k, v := range selector { + if p.Spec.Attributes[k] != v { + return false + } + } + return true +} + +// GET /v1/proxies?attr.=&healthy=true|false +func (s *Server) handleListProxies(w http.ResponseWriter, r *http.Request) { + selector := map[string]string{} + var healthyFilter *bool + for key, values := range r.URL.Query() { + switch { + case key == "healthy": + switch values[0] { + case "true": + healthyFilter = ptr(true) + case "false": + healthyFilter = ptr(false) + default: + writeError(w, http.StatusBadRequest, "invalid_query", + fmt.Sprintf("healthy must be true or false, got %q", values[0])) + return + } + case strings.HasPrefix(key, "attr."): + selector[strings.TrimPrefix(key, "attr.")] = values[0] + } + } + + var list crawlv1alpha1.ProxyList + if err := s.Reader.List(r.Context(), &list); err != nil { + s.log.Error(err, "listing proxies") + writeError(w, http.StatusInternalServerError, "internal", "listing proxies failed") + return + } + counts := s.Store.Counts() + + views := []proxyView{} + for i := range list.Items { + p := &list.Items[i] + if !p.DeletionTimestamp.IsZero() || !matchesAttributes(p, selector) { + continue + } + v := viewOf(p, counts[client.ObjectKeyFromObject(p).String()]) + if healthyFilter != nil && v.Healthy != *healthyFilter { + continue + } + views = append(views, v) + } + slices.SortFunc(views, func(a, b proxyView) int { return strings.Compare(a.ID, b.ID) }) + + writeJSON(w, http.StatusOK, map[string]any{"proxies": views, "count": len(views)}) +} + +type leaseRequest struct { + Selector map[string]string `json:"selector"` + TTLSeconds int64 `json:"ttlSeconds"` + Target string `json:"target"` +} + +type leaseResponse struct { + LeaseID string `json:"leaseID"` + Proxy proxyView `json:"proxy"` + ExpiresAt time.Time `json:"expiresAt"` + TTLSeconds int64 `json:"ttlSeconds"` +} + +// POST /v1/leases +func (s *Server) handleAcquireLease(w http.ResponseWriter, r *http.Request) { + var req leaseRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid_body", err.Error()) + return + } + + ttl := time.Duration(req.TTLSeconds) * time.Second + if req.TTLSeconds == 0 { + ttl = defaultTTL + } + if ttl < 0 || ttl > s.MaxLeaseTTL { + writeError(w, http.StatusBadRequest, "invalid_ttl", + fmt.Sprintf("ttlSeconds must be between 1 and %d", int64(s.MaxLeaseTTL.Seconds()))) + return + } + + var list crawlv1alpha1.ProxyList + if err := s.Reader.List(r.Context(), &list); err != nil { + s.log.Error(err, "listing proxies for lease") + writeError(w, http.StatusInternalServerError, "internal", "listing proxies failed") + return + } + + // The store gets only healthy, live candidates; unhealthy matches are + // counted here because the store never sees them. + var candidates []lease.Candidate + byKey := map[string]*crawlv1alpha1.Proxy{} + unhealthy := 0 + for i := range list.Items { + p := &list.Items[i] + if !p.DeletionTimestamp.IsZero() || !matchesAttributes(p, req.Selector) { + continue + } + if !isHealthy(p) { + unhealthy++ + continue + } + key := client.ObjectKeyFromObject(p).String() + byKey[key] = p + candidates = append(candidates, lease.Candidate{ + Proxy: key, + MaxLeases: p.MaxLeasesOrDefault(), + Latency: time.Duration(p.Status.LatencyMillis) * time.Millisecond, + }) + } + + granted, stats, err := s.Store.Acquire(r.Context(), lease.AcquireRequest{ + Candidates: candidates, + Target: req.Target, + TTL: ttl, + }) + if errors.Is(err, lease.ErrNoMatch) { + writeJSON(w, http.StatusConflict, map[string]any{ + "error": "no_match", + "message": "no healthy proxy with free capacity matched the selector", + "considered": stats.Considered + unhealthy, + "atCapacity": stats.AtCapacity, + "inCooldown": stats.InCooldown, + "unhealthy": unhealthy, + }) + return + } + if err != nil { + writeError(w, http.StatusBadRequest, "invalid_request", err.Error()) + return + } + + writeJSON(w, http.StatusCreated, leaseResponse{ + LeaseID: granted.ID, + Proxy: viewOf(byKey[granted.Proxy], s.Store.Counts()[granted.Proxy]), + ExpiresAt: granted.ExpiresAt, + TTLSeconds: int64(ttl.Seconds()), + }) +} + +// DELETE /v1/leases/{id} — early release, always 204: releasing an unknown +// or already expired lease is not an error. +func (s *Server) handleReleaseLease(w http.ResponseWriter, r *http.Request) { + s.Store.Release(r.Context(), r.PathValue("id")) + w.WriteHeader(http.StatusNoContent) +} + +type reportRequest struct { + Result string `json:"result"` + Target string `json:"target"` +} + +// POST /v1/leases/{id}/report +func (s *Server) handleReportLease(w http.ResponseWriter, r *http.Request) { + var req reportRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid_body", err.Error()) + return + } + result, ok := lease.ParseResult(req.Result) + if !ok { + writeError(w, http.StatusBadRequest, "invalid_result", + fmt.Sprintf("result must be one of ok, rate_limited, banned; got %q", req.Result)) + return + } + if err := s.Store.Report(r.Context(), r.PathValue("id"), result, req.Target); err != nil { + if errors.Is(err, lease.ErrUnknownLease) { + writeError(w, http.StatusNotFound, "unknown_lease", "no such lease") + return + } + s.log.Error(err, "reporting lease", "leaseID", r.PathValue("id")) + writeError(w, http.StatusInternalServerError, "internal", "report failed") + return + } + w.WriteHeader(http.StatusNoContent) +} + +func ptr[T any](v T) *T { return &v } diff --git a/internal/discovery/server.go b/internal/discovery/server.go new file mode 100644 index 0000000..16445fe --- /dev/null +++ b/internal/discovery/server.go @@ -0,0 +1,225 @@ +// Package discovery implements the HTTP API crawler clients use to find +// and lease proxies: list healthy proxies filtered by attributes, acquire a +// TTL-based lease, release it early, and report how a target treated the +// proxy. Reads go through the manager's informer cache; lease state lives +// in the injected store. +package discovery + +import ( + "context" + "crypto/subtle" + "encoding/json" + "net" + "net/http" + "strings" + "sync" + "time" + + "github.com/go-logr/logr" + "sigs.k8s.io/controller-runtime/pkg/client" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease" +) + +// LeaseStore is what the handlers need from a lease backend. Defined here, +// consumer-side, so a CRD- or Redis-backed store can replace the in-memory +// one (which internal/lease's *Store satisfies) without touching handlers. +type LeaseStore interface { + Acquire(ctx context.Context, req lease.AcquireRequest) (*lease.Lease, lease.AcquireStats, error) + Release(ctx context.Context, id string) + Report(ctx context.Context, id string, result lease.Result, target string) error + Counts() map[string]int +} + +const ( + defaultAddr = ":8090" + defaultTTL = 5 * time.Minute + defaultMaxTTL = time.Hour + maxBodyBytes = 64 << 10 + shutdownGrace = 10 * time.Second + readHeadTimeout = 5 * time.Second +) + +// Server serves the discovery API as a manager Runnable. +type Server struct { + // Reader lists Proxies from the manager's cache. + Reader client.Reader + // Store is the lease backend. + Store LeaseStore + // Addr is the listen address (default ":8090"; --discovery-addr). + Addr string + // Token is the static bearer token from DISCOVERY_TOKEN. Empty + // disables auth — allowed for the prototype, but loudly warned about + // at startup, because in-cluster that is a silent security hole. + Token string + // MaxLeaseTTL caps requested lease TTLs (default 1h; --max-lease-ttl). + MaxLeaseTTL time.Duration + + log logr.Logger + + mu sync.Mutex + boundAddr string +} + +// NeedLeaderElection is false, and the deployment ships replicas: 1. +// Verified against controller-runtime's runnable ordering: caches start and +// sync before non-leader-election runnables, so cache reads here are safe. +// If this were leader-elected, non-leader replicas would refuse connections +// while still being Service endpoints. The 1-replica constraint comes from +// lease state being per-process — both facts are README caveats. +func (s *Server) NeedLeaderElection() bool { return false } + +// BoundAddr returns the actual listen address once Start has bound it — +// meaningful when Addr uses port 0 (tests). +func (s *Server) BoundAddr() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.boundAddr +} + +// Start listens and serves until ctx ends, then shuts down gracefully with +// a 10-second grace period. +func (s *Server) Start(ctx context.Context) error { + if s.Addr == "" { + s.Addr = defaultAddr + } + if s.MaxLeaseTTL == 0 { + s.MaxLeaseTTL = defaultMaxTTL + } + s.log = logf.FromContext(ctx).WithName("discovery") + if s.Token == "" { + s.log.Info("WARNING: DISCOVERY_TOKEN is empty — the discovery API is served without authentication") + } + + ln, err := net.Listen("tcp", s.Addr) + if err != nil { + return err + } + s.mu.Lock() + s.boundAddr = ln.Addr().String() + s.mu.Unlock() + + srv := &http.Server{ + Handler: s.handler(), + ReadHeaderTimeout: readHeadTimeout, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + IdleTimeout: 60 * time.Second, + } + + errCh := make(chan error, 1) + go func() { errCh <- srv.Serve(ln) }() + s.log.Info("discovery API listening", "addr", s.boundAddr) + + select { + case err := <-errCh: + return err + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGrace) + defer cancel() + if err := srv.Shutdown(shutdownCtx); err != nil { + return err + } + <-errCh // always http.ErrServerClosed after a clean Shutdown + return nil + } +} + +// handler assembles the mux and the middleware chain, outermost first: +// recover → request-log → body-size cap → bearer auth. +func (s *Server) handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok\n")) + }) + mux.HandleFunc("GET /v1/proxies", s.handleListProxies) + mux.HandleFunc("POST /v1/leases", s.handleAcquireLease) + mux.HandleFunc("DELETE /v1/leases/{id}", s.handleReleaseLease) + mux.HandleFunc("POST /v1/leases/{id}/report", s.handleReportLease) + + var h http.Handler = mux + h = s.authMiddleware(h) + h = maxBytesMiddleware(h) + h = s.logMiddleware(h) + h = s.recoverMiddleware(h) + return h +} + +func (s *Server) recoverMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer func() { + if p := recover(); p != nil { + s.log.Error(nil, "panic in discovery handler", "panic", p, "path", r.URL.Path) + writeError(w, http.StatusInternalServerError, "internal", "internal server error") + } + }() + next.ServeHTTP(w, r) + }) +} + +// statusRecorder captures the response code for the request log. +type statusRecorder struct { + http.ResponseWriter + status int +} + +func (r *statusRecorder) WriteHeader(code int) { + r.status = code + r.ResponseWriter.WriteHeader(code) +} + +func (s *Server) logMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/healthz" { + next.ServeHTTP(w, r) // probes are noise + return + } + rec := &statusRecorder{ResponseWriter: w, status: http.StatusOK} + start := time.Now() + next.ServeHTTP(rec, r) + s.log.Info("request", + "method", r.Method, "path", r.URL.Path, + "status", rec.status, "duration", time.Since(start).String()) + }) +} + +func maxBytesMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + next.ServeHTTP(w, r) + }) +} + +func (s *Server) authMiddleware(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if s.Token == "" || r.URL.Path == "/healthz" { + next.ServeHTTP(w, r) + return + } + token, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ") + if !ok || subtle.ConstantTimeCompare([]byte(token), []byte(s.Token)) != 1 { + writeError(w, http.StatusUnauthorized, "unauthorized", "missing or invalid bearer token") + return + } + next.ServeHTTP(w, r) + }) +} + +// errorBody is the shared error shape: +// {"error":"","message":""}. +type errorBody struct { + Error string `json:"error"` + Message string `json:"message"` +} + +func writeError(w http.ResponseWriter, status int, code, message string) { + writeJSON(w, status, errorBody{Error: code, Message: message}) +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} diff --git a/internal/discovery/server_test.go b/internal/discovery/server_test.go new file mode 100644 index 0000000..6fbaed8 --- /dev/null +++ b/internal/discovery/server_test.go @@ -0,0 +1,386 @@ +package discovery + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease" +) + +func testProxy(name string, attrs map[string]string, healthy bool, mut ...func(*crawlv1alpha1.Proxy)) *crawlv1alpha1.Proxy { + p := &crawlv1alpha1.Proxy{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, Namespace: "default", UID: types.UID("uid-" + name), + }, + Spec: crawlv1alpha1.ProxySpec{ + Mode: crawlv1alpha1.ModeExternal, + Endpoint: &crawlv1alpha1.EndpointSpec{Host: "10.0.0.1", Port: 3128}, + Attributes: attrs, + }, + } + p.Status.IP = "10.0.0.1" + p.Status.Phase = crawlv1alpha1.PhaseReady + status := metav1.ConditionFalse + if healthy { + status = metav1.ConditionTrue + } + p.Status.Conditions = []metav1.Condition{{ + Type: crawlv1alpha1.ConditionHealthy, Status: status, + Reason: "Probing", LastTransitionTime: metav1.Now(), + }} + for _, m := range mut { + m(p) + } + return p +} + +func withMaxLeases(n int32) func(*crawlv1alpha1.Proxy) { + return func(p *crawlv1alpha1.Proxy) { p.Spec.MaxLeases = &n } +} + +func withLatency(ms int64) func(*crawlv1alpha1.Proxy) { + return func(p *crawlv1alpha1.Proxy) { p.Status.LatencyMillis = ms } +} + +// newTestServer wires the handler chain to a fake cache reader and a real +// lease store, served over httptest. +func newTestServer(t *testing.T, token string, proxies ...*crawlv1alpha1.Proxy) (*httptest.Server, *Server) { + t.Helper() + s := runtime.NewScheme() + if err := crawlv1alpha1.AddToScheme(s); err != nil { + t.Fatalf("scheme: %v", err) + } + builder := fake.NewClientBuilder().WithScheme(s) + for _, p := range proxies { + builder = builder.WithObjects(p) + } + srv := &Server{ + Reader: builder.Build(), + Store: lease.NewStore(15 * time.Minute), + Token: token, + MaxLeaseTTL: time.Hour, + } + ts := httptest.NewServer(srv.handler()) + t.Cleanup(ts.Close) + return ts, srv +} + +type response struct { + status int + body map[string]any +} + +func do(t *testing.T, ts *httptest.Server, method, path, token string, body any) response { + t.Helper() + var reader io.Reader + if body != nil { + if s, ok := body.(string); ok { + reader = bytes.NewBufferString(s) + } else { + b, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshaling request body: %v", err) + } + reader = bytes.NewBuffer(b) + } + } + req, err := http.NewRequestWithContext(context.Background(), method, ts.URL+path, reader) + if err != nil { + t.Fatalf("building request: %v", err) + } + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + resp, err := ts.Client().Do(req) + if err != nil { + t.Fatalf("%s %s: %v", method, path, err) + } + defer func() { _ = resp.Body.Close() }() + out := response{status: resp.StatusCode} + raw, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("reading response: %v", err) + } + if len(raw) > 0 && resp.Header.Get("Content-Type") == "application/json" { + if err := json.Unmarshal(raw, &out.body); err != nil { + t.Fatalf("decoding response %q: %v", raw, err) + } + } + return out +} + +func TestAuth(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "sekrit", testProxy("p1", nil, true)) + + if got := do(t, ts, http.MethodGet, "/v1/proxies", "", nil); got.status != http.StatusUnauthorized { + t.Errorf("no token: status %d, want 401", got.status) + } + if got := do(t, ts, http.MethodGet, "/v1/proxies", "wrong", nil); got.status != http.StatusUnauthorized { + t.Errorf("wrong token: status %d, want 401", got.status) + } + if got := do(t, ts, http.MethodGet, "/v1/proxies", "sekrit", nil); got.status != http.StatusOK { + t.Errorf("correct token: status %d, want 200", got.status) + } + if got := do(t, ts, http.MethodGet, "/healthz", "", nil); got.status != http.StatusOK { + t.Errorf("healthz without token: status %d, want 200 (always unauthenticated)", got.status) + } +} + +func TestAuth_disabledWithEmptyToken(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "", testProxy("p1", nil, true)) + if got := do(t, ts, http.MethodGet, "/v1/proxies", "", nil); got.status != http.StatusOK { + t.Errorf("status %d, want 200 with auth disabled", got.status) + } +} + +func TestListProxies(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "", + testProxy("eu-healthy", map[string]string{"geo": "eu", "purpose": "crawl"}, true), + testProxy("eu-sick", map[string]string{"geo": "eu"}, false), + testProxy("us-healthy", map[string]string{"geo": "us"}, true), + ) + + tests := []struct { + name string + query string + wantCount int + wantFirst string + }{ + {name: "no filter returns everything", query: "", wantCount: 3, wantFirst: "default/eu-healthy"}, + {name: "healthy filter", query: "?healthy=true", wantCount: 2}, + {name: "unhealthy filter", query: "?healthy=false", wantCount: 1, wantFirst: "default/eu-sick"}, + {name: "attribute filter", query: "?attr.geo=eu", wantCount: 2}, + {name: "attribute and health combined", query: "?attr.geo=eu&healthy=true", wantCount: 1, wantFirst: "default/eu-healthy"}, + {name: "two attributes must both match", query: "?attr.geo=eu&attr.purpose=crawl", wantCount: 1, wantFirst: "default/eu-healthy"}, + {name: "no matches is 200 with count 0", query: "?attr.geo=mars", wantCount: 0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := do(t, ts, http.MethodGet, "/v1/proxies"+tc.query, "", nil) + if got.status != http.StatusOK { + t.Fatalf("status %d, want 200", got.status) + } + count := int(got.body["count"].(float64)) + proxies := got.body["proxies"].([]any) + if count != tc.wantCount || len(proxies) != tc.wantCount { + t.Fatalf("count = %d (len %d), want %d", count, len(proxies), tc.wantCount) + } + if tc.wantFirst != "" { + first := proxies[0].(map[string]any) + if first["id"] != tc.wantFirst { + t.Errorf("first id = %v, want %s", first["id"], tc.wantFirst) + } + } + }) + } + + t.Run("invalid healthy value is 400", func(t *testing.T) { + t.Parallel() + if got := do(t, ts, http.MethodGet, "/v1/proxies?healthy=maybe", "", nil); got.status != http.StatusBadRequest { + t.Errorf("status %d, want 400", got.status) + } + }) +} + +func TestAcquireLease_grantShape(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "", + testProxy("eu1", map[string]string{"geo": "eu"}, true, withLatency(30)), + testProxy("eu2", map[string]string{"geo": "eu"}, true, withLatency(10)), + ) + + got := do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{ + "selector": map[string]string{"geo": "eu"}, + }) + if got.status != http.StatusCreated { + t.Fatalf("status %d (%v), want 201", got.status, got.body) + } + if got.body["leaseID"] == "" || got.body["leaseID"] == nil { + t.Error("empty leaseID") + } + if got.body["ttlSeconds"].(float64) != 300 { + t.Errorf("ttlSeconds = %v, want the 300 default", got.body["ttlSeconds"]) + } + proxy := got.body["proxy"].(map[string]any) + if proxy["id"] != "default/eu2" { + t.Errorf("granted %v, want default/eu2 (lower latency at equal load)", proxy["id"]) + } + if proxy["activeLeases"].(float64) != 1 { + t.Errorf("activeLeases = %v, want 1 (this grant included)", proxy["activeLeases"]) + } + if _, err := time.Parse(time.RFC3339, got.body["expiresAt"].(string)); err != nil { + t.Errorf("expiresAt %v is not RFC3339: %v", got.body["expiresAt"], err) + } +} + +func TestAcquireLease_noMatchBody(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "", + testProxy("eu-tiny", map[string]string{"geo": "eu"}, true, withMaxLeases(1)), + testProxy("eu-sick", map[string]string{"geo": "eu"}, false), + ) + + body := map[string]any{"selector": map[string]string{"geo": "eu"}} + if got := do(t, ts, http.MethodPost, "/v1/leases", "", body); got.status != http.StatusCreated { + t.Fatalf("first acquire: status %d, want 201", got.status) + } + + got := do(t, ts, http.MethodPost, "/v1/leases", "", body) + if got.status != http.StatusConflict { + t.Fatalf("second acquire: status %d, want 409", got.status) + } + want := map[string]float64{"considered": 2, "atCapacity": 1, "inCooldown": 0, "unhealthy": 1} + for k, v := range want { + if got.body[k].(float64) != v { + t.Errorf("%s = %v, want %v (body %v)", k, got.body[k], v, got.body) + } + } + if got.body["error"] != "no_match" { + t.Errorf("error = %v, want no_match", got.body["error"]) + } +} + +func TestAcquireLease_badRequests(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "", testProxy("p1", nil, true)) + + tests := []struct { + name string + body any + wantCode string + }{ + {name: "ttl above the cap", body: map[string]any{"ttlSeconds": 999999}, wantCode: "invalid_ttl"}, + {name: "negative ttl", body: map[string]any{"ttlSeconds": -5}, wantCode: "invalid_ttl"}, + {name: "malformed json", body: "{not json", wantCode: "invalid_body"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := do(t, ts, http.MethodPost, "/v1/leases", "", tc.body) + if got.status != http.StatusBadRequest || got.body["error"] != tc.wantCode { + t.Errorf("= %d/%v, want 400/%s", got.status, got.body["error"], tc.wantCode) + } + }) + } +} + +func TestReleaseLease_alwaysNoContent(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "", testProxy("p1", nil, true)) + + got := do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{}) + if got.status != http.StatusCreated { + t.Fatalf("acquire: status %d, want 201", got.status) + } + id := got.body["leaseID"].(string) + + for _, path := range []string{"/v1/leases/" + id, "/v1/leases/" + id, "/v1/leases/never-existed"} { + if got := do(t, ts, http.MethodDelete, path, "", nil); got.status != http.StatusNoContent { + t.Errorf("DELETE %s: status %d, want 204", path, got.status) + } + } +} + +func TestReportLease(t *testing.T) { + t.Parallel() + ts, _ := newTestServer(t, "", testProxy("p1", map[string]string{"geo": "eu"}, true)) + + got := do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{ + "selector": map[string]string{"geo": "eu"}, "target": "example.com", + }) + if got.status != http.StatusCreated { + t.Fatalf("acquire: status %d, want 201", got.status) + } + id := got.body["leaseID"].(string) + reportPath := fmt.Sprintf("/v1/leases/%s/report", id) + + if got := do(t, ts, http.MethodPost, reportPath, "", map[string]any{"result": "rate_limited", "target": "example.com"}); got.status != http.StatusNoContent { + t.Fatalf("report: status %d, want 204", got.status) + } + + // The cooldown from the report now blocks same-target acquisition. + got = do(t, ts, http.MethodPost, "/v1/leases", "", map[string]any{ + "selector": map[string]string{"geo": "eu"}, "target": "example.com", + }) + if got.status != http.StatusConflict || got.body["inCooldown"].(float64) != 1 { + t.Errorf("post-report acquire = %d/%v, want 409 with inCooldown 1", got.status, got.body) + } + + t.Run("invalid result value", func(t *testing.T) { + got := do(t, ts, http.MethodPost, reportPath, "", map[string]any{"result": "throttled"}) + if got.status != http.StatusBadRequest || got.body["error"] != "invalid_result" { + t.Errorf("= %d/%v, want 400/invalid_result", got.status, got.body["error"]) + } + }) + t.Run("unknown lease", func(t *testing.T) { + got := do(t, ts, http.MethodPost, "/v1/leases/never-existed/report", "", map[string]any{"result": "ok"}) + if got.status != http.StatusNotFound || got.body["error"] != "unknown_lease" { + t.Errorf("= %d/%v, want 404/unknown_lease", got.status, got.body["error"]) + } + }) +} + +func TestStart_servesAndShutsDown(t *testing.T) { + t.Parallel() + s := runtime.NewScheme() + if err := crawlv1alpha1.AddToScheme(s); err != nil { + t.Fatalf("scheme: %v", err) + } + srv := &Server{ + Reader: fake.NewClientBuilder().WithScheme(s).Build(), + Store: lease.NewStore(time.Minute), + Addr: "127.0.0.1:0", + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- srv.Start(ctx) }() + + var addr string + deadline := time.After(5 * time.Second) + for addr == "" { + select { + case <-deadline: + t.Fatal("server never bound") + case <-time.After(5 * time.Millisecond): + addr = srv.BoundAddr() + } + } + + resp, err := http.Get("http://" + addr + "/healthz") + if err != nil { + t.Fatalf("healthz: %v", err) + } + _ = resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("healthz status %d, want 200", resp.StatusCode) + } + + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("Start returned %v, want nil after graceful shutdown", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Start did not stop on cancel") + } +}