diff --git a/docs/architecture.md b/docs/architecture.md index fb4d751..d7e0a0d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,10 +1,10 @@ # Architecture -> **Status:** the operator is built through Step 8 (GCP provider) of +> **Status:** the operator is built through Step 9 (orphan GC + metrics) 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 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. +> This document covers the event/reconcile flow, the HTTP-driven +> lease/discovery path, and the GC sweep; the components table and the +> Decisions section arrive with Step 10. ## Event flow: cluster events → reconciler functions @@ -240,3 +240,50 @@ 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) ``` + +### 8. Orphan GC (`internal/gc/`) — the crash-safety net + +Timer-driven, leader-elected (destructive ⇒ single writer). Exists for the +one gap the reconciler cannot close alone: a crash after a provider Create +but before the status write that records the instance. + +```text +Sweeper.Start(ctx) ── refuses to run when the cache is namespace- + │ restricted unless --gc-allow-namespaced is explicit + │ (an incomplete live set would "orphan" live VMs) + └─ every Interval (10m; first sweep a full interval after start): + sweep(ctx) + │ Reader.List(Proxies) → live UID set + │ List fails → skip the whole sweep (never guess) + │ a CR with deletionTimestamp still counts as LIVE — its + │ finalizer owns that deletion; GC racing it double-deletes + └ per provider: ListByTag + │ error → log, continue with the next provider + └ delete only when ALL hold: + has the proxy-operator-uid label (ownership proof) + older than MinAge (10m) (not mid-create) + UID matches no existing CR (truly orphaned) + each kill logged loudly with provider, providerID, UID +``` + +### 9. Metrics (`internal/metrics/`) + +Registered explicitly from `cmd/main.go` (no `init()`; tests use fresh +registries). Two kinds: + +- **Scrape-time collectors** — `proxy_operator_proxies{phase}` and + `proxy_operator_leases_active` read the cache / lease store at every + scrape; reconcile-incremented gauges inevitably drift and leak series. +- **Fed vectors** — `healthcheck_duration_seconds{proxy}` and + `healthcheck_failures_total{proxy}` observe EVERY probe (status writes + are transition-only; metrics carry the high-frequency signal), and the + health engine deletes a proxy's series when it prunes its state; + `lease_requests_total{outcome}` from the discovery handlers; + `provider_requests_total{provider,op,result}` from the + `provider.WithMetrics` decorator — the one place `Class()` is called + purely for observability. + +Each consuming package defines its own small recorder interface +(`health.ProbeMetrics`, `discovery.LeaseMetrics`, `provider.RequestRecorder`); +`metrics.Metrics` satisfies all of them structurally, so no package other +than `cmd/main.go` imports the metrics package. 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 042ee77..670ff97 100644 --- a/docs/plans-executions/2026-08-07-1747-proxy-operator.md +++ b/docs/plans-executions/2026-08-07-1747-proxy-operator.md @@ -13,7 +13,7 @@ Pairs with [docs/plans/2026-08-07-1747-proxy-operator.md](../plans/2026-08-07-17 - [x] Step 6 — Lease store (`internal/lease/`) - [x] Step 7 — Discovery API (`internal/discovery/`) - [x] Step 8 — GCP provider (`internal/provider/gcp/`) -- [ ] Step 9 — Orphan GC + metrics +- [x] Step 9 — Orphan GC + metrics - [ ] Step 10 — Wiring, config, docs - [ ] Step 11 — Tests - [ ] Verification (vet/test/kind e2e) + commit, push, open MR @@ -875,3 +875,67 @@ protobuf-construction idiom and matches every example in the SDK docs. Registry wiring (`"gcp": gcp.New`) happens at the composition root in Step 10, as designed in Step 2. `docs/architecture.md` §5 now shows both providers' call mappings. + +## Step 9 — Orphan GC + metrics + +Implemented `internal/gc/gc.go` (the `Sweeper` Runnable) and +`internal/metrics/metrics.go` (explicit-registration metric set), plus the +`provider.WithMetrics` decorator deferred from Step 2 into +`internal/provider/metrics.go`, and the observation hooks in the health +engine and discovery server. + +**GC**, per the plan: `NeedLeaderElection() = true` (destructive ⇒ single +writer), 10 min ticker with the first sweep one full interval after start, +per-provider `ListByTag` with log-and-continue on provider errors, and a +kill requires all three of: our UID label present, older than `MinAge` +(10 min), and the UID matching no existing CR — where a CR with a +`deletionTimestamp` still counts as live (its finalizer owns that +deletion; GC racing it would double-delete). Two safety behaviors worth +naming: a failed Proxy `List` skips the whole sweep (an unreadable live +set proves nothing orphaned), and the namespace-scope guard makes `Start` +refuse to run against a namespace-restricted cache unless +`--gc-allow-namespaced` is explicit, with the flag named in the error. +One deviation of record: the plan says kills log "at Warn", but logr has +no Warn level — kills log at Info with a `WARNING:` prefix carrying +provider, providerID, and UID, same convention as the discovery server's +empty-token warning. + +**Metrics**, per the plan: no `init()` — `Metrics.Register(reg, phases, +leases)` is called explicitly by the composition root (Step 10), which is +also what lets every test use a fresh registry (asserted by a test that +registers two sets on two registries). `proxy_operator_proxies{phase}` +and `proxy_operator_leases_active` are scrape-time collectors fed by +closures — a reconcile-incremented gauge drifts and leaks series on +delete; reading the source of truth at scrape time cannot. The +probe vectors observe **every** probe (status stays transition-only; +metrics carry the high-frequency signal), and the health engine calls +`ForgetProxy` when it prunes a state entry so per-proxy series don't leak. +`provider_requests_total{provider,op,result}` comes from the +`WithMetrics` decorator — the one place `Class()` is called purely for +observability, with results labelled ok / not_found / quota_exceeded / +permanent / transient. + +**Decoupling shape:** each consuming package defines its own small +recorder interface (`health.ProbeMetrics`, `discovery.LeaseMetrics`, +`provider.RequestRecorder`); `metrics.Metrics` satisfies all of them +structurally. Only `cmd/main.go` will import `internal/metrics`. + +Tests (`-race -count=2` clean): GC's true-orphan matrix in one sweep +(live kept, deleting-CR kept, young kept, unlabelled kept, orphan +deleted), broken-provider isolation, list-failure skips sweep, the +namespace guard both ways, and the Start loop sweeping then stopping; +metrics via `prometheus/testutil` — `GatherAndCompare` on the scrape-time +collectors, ForgetProxy dropping series, outcome/result label counts; +the decorator's five-way classification table with error passthrough +asserted. Coverage: gc 86.1%, metrics 95.0%, provider up to 97.1%; +health/discovery re-ran green with the hooks in place. + +```bash +go test -race -count=2 ./internal/gc/ ./internal/metrics/ ./internal/provider/ ./internal/health/ ./internal/discovery/ +make test # whole repo green +``` + +Worth noting: `prometheus/client_golang` was already in the module via +controller-runtime's metrics server, so no new dependency — `go mod tidy` +just promoted it to direct. `docs/architecture.md` gained §8 (GC sweep) +and §9 (metrics shape). diff --git a/go.mod b/go.mod index c691526..de4d5ef 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/go-logr/logr v1.4.3 github.com/onsi/ginkgo/v2 v2.27.4 github.com/onsi/gomega v1.39.0 + github.com/prometheus/client_golang v1.23.2 google.golang.org/api v0.292.0 google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af k8s.io/api v0.36.0 @@ -51,12 +52,12 @@ require ( github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/mailru/easyjson v0.7.7 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect - github.com/prometheus/client_golang v1.23.2 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.67.5 // indirect github.com/prometheus/procfs v0.19.2 // indirect diff --git a/internal/discovery/handlers.go b/internal/discovery/handlers.go index 8975d0b..0df67e2 100644 --- a/internal/discovery/handlers.go +++ b/internal/discovery/handlers.go @@ -172,6 +172,9 @@ func (s *Server) handleAcquireLease(w http.ResponseWriter, r *http.Request) { TTL: ttl, }) if errors.Is(err, lease.ErrNoMatch) { + if s.Metrics != nil { + s.Metrics.LeaseRequest("no_match") + } writeJSON(w, http.StatusConflict, map[string]any{ "error": "no_match", "message": "no healthy proxy with free capacity matched the selector", @@ -187,6 +190,9 @@ func (s *Server) handleAcquireLease(w http.ResponseWriter, r *http.Request) { return } + if s.Metrics != nil { + s.Metrics.LeaseRequest("granted") + } writeJSON(w, http.StatusCreated, leaseResponse{ LeaseID: granted.ID, Proxy: viewOf(byKey[granted.Proxy], s.Store.Counts()[granted.Proxy]), diff --git a/internal/discovery/server.go b/internal/discovery/server.go index 16445fe..5c95fa5 100644 --- a/internal/discovery/server.go +++ b/internal/discovery/server.go @@ -32,6 +32,13 @@ type LeaseStore interface { Counts() map[string]int } +// LeaseMetrics counts lease acquisitions by outcome. Implemented by +// internal/metrics; defined here so this package carries no metrics +// dependency. +type LeaseMetrics interface { + LeaseRequest(outcome string) +} + const ( defaultAddr = ":8090" defaultTTL = 5 * time.Minute @@ -55,6 +62,8 @@ type Server struct { Token string // MaxLeaseTTL caps requested lease TTLs (default 1h; --max-lease-ttl). MaxLeaseTTL time.Duration + // Metrics, when non-nil, counts lease requests by outcome. + Metrics LeaseMetrics log logr.Logger diff --git a/internal/gc/gc.go b/internal/gc/gc.go new file mode 100644 index 0000000..269f1dd --- /dev/null +++ b/internal/gc/gc.go @@ -0,0 +1,125 @@ +// Package gc implements orphan garbage collection: a periodic sweep that +// deletes provider instances tagged by this operator whose owning Proxy CR +// no longer exists — the safety net for crashes between a provider Create +// and the status write that records it. +package gc + +import ( + "context" + "errors" + "time" + + "sigs.k8s.io/controller-runtime/pkg/client" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// Sweeper is the manager Runnable running the sweep loop. +type Sweeper struct { + // Reader lists Proxies from the manager's cache to establish the live + // UID set. + Reader client.Reader + // Providers are the configured backends; each is swept independently. + Providers map[string]provider.Provider + + // Interval between sweeps (default 10m). The first sweep runs one full + // interval after start, not immediately — right after startup the + // cache is coldest and an in-flight create is most likely. + Interval time.Duration + // MinAge exempts young instances (default 10m): an instance mid-create + // may not have its status write landed yet; deleting it would race the + // reconciler. + MinAge time.Duration + + // NamespaceRestricted must be set when the manager cache is limited to + // one namespace. Then the live-UID set is incomplete, and a sweep + // would delete VMs owned by Proxies the cache cannot see — so Start + // refuses unless AllowNamespaced (--gc-allow-namespaced) is explicit. + NamespaceRestricted bool + AllowNamespaced bool + + now func() time.Time +} + +// NeedLeaderElection is true: the sweep is destructive and must have a +// single writer. +func (s *Sweeper) NeedLeaderElection() bool { return true } + +// Start runs the sweep loop until ctx ends. +func (s *Sweeper) Start(ctx context.Context) error { + if s.NamespaceRestricted && !s.AllowNamespaced { + return errors.New( + "orphan GC refuses to run against a namespace-restricted cache: proxies outside the namespace " + + "would count as orphans and their instances would be deleted; pass --gc-allow-namespaced to override") + } + if s.Interval == 0 { + s.Interval = 10 * time.Minute + } + if s.MinAge == 0 { + s.MinAge = 10 * time.Minute + } + if s.now == nil { + s.now = time.Now + } + + ticker := time.NewTicker(s.Interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + s.sweep(ctx) + } + } +} + +// sweep deletes tagged instances whose UID matches no existing Proxy CR. +// A CR with a deletionTimestamp still counts as live: its finalizer owns +// that deletion, and GC racing it would double-delete. A UID becomes +// orphan-eligible only once the object is fully gone. +func (s *Sweeper) sweep(ctx context.Context) { + log := logf.FromContext(ctx).WithName("orphan-gc") + + var list crawlv1alpha1.ProxyList + if err := s.Reader.List(ctx, &list); err != nil { + // Without the live set nothing can be proven orphaned; skip the + // whole sweep rather than guess. + log.Error(err, "listing proxies; skipping this sweep") + return + } + live := make(map[string]bool, len(list.Items)) + for i := range list.Items { + live[string(list.Items[i].UID)] = true + } + + for name, prov := range s.Providers { + instances, err := prov.ListByTag(ctx) + if err != nil { + // One broken provider must not abort the sweep for the rest. + log.Error(err, "listing instances; skipping this provider", "provider", name) + continue + } + for _, inst := range instances { + switch { + case inst.UID == "": + // Managed label without a UID label shouldn't exist for + // anything this operator created; without ownership proof, + // never delete. + continue + case live[inst.UID]: + continue + case s.now().Sub(inst.CreatedAt) < s.MinAge: + continue + } + log.Info("WARNING: deleting orphaned instance", + "provider", name, "providerID", inst.ID, "uid", inst.UID) + if err := prov.Delete(ctx, inst.ID); err != nil { + log.Error(err, "deleting orphaned instance", + "provider", name, "providerID", inst.ID, "uid", inst.UID) + } + } + } +} diff --git a/internal/gc/gc_test.go b/internal/gc/gc_test.go new file mode 100644 index 0000000..b899422 --- /dev/null +++ b/internal/gc/gc_test.go @@ -0,0 +1,214 @@ +package gc + +import ( + "context" + "errors" + "strings" + "sync" + "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" + "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/provider" +) + +// listProvider serves a canned instance list and records deletions. +type listProvider struct { + mu sync.Mutex + instances []provider.Instance + listErr error + deleted []string +} + +func (l *listProvider) Create(context.Context, provider.CreateRequest) (string, error) { + return "", errors.New("not used") +} + +func (l *listProvider) Get(context.Context, string) (*provider.Instance, error) { + return nil, provider.ErrNotFound +} + +func (l *listProvider) Delete(_ context.Context, id string) error { + l.mu.Lock() + defer l.mu.Unlock() + l.deleted = append(l.deleted, id) + return nil +} + +func (l *listProvider) ListByTag(context.Context) ([]provider.Instance, error) { + return l.instances, l.listErr +} + +func (l *listProvider) deletedIDs() []string { + l.mu.Lock() + defer l.mu.Unlock() + return append([]string(nil), l.deleted...) +} + +func proxyWithUID(name, uid string, mut ...func(*crawlv1alpha1.Proxy)) *crawlv1alpha1.Proxy { + p := &crawlv1alpha1.Proxy{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default", UID: types.UID(uid)}, + Spec: crawlv1alpha1.ProxySpec{ + Mode: crawlv1alpha1.ModeManaged, + Provider: "stub", + }, + } + for _, m := range mut { + m(p) + } + return p +} + +func newReader(t *testing.T, objs ...client.Object) client.Reader { + t.Helper() + s := runtime.NewScheme() + if err := crawlv1alpha1.AddToScheme(s); err != nil { + t.Fatalf("scheme: %v", err) + } + return fake.NewClientBuilder().WithScheme(s).WithObjects(objs...).Build() +} + +func oldInstance(id, uid string) provider.Instance { + return provider.Instance{ID: id, UID: uid, State: provider.StateRunning, + CreatedAt: time.Now().Add(-time.Hour)} +} + +func newSweeper(reader client.Reader, providers map[string]provider.Provider) *Sweeper { + return &Sweeper{ + Reader: reader, + Providers: providers, + Interval: 10 * time.Minute, + MinAge: 10 * time.Minute, + now: time.Now, + } +} + +func TestSweep_deletesOnlyTrueOrphans(t *testing.T) { + t.Parallel() + + deletingCR := proxyWithUID("deleting", "uid-deleting", func(p *crawlv1alpha1.Proxy) { + now := metav1.Now() + p.DeletionTimestamp = &now + p.Finalizers = []string{crawlv1alpha1.FinalizerName} + }) + prov := &listProvider{instances: []provider.Instance{ + oldInstance("inst-live", "uid-live"), + oldInstance("inst-orphan", "uid-orphan"), + oldInstance("inst-deleting", "uid-deleting"), + {ID: "inst-young", UID: "uid-young-orphan", State: provider.StateRunning, + CreatedAt: time.Now().Add(-time.Minute)}, + oldInstance("inst-unlabelled", ""), + }} + s := newSweeper( + newReader(t, proxyWithUID("live", "uid-live"), deletingCR), + map[string]provider.Provider{"stub": prov}, + ) + + s.sweep(context.Background()) + + got := prov.deletedIDs() + if len(got) != 1 || got[0] != "inst-orphan" { + t.Errorf("deleted %v, want exactly [inst-orphan]:\n"+ + "live CR's instance must stay; a deleting CR still owns its instance (finalizer, not GC);\n"+ + "young instances may be mid-create; unlabelled instances have no ownership proof", got) + } +} + +func TestSweep_providerErrorDoesNotAbortOthers(t *testing.T) { + t.Parallel() + broken := &listProvider{listErr: errors.New("cloud is down")} + working := &listProvider{instances: []provider.Instance{oldInstance("inst-orphan", "uid-orphan")}} + s := newSweeper(newReader(t), map[string]provider.Provider{ + "broken": broken, + "working": working, + }) + + s.sweep(context.Background()) + + if got := working.deletedIDs(); len(got) != 1 { + t.Errorf("working provider deletions = %v, want the orphan despite the broken provider", got) + } +} + +// errReader fails every List: without the live set nothing can be proven +// orphaned, so the sweep must delete nothing. +type errReader struct{ client.Reader } + +func (errReader) List(context.Context, client.ObjectList, ...client.ListOption) error { + return errors.New("cache broken") +} + +func TestSweep_listFailureSkipsSweep(t *testing.T) { + t.Parallel() + prov := &listProvider{instances: []provider.Instance{oldInstance("inst-orphan", "uid-orphan")}} + s := newSweeper(errReader{}, map[string]provider.Provider{"stub": prov}) + + s.sweep(context.Background()) + + if got := prov.deletedIDs(); len(got) != 0 { + t.Errorf("deleted %v with an unreadable live set, want nothing", got) + } +} + +func TestStart_namespaceGuard(t *testing.T) { + t.Parallel() + + s := newSweeper(newReader(t), nil) + s.NamespaceRestricted = true + err := s.Start(context.Background()) + if err == nil || !strings.Contains(err.Error(), "--gc-allow-namespaced") { + t.Errorf("Start with restricted cache = %v, want refusal naming the override flag", err) + } + + s2 := newSweeper(newReader(t), nil) + s2.NamespaceRestricted = true + s2.AllowNamespaced = true + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s2.Start(ctx) }() + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("Start with override = %v, want it to run until cancel", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Start did not stop on cancel") + } +} + +func TestStart_sweepsOnIntervalAndStops(t *testing.T) { + t.Parallel() + prov := &listProvider{instances: []provider.Instance{oldInstance("inst-orphan", "uid-orphan")}} + s := newSweeper(newReader(t), map[string]provider.Provider{"stub": prov}) + s.Interval = 5 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- s.Start(ctx) }() + + deadline := time.After(5 * time.Second) + for len(prov.deletedIDs()) == 0 { + select { + case <-deadline: + t.Fatal("no sweep ran") + case <-time.After(2 * time.Millisecond): + } + } + + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("Start = %v, want nil", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Start did not stop on cancel") + } +} diff --git a/internal/health/engine.go b/internal/health/engine.go index 4377e4e..e35257a 100644 --- a/internal/health/engine.go +++ b/internal/health/engine.go @@ -62,6 +62,14 @@ type probeJob struct { interval time.Duration } +// ProbeMetrics receives every probe result and the retirement of a +// proxy's series. Implemented by internal/metrics; defined here so this +// package carries no metrics dependency. +type ProbeMetrics interface { + ObserveProbe(proxy string, latency time.Duration, success bool) + ForgetProxy(proxy string) +} + // Engine runs the probe scheduler and worker pool as a manager Runnable. It // never writes Proxy status itself — keeping the reconciler the single // status writer — and instead emits a GenericEvent per status-affecting @@ -87,6 +95,10 @@ type Engine struct { // ProbeTLSConfig overrides TLS verification for https probe URLs; nil // means system roots. Needed for private CAs (and tests). ProbeTLSConfig *tls.Config + // Metrics, when non-nil, is fed on every probe — the status writes are + // transition-only by design, so metrics are where high-frequency + // signal (true probe recency, every latency sample) lives. + Metrics ProbeMetrics probeFn func(context.Context, *url.URL, crawlv1alpha1.HealthCheckSpec, *tls.Config) probeResult @@ -220,6 +232,11 @@ func (e *Engine) tick(ctx context.Context, now time.Time, jobs chan<- probeJob) for key := range e.states { if _, ok := probeable[key]; !ok { delete(e.states, key) + if e.Metrics != nil { + // Retire the per-proxy series with the state, or series + // for deleted proxies leak forever. + e.Metrics.ForgetProxy(key.String()) + } } } } @@ -252,6 +269,10 @@ func newState(p *crawlv1alpha1.Proxy, now time.Time, interval time.Duration) *st // threshold-crossing flip, or a material latency change (beyond // max(LatencyFloor, 50% of reported) and rate-limited by MinReportInterval). func (e *Engine) record(job probeJob, res probeResult, now time.Time) { + if e.Metrics != nil { + e.Metrics.ObserveProbe(job.key.String(), res.latency, res.ok) + } + e.mu.Lock() defer e.mu.Unlock() diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 0000000..b827395 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,119 @@ +// Package metrics defines the operator's Prometheus metrics. Nothing here +// registers itself — no init(), per house rules — the composition root +// calls Register explicitly, which also lets every test use a fresh +// registry. +package metrics + +import ( + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +// Metrics holds the vector metrics the operator's components feed. The +// consuming packages (health, discovery, provider) each define their own +// small recorder interface; *Metrics satisfies all of them structurally, +// so none of them import this package's prometheus surface. +type Metrics struct { + healthcheckDuration *prometheus.HistogramVec + healthcheckFailures *prometheus.CounterVec + leaseRequests *prometheus.CounterVec + providerRequests *prometheus.CounterVec +} + +// New builds the metric set, unregistered. +func New() *Metrics { + return &Metrics{ + healthcheckDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "proxy_operator_healthcheck_duration_seconds", + Help: "Duration of through-the-proxy health probes.", + Buckets: prometheus.DefBuckets, + }, []string{"proxy"}), + healthcheckFailures: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "proxy_operator_healthcheck_failures_total", + Help: "Failed health probes.", + }, []string{"proxy"}), + leaseRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "proxy_operator_lease_requests_total", + Help: "Lease acquisition requests by outcome.", + }, []string{"outcome"}), + providerRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "proxy_operator_provider_requests_total", + Help: "Provider API calls by operation and classified result.", + }, []string{"provider", "op", "result"}), + } +} + +// Register registers the vectors plus the two scrape-time collectors. +// proxyPhases and activeLeases are read at every scrape: gauges derived +// from reconcile-time increments inevitably drift and leak series on +// delete; reading the source of truth cannot. +func (m *Metrics) Register(reg prometheus.Registerer, proxyPhases func() map[string]int, activeLeases func() int) error { + collectors := []prometheus.Collector{ + m.healthcheckDuration, + m.healthcheckFailures, + m.leaseRequests, + m.providerRequests, + &constCollector{ + desc: prometheus.NewDesc("proxy_operator_proxies", + "Proxy objects by phase.", []string{"phase"}, nil), + read: proxyPhases, + }, + &constCollector{ + desc: prometheus.NewDesc("proxy_operator_leases_active", + "Currently active leases.", nil, nil), + read: func() map[string]int { return map[string]int{"": activeLeases()} }, + }, + } + for _, c := range collectors { + if err := reg.Register(c); err != nil { + return err + } + } + return nil +} + +// ObserveProbe records one health probe. Called on every probe — metrics +// are the home for high-frequency signal that must never touch status. +func (m *Metrics) ObserveProbe(proxy string, latency time.Duration, success bool) { + m.healthcheckDuration.WithLabelValues(proxy).Observe(latency.Seconds()) + if !success { + m.healthcheckFailures.WithLabelValues(proxy).Inc() + } +} + +// ForgetProxy drops the per-proxy series when the health engine prunes its +// state — without this, series for deleted proxies leak forever. +func (m *Metrics) ForgetProxy(proxy string) { + m.healthcheckDuration.DeleteLabelValues(proxy) + m.healthcheckFailures.DeleteLabelValues(proxy) +} + +// LeaseRequest records a lease acquisition outcome ("granted"|"no_match"). +func (m *Metrics) LeaseRequest(outcome string) { + m.leaseRequests.WithLabelValues(outcome).Inc() +} + +// ProviderRequest records one provider API call with its classified result. +func (m *Metrics) ProviderRequest(provider, op, result string) { + m.providerRequests.WithLabelValues(provider, op, result).Inc() +} + +// constCollector reads a label→value map at scrape time and emits one +// gauge sample per entry. An empty-string label key means "no labels". +type constCollector struct { + desc *prometheus.Desc + read func() map[string]int +} + +func (c *constCollector) Describe(ch chan<- *prometheus.Desc) { ch <- c.desc } + +func (c *constCollector) Collect(ch chan<- prometheus.Metric) { + for label, value := range c.read() { + if label == "" { + ch <- prometheus.MustNewConstMetric(c.desc, prometheus.GaugeValue, float64(value)) + continue + } + ch <- prometheus.MustNewConstMetric(c.desc, prometheus.GaugeValue, float64(value), label) + } +} diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go new file mode 100644 index 0000000..d6e87a7 --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -0,0 +1,113 @@ +package metrics + +import ( + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// register wires a fresh registry — the reason Register exists instead of +// init()-time self-registration. +func register(t *testing.T, m *Metrics, phases map[string]int, active int) *prometheus.Registry { + t.Helper() + reg := prometheus.NewRegistry() + err := m.Register(reg, + func() map[string]int { return phases }, + func() int { return active }, + ) + if err != nil { + t.Fatalf("Register: %v", err) + } + return reg +} + +func TestRegister_scrapeTimeCollectors(t *testing.T) { + t.Parallel() + m := New() + reg := register(t, m, map[string]int{"Ready": 3, "Provisioning": 1}, 7) + + expected := ` +# HELP proxy_operator_leases_active Currently active leases. +# TYPE proxy_operator_leases_active gauge +proxy_operator_leases_active 7 +# HELP proxy_operator_proxies Proxy objects by phase. +# TYPE proxy_operator_proxies gauge +proxy_operator_proxies{phase="Provisioning"} 1 +proxy_operator_proxies{phase="Ready"} 3 +` + if err := testutil.GatherAndCompare(reg, strings.NewReader(expected), + "proxy_operator_proxies", "proxy_operator_leases_active"); err != nil { + t.Error(err) + } +} + +func TestObserveProbe_andForget(t *testing.T) { + t.Parallel() + m := New() + reg := register(t, m, nil, 0) + + m.ObserveProbe("default/p1", 30*time.Millisecond, true) + m.ObserveProbe("default/p1", 40*time.Millisecond, false) + m.ObserveProbe("default/p2", 10*time.Millisecond, true) + + if got := testutil.CollectAndCount(m.healthcheckDuration); got != 2 { + t.Errorf("duration series = %d, want 2 (one per proxy)", got) + } + if got := testutil.ToFloat64(m.healthcheckFailures.WithLabelValues("default/p1")); got != 1 { + t.Errorf("p1 failures = %v, want 1 (only the failed probe)", got) + } + + m.ForgetProxy("default/p1") + if got := testutil.CollectAndCount(m.healthcheckDuration); got != 1 { + t.Errorf("duration series after ForgetProxy = %d, want 1 — series must not leak", got) + } + if got := testutil.CollectAndCount(m.healthcheckFailures); got != 0 { + t.Errorf("failure series after ForgetProxy = %d, want 0", got) + } + _ = reg +} + +func TestLeaseRequest(t *testing.T) { + t.Parallel() + m := New() + register(t, m, nil, 0) + + m.LeaseRequest("granted") + m.LeaseRequest("granted") + m.LeaseRequest("no_match") + + if got := testutil.ToFloat64(m.leaseRequests.WithLabelValues("granted")); got != 2 { + t.Errorf("granted = %v, want 2", got) + } + if got := testutil.ToFloat64(m.leaseRequests.WithLabelValues("no_match")); got != 1 { + t.Errorf("no_match = %v, want 1", got) + } +} + +func TestProviderRequest(t *testing.T) { + t.Parallel() + m := New() + register(t, m, nil, 0) + + m.ProviderRequest("gcp-eu", "create", "ok") + m.ProviderRequest("gcp-eu", "create", "quota_exceeded") + + if got := testutil.ToFloat64(m.providerRequests.WithLabelValues("gcp-eu", "create", "ok")); got != 1 { + t.Errorf("ok = %v, want 1", got) + } + if got := testutil.ToFloat64(m.providerRequests.WithLabelValues("gcp-eu", "create", "quota_exceeded")); got != 1 { + t.Errorf("quota_exceeded = %v, want 1", got) + } +} + +func TestRegister_freshRegistryPerTest(t *testing.T) { + t.Parallel() + // Registering the same metric set on two registries must both succeed — + // the property init()-style global registration would break. + m1, m2 := New(), New() + register(t, m1, nil, 0) + register(t, m2, nil, 0) +} diff --git a/internal/provider/metrics.go b/internal/provider/metrics.go new file mode 100644 index 0000000..96b24fa --- /dev/null +++ b/internal/provider/metrics.go @@ -0,0 +1,66 @@ +package provider + +import "context" + +// RequestRecorder receives one record per provider API call. Implemented +// by internal/metrics; defined here so this package needs no metrics +// dependency. +type RequestRecorder interface { + ProviderRequest(provider, op, result string) +} + +// WithMetrics wraps a Provider so every call is recorded with its +// classified result — zero-cost instrumentation for the next five +// providers, and the one place Class is called purely for observability. +func WithMetrics(name string, p Provider, rec RequestRecorder) Provider { + return &instrumented{name: name, inner: p, rec: rec} +} + +type instrumented struct { + name string + inner Provider + rec RequestRecorder +} + +func (i *instrumented) Create(ctx context.Context, req CreateRequest) (string, error) { + id, err := i.inner.Create(ctx, req) + i.record("create", err) + return id, err +} + +func (i *instrumented) Get(ctx context.Context, providerID string) (*Instance, error) { + inst, err := i.inner.Get(ctx, providerID) + i.record("get", err) + return inst, err +} + +func (i *instrumented) Delete(ctx context.Context, providerID string) error { + err := i.inner.Delete(ctx, providerID) + i.record("delete", err) + return err +} + +func (i *instrumented) ListByTag(ctx context.Context) ([]Instance, error) { + instances, err := i.inner.ListByTag(ctx) + i.record("list", err) + return instances, err +} + +func (i *instrumented) record(op string, err error) { + i.rec.ProviderRequest(i.name, op, resultLabel(err)) +} + +func resultLabel(err error) string { + switch Class(err) { + case nil: + return "ok" + case ErrNotFound: + return "not_found" + case ErrQuotaExceeded: + return "quota_exceeded" + case ErrPermanent: + return "permanent" + default: + return "transient" + } +} diff --git a/internal/provider/metrics_test.go b/internal/provider/metrics_test.go new file mode 100644 index 0000000..9f30eec --- /dev/null +++ b/internal/provider/metrics_test.go @@ -0,0 +1,100 @@ +package provider + +import ( + "context" + "errors" + "testing" +) + +type recordedCall struct{ provider, op, result string } + +type fakeRecorder struct{ calls []recordedCall } + +func (f *fakeRecorder) ProviderRequest(provider, op, result string) { + f.calls = append(f.calls, recordedCall{provider, op, result}) +} + +// staticProvider returns canned values; only the classification of its +// errors matters here. +type staticProvider struct { + createErr, deleteErr, getErr, listErr error +} + +func (s *staticProvider) Create(context.Context, CreateRequest) (string, error) { + return "id-1", s.createErr +} +func (s *staticProvider) Get(context.Context, string) (*Instance, error) { + return &Instance{ID: "id-1"}, s.getErr +} +func (s *staticProvider) Delete(context.Context, string) error { return s.deleteErr } +func (s *staticProvider) ListByTag(context.Context) ([]Instance, error) { return nil, s.listErr } + +func TestWithMetrics_recordsClassifiedResults(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + inner *staticProvider + call func(p Provider) error + wantOp string + wantResult string + }{ + { + name: "successful create is ok", + inner: &staticProvider{}, + call: func(p Provider) error { _, err := p.Create(context.Background(), CreateRequest{}); return err }, + wantOp: "create", + wantResult: "ok", + }, + { + name: "get NotFound", + inner: &staticProvider{getErr: Wrap(ErrNotFound, "get", "x", "id-1", nil)}, + call: func(p Provider) error { _, err := p.Get(context.Background(), "id-1"); return err }, + wantOp: "get", + wantResult: "not_found", + }, + { + name: "create quota", + inner: &staticProvider{createErr: Wrap(ErrQuotaExceeded, "create", "x", "", nil)}, + call: func(p Provider) error { _, err := p.Create(context.Background(), CreateRequest{}); return err }, + wantOp: "create", + wantResult: "quota_exceeded", + }, + { + name: "delete permanent", + inner: &staticProvider{deleteErr: Wrap(ErrPermanent, "delete", "x", "id-1", nil)}, + call: func(p Provider) error { return p.Delete(context.Background(), "id-1") }, + wantOp: "delete", + wantResult: "permanent", + }, + { + name: "unclassified list error is transient", + inner: &staticProvider{listErr: errors.New("connection reset")}, + call: func(p Provider) error { _, err := p.ListByTag(context.Background()); return err }, + wantOp: "list", + wantResult: "transient", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + rec := &fakeRecorder{} + p := WithMetrics("gcp-eu", tc.inner, rec) + + callErr := tc.call(p) + + if len(rec.calls) != 1 { + t.Fatalf("recorded %d calls, want 1", len(rec.calls)) + } + want := recordedCall{provider: "gcp-eu", op: tc.wantOp, result: tc.wantResult} + if rec.calls[0] != want { + t.Errorf("recorded %+v, want %+v", rec.calls[0], want) + } + // The decorator must be transparent: errors pass through. + if (tc.wantResult == "ok") != (callErr == nil) { + t.Errorf("error passthrough broken: result %s but err %v", tc.wantResult, callErr) + } + }) + } +}