Add orphan GC sweeper and Prometheus metrics with explicit registration

Co-Authored-By: Claude <noreply@anthropic.com>

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
+> This document covers the event/reconcile flow, the HTTP-driven
-> HTTP-driven lease/discovery path; the components table and the Decisions
+> lease/discovery path, and the GC sweep; the components table and the
-> section arrive with Step 10, and the orphan-GC flow lands with Step 9.
+> 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.

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).

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

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]),

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
 

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)
+			}
+		}
+	}
+}

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")
+	}
+}

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()
 

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)
+	}
+}

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)
+}

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"
+	}
+}

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)
+			}
+		})
+	}
+}