Add the health engine: through-proxy probes, thresholds, channel-fed transitions

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

internal/health/engine.go

@@ -0,0 +1,340 @@
+package health
+
+import (
+	"context"
+	"crypto/tls"
+	"math/rand/v2"
+	"net"
+	"net/url"
+	"strconv"
+	"sync"
+	"time"
+
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+
+	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
+)
+
+// Snapshot is the engine's current verdict for one proxy, read by the
+// reconciler when it represents health in the Proxy's status.
+type Snapshot struct {
+	Healthy bool
+	// Latency is the wall time of the most recent successful probe.
+	Latency time.Duration
+	// LastProbe is when the most recent probe (of either outcome) finished.
+	LastProbe time.Time
+	// LastError is the most recent probe failure; empty after a success.
+	LastError string
+	// ConsecutiveFailures is the current failure streak.
+	ConsecutiveFailures int32
+}
+
+// state is the engine's threshold bookkeeping for one proxy. The reported*
+// fields track what has been delivered over Events; they only advance when a
+// send succeeds, so a dropped event is retried after the next probe.
+type state struct {
+	uid      types.UID
+	inFlight bool
+	nextDue  time.Time
+
+	healthy    *bool // nil until a first verdict exists
+	consecOK   int32
+	consecFail int32
+	latency    time.Duration
+	lastProbe  time.Time
+	lastErr    string
+
+	reportedHealthy *bool
+	reportedLatency time.Duration
+	lastReport      time.Time
+}
+
+type probeJob struct {
+	key      types.NamespacedName
+	uid      types.UID
+	proxyURL *url.URL
+	hc       crawlv1alpha1.HealthCheckSpec
+	interval time.Duration
+}
+
+// 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
+// transition, which the reconciler consumes via source.Channel.
+type Engine struct {
+	// Reader lists Proxies from the manager's cache each tick.
+	Reader client.Reader
+	// Events carries one enqueue-request per status-affecting transition.
+	// Sends are non-blocking: a wedged reconciler must never stall probing.
+	Events chan event.GenericEvent
+
+	// Workers is the probe worker pool size (default 8).
+	Workers int
+	// Tick is the scheduler interval (default 1s). At tens of proxies a
+	// per-second list scan is free; a timer wheel would be unjustified.
+	Tick time.Duration
+	// MinReportInterval rate-limits latency-only status reports (default 60s).
+	MinReportInterval time.Duration
+	// LatencyFloor is the absolute change below which a latency move is
+	// never status-affecting (default 20ms), so a proxy jittering around a
+	// small latency doesn't write status forever.
+	LatencyFloor time.Duration
+	// ProbeTLSConfig overrides TLS verification for https probe URLs; nil
+	// means system roots. Needed for private CAs (and tests).
+	ProbeTLSConfig *tls.Config
+
+	probeFn func(context.Context, *url.URL, crawlv1alpha1.HealthCheckSpec, *tls.Config) probeResult
+
+	mu     sync.Mutex
+	states map[types.NamespacedName]*state
+}
+
+// NewEngine returns an Engine with a buffered Events channel, ready to be
+// handed to both mgr.Add and the reconciler (Health + HealthEvents fields).
+func NewEngine(reader client.Reader) *Engine {
+	e := &Engine{Reader: reader}
+	e.applyDefaults()
+	return e
+}
+
+func (e *Engine) applyDefaults() {
+	if e.Events == nil {
+		e.Events = make(chan event.GenericEvent, 64)
+	}
+	if e.Workers == 0 {
+		e.Workers = 8
+	}
+	if e.Tick == 0 {
+		e.Tick = time.Second
+	}
+	if e.MinReportInterval == 0 {
+		e.MinReportInterval = time.Minute
+	}
+	if e.LatencyFloor == 0 {
+		e.LatencyFloor = 20 * time.Millisecond
+	}
+	if e.probeFn == nil {
+		e.probeFn = probe
+	}
+	if e.states == nil {
+		e.states = map[types.NamespacedName]*state{}
+	}
+}
+
+// NeedLeaderElection makes the engine run only on the leader: probing from
+// every replica would multiply load on the proxies, and only the leader's
+// reconciler can represent the results anyway.
+func (e *Engine) NeedLeaderElection() bool { return true }
+
+// Start runs the scheduler tick loop and the worker pool until ctx ends.
+func (e *Engine) Start(ctx context.Context) error {
+	e.applyDefaults()
+	jobs := make(chan probeJob)
+	var wg sync.WaitGroup
+	for range e.Workers {
+		wg.Go(func() {
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case job := <-jobs:
+					res := e.probeFn(ctx, job.proxyURL, job.hc, e.ProbeTLSConfig)
+					e.record(job, res, time.Now())
+				}
+			}
+		})
+	}
+
+	ticker := time.NewTicker(e.Tick)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			wg.Wait()
+			return nil
+		case now := <-ticker.C:
+			e.tick(ctx, now, jobs)
+		}
+	}
+}
+
+// tick lists proxies from the cache, refreshes the state map (create, seed,
+// prune, UID-mismatch reset), and hands due proxies to the worker pool.
+func (e *Engine) tick(ctx context.Context, now time.Time, jobs chan<- probeJob) {
+	var list crawlv1alpha1.ProxyList
+	if err := e.Reader.List(ctx, &list); err != nil {
+		logf.FromContext(ctx).Error(err, "health engine: listing proxies")
+		return
+	}
+
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	probeable := make(map[types.NamespacedName]struct{}, len(list.Items))
+	for i := range list.Items {
+		p := &list.Items[i]
+		host := p.EffectiveHost()
+		if host == "" || !p.DeletionTimestamp.IsZero() {
+			// Not probeable (provisioning, being replaced, or deleting).
+			// Its state gets pruned below, so a replacement instance starts
+			// with fresh counters.
+			continue
+		}
+		key := client.ObjectKeyFromObject(p)
+		probeable[key] = struct{}{}
+
+		hc := p.HealthCheckOrDefault()
+		interval := time.Duration(hc.IntervalSeconds) * time.Second
+
+		st := e.states[key]
+		if st == nil || st.uid != p.UID {
+			// New proxy, or a delete+recreate under the same name — never
+			// inherit the old object's counters.
+			st = newState(p, now, interval)
+			e.states[key] = st
+		}
+		if st.inFlight || now.Before(st.nextDue) {
+			continue
+		}
+		job := probeJob{
+			key:      key,
+			uid:      p.UID,
+			proxyURL: &url.URL{Scheme: "http", Host: net.JoinHostPort(host, strconv.Itoa(int(p.EffectivePort())))},
+			hc:       hc,
+			interval: interval,
+		}
+		select {
+		case jobs <- job:
+			st.inFlight = true
+		default:
+			// Worker pool saturated; the proxy stays due and is retried on
+			// the next tick.
+		}
+	}
+
+	for key := range e.states {
+		if _, ok := probeable[key]; !ok {
+			delete(e.states, key)
+		}
+	}
+}
+
+// newState seeds bookkeeping for a proxy the engine hasn't tracked yet. If
+// the CR already carries a Healthy verdict (leader handover, operator
+// restart), the verdict is kept — so a healthy proxy doesn't flap to
+// unknown — counters stay at zero so a real transition still needs a full
+// threshold run, and the first probe is jittered across the interval so a
+// restart doesn't fire the whole fleet's probes at once. A proxy with no
+// prior verdict is probed immediately.
+func newState(p *crawlv1alpha1.Proxy, now time.Time, interval time.Duration) *state {
+	st := &state{uid: p.UID, nextDue: now}
+	cond := apimeta.FindStatusCondition(p.Status.Conditions, crawlv1alpha1.ConditionHealthy)
+	if cond == nil || cond.Status == metav1.ConditionUnknown {
+		return st
+	}
+	healthy := cond.Status == metav1.ConditionTrue
+	reported := healthy
+	st.healthy = &healthy
+	st.reportedHealthy = &reported
+	st.latency = time.Duration(p.Status.LatencyMillis) * time.Millisecond
+	st.reportedLatency = st.latency
+	st.nextDue = now.Add(rand.N(interval))
+	return st
+}
+
+// record folds one probe result into the proxy's threshold state and emits
+// an event when the result is status-affecting: a first-ever verdict, a
+// 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) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	st := e.states[job.key]
+	if st == nil || st.uid != job.uid {
+		return // pruned or replaced while the probe was in flight
+	}
+	st.inFlight = false
+	st.nextDue = now.Add(job.interval)
+	st.lastProbe = now
+
+	if res.ok {
+		st.consecOK++
+		st.consecFail = 0
+		st.latency = res.latency
+		st.lastErr = ""
+	} else {
+		st.consecFail++
+		st.consecOK = 0
+		st.lastErr = res.err.Error()
+	}
+
+	switch {
+	case st.healthy == nil:
+		healthy := res.ok
+		st.healthy = &healthy
+	case *st.healthy && st.consecFail >= job.hc.FailureThreshold:
+		healthy := false
+		st.healthy = &healthy
+	case !*st.healthy && st.consecOK >= job.hc.SuccessThreshold:
+		healthy := true
+		st.healthy = &healthy
+	}
+
+	var emit bool
+	switch {
+	case st.reportedHealthy == nil:
+		emit = true
+	case *st.reportedHealthy != *st.healthy:
+		emit = true
+	case res.ok && *st.healthy:
+		// Latency-only updates matter only for a healthy verdict; a success
+		// streak still below successThreshold must stay silent.
+		delta := st.latency - st.reportedLatency
+		if delta < 0 {
+			delta = -delta
+		}
+		emit = delta > max(e.LatencyFloor, st.reportedLatency/2) &&
+			now.Sub(st.lastReport) > e.MinReportInterval
+	}
+	if !emit {
+		return
+	}
+
+	evt := event.GenericEvent{Object: &crawlv1alpha1.Proxy{
+		ObjectMeta: metav1.ObjectMeta{Namespace: job.key.Namespace, Name: job.key.Name},
+	}}
+	select {
+	case e.Events <- evt:
+		reported := *st.healthy
+		st.reportedHealthy = &reported
+		st.reportedLatency = st.latency
+		st.lastReport = now
+	default:
+		// Channel full (reconciler wedged): drop, and deliberately do not
+		// advance the reported markers, so the next probe retries the emit.
+	}
+}
+
+// Snapshot returns the engine's current verdict for key; ok is false while
+// no verdict exists (never probed, or state was reset).
+func (e *Engine) Snapshot(key types.NamespacedName) (Snapshot, bool) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	st := e.states[key]
+	if st == nil || st.healthy == nil {
+		return Snapshot{}, false
+	}
+	return Snapshot{
+		Healthy:             *st.healthy,
+		Latency:             st.latency,
+		LastProbe:           st.lastProbe,
+		LastError:           st.lastErr,
+		ConsecutiveFailures: st.consecFail,
+	}, true
+}

internal/health/engine_test.go

@@ -0,0 +1,372 @@
+package health
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"net/url"
+	"strconv"
+	"strings"
+	"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"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+
+	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
+)
+
+var testKey = types.NamespacedName{Namespace: "default", Name: "p1"}
+
+func testEngine() *Engine {
+	e := &Engine{Events: make(chan event.GenericEvent, 8)}
+	e.applyDefaults()
+	return e
+}
+
+func testJob(failureThreshold, successThreshold int32) probeJob {
+	return probeJob{
+		key: testKey,
+		uid: "uid-1",
+		hc: crawlv1alpha1.HealthCheckSpec{
+			FailureThreshold: failureThreshold,
+			SuccessThreshold: successThreshold,
+		},
+		interval: 30 * time.Second,
+	}
+}
+
+func drainOneEvent(t *testing.T, e *Engine) event.GenericEvent {
+	t.Helper()
+	select {
+	case evt := <-e.Events:
+		return evt
+	default:
+		t.Fatal("expected an event, channel is empty")
+		return event.GenericEvent{}
+	}
+}
+
+func assertNoEvent(t *testing.T, e *Engine) {
+	t.Helper()
+	select {
+	case <-e.Events:
+		t.Fatal("unexpected event emitted")
+	default:
+	}
+}
+
+func boolPtr(b bool) *bool { return &b }
+
+func TestRecord_firstResultEmits(t *testing.T) {
+	t.Parallel()
+	e := testEngine()
+	e.states[testKey] = &state{uid: "uid-1"}
+
+	e.record(testJob(3, 1), probeResult{ok: true, latency: 30 * time.Millisecond}, time.Now())
+
+	evt := drainOneEvent(t, e)
+	if got := evt.Object.GetName(); got != "p1" {
+		t.Errorf("event object name = %q, want p1", got)
+	}
+	snap, ok := e.Snapshot(testKey)
+	if !ok || !snap.Healthy {
+		t.Errorf("Snapshot = %+v, %v; want healthy verdict", snap, ok)
+	}
+	if snap.Latency != 30*time.Millisecond {
+		t.Errorf("latency = %v, want 30ms", snap.Latency)
+	}
+}
+
+func TestRecord_failureThresholdFlips(t *testing.T) {
+	t.Parallel()
+	e := testEngine()
+	e.states[testKey] = &state{uid: "uid-1", healthy: boolPtr(true), reportedHealthy: boolPtr(true)}
+	job := testJob(3, 1)
+	probeErr := probeResult{err: errors.New("connect refused")}
+
+	e.record(job, probeErr, time.Now())
+	e.record(job, probeErr, time.Now())
+	assertNoEvent(t, e)
+	if snap, _ := e.Snapshot(testKey); !snap.Healthy {
+		t.Fatal("flipped unhealthy before failureThreshold was reached")
+	}
+
+	e.record(job, probeErr, time.Now())
+	drainOneEvent(t, e)
+	snap, _ := e.Snapshot(testKey)
+	if snap.Healthy {
+		t.Error("still healthy after failureThreshold consecutive failures")
+	}
+	if snap.ConsecutiveFailures != 3 {
+		t.Errorf("ConsecutiveFailures = %d, want 3", snap.ConsecutiveFailures)
+	}
+	if !strings.Contains(snap.LastError, "connect refused") {
+		t.Errorf("LastError = %q, want the probe error", snap.LastError)
+	}
+}
+
+func TestRecord_successThresholdFlips(t *testing.T) {
+	t.Parallel()
+	e := testEngine()
+	e.states[testKey] = &state{uid: "uid-1", healthy: boolPtr(false), reportedHealthy: boolPtr(false)}
+	job := testJob(3, 2)
+	success := probeResult{ok: true, latency: 25 * time.Millisecond}
+
+	e.record(job, success, time.Now())
+	assertNoEvent(t, e)
+
+	e.record(job, success, time.Now())
+	drainOneEvent(t, e)
+	if snap, _ := e.Snapshot(testKey); !snap.Healthy {
+		t.Error("not healthy after successThreshold consecutive successes")
+	}
+}
+
+func TestRecord_latencySuppression(t *testing.T) {
+	t.Parallel()
+	now := time.Now()
+
+	tests := []struct {
+		name            string
+		reportedLatency time.Duration
+		lastReport      time.Time
+		newLatency      time.Duration
+		wantEmit        bool
+	}{
+		{
+			name:            "small change under the relative floor is suppressed",
+			reportedLatency: 100 * time.Millisecond,
+			lastReport:      now.Add(-2 * time.Minute),
+			newLatency:      110 * time.Millisecond,
+			wantEmit:        false,
+		},
+		{
+			name:            "small absolute jitter at low latency is suppressed",
+			reportedLatency: 5 * time.Millisecond,
+			lastReport:      now.Add(-2 * time.Minute),
+			newLatency:      20 * time.Millisecond, // >50% but under the 20ms floor
+			wantEmit:        false,
+		},
+		{
+			name:            "material change after the rate window emits",
+			reportedLatency: 100 * time.Millisecond,
+			lastReport:      now.Add(-2 * time.Minute),
+			newLatency:      200 * time.Millisecond,
+			wantEmit:        true,
+		},
+		{
+			name:            "material change inside the rate window is suppressed",
+			reportedLatency: 100 * time.Millisecond,
+			lastReport:      now.Add(-10 * time.Second),
+			newLatency:      400 * time.Millisecond,
+			wantEmit:        false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			e := testEngine()
+			e.states[testKey] = &state{
+				uid:             "uid-1",
+				healthy:         boolPtr(true),
+				reportedHealthy: boolPtr(true),
+				reportedLatency: tc.reportedLatency,
+				lastReport:      tc.lastReport,
+			}
+			e.record(testJob(3, 1), probeResult{ok: true, latency: tc.newLatency}, now)
+			if tc.wantEmit {
+				drainOneEvent(t, e)
+			} else {
+				assertNoEvent(t, e)
+			}
+		})
+	}
+}
+
+func TestRecord_droppedEventIsRetried(t *testing.T) {
+	t.Parallel()
+	e := testEngine()
+	e.Events = make(chan event.GenericEvent) // unbuffered, nobody reading
+	e.states[testKey] = &state{uid: "uid-1"}
+	success := probeResult{ok: true, latency: 30 * time.Millisecond}
+
+	e.record(testJob(3, 1), success, time.Now())
+	e.mu.Lock()
+	reported := e.states[testKey].reportedHealthy
+	e.mu.Unlock()
+	if reported != nil {
+		t.Fatal("reported marker advanced although the event was dropped")
+	}
+
+	// Channel drains (reconciler recovers): the next probe re-emits.
+	e.Events = make(chan event.GenericEvent, 1)
+	e.record(testJob(3, 1), success, time.Now())
+	drainOneEvent(t, e)
+}
+
+func TestRecord_staleJobIsIgnored(t *testing.T) {
+	t.Parallel()
+	e := testEngine()
+	e.states[testKey] = &state{uid: "uid-NEW"}
+
+	job := testJob(3, 1)
+	job.uid = "uid-OLD"
+	e.record(job, probeResult{ok: true, latency: time.Millisecond}, time.Now())
+
+	assertNoEvent(t, e)
+	if _, ok := e.Snapshot(testKey); ok {
+		t.Error("stale probe produced a verdict for the new object")
+	}
+}
+
+func externalProxy(name, host string, 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: host, Port: 3128},
+		},
+	}
+	for _, m := range mut {
+		m(p)
+	}
+	return p
+}
+
+func TestTick_schedulingAndPruning(t *testing.T) {
+	t.Parallel()
+	s := runtime.NewScheme()
+	if err := crawlv1alpha1.AddToScheme(s); err != nil {
+		t.Fatalf("scheme: %v", err)
+	}
+
+	probeable := externalProxy("probeable", "10.0.0.1")
+	seeded := externalProxy("seeded", "10.0.0.2", func(p *crawlv1alpha1.Proxy) {
+		p.Status.Conditions = []metav1.Condition{{
+			Type: crawlv1alpha1.ConditionHealthy, Status: metav1.ConditionTrue,
+			Reason: "ProbeSucceeded", LastTransitionTime: metav1.Now(),
+		}}
+		p.Status.LatencyMillis = 42
+	})
+	noIP := &crawlv1alpha1.Proxy{
+		ObjectMeta: metav1.ObjectMeta{Name: "no-ip", Namespace: "default", UID: "uid-no-ip"},
+		Spec:       crawlv1alpha1.ProxySpec{Mode: crawlv1alpha1.ModeManaged, Provider: "stub"},
+	}
+
+	e := testEngine()
+	e.Reader = fake.NewClientBuilder().WithScheme(s).
+		WithObjects(probeable, seeded, noIP).Build()
+	// Stale entries: one for a proxy that no longer exists, one under a key
+	// that now belongs to a different UID (delete + recreate).
+	e.states[types.NamespacedName{Namespace: "default", Name: "gone"}] = &state{uid: "uid-gone"}
+	e.states[types.NamespacedName{Namespace: "default", Name: "probeable"}] = &state{
+		uid: "uid-previous-incarnation", healthy: boolPtr(false),
+	}
+
+	jobs := make(chan probeJob, 8)
+	e.tick(context.Background(), time.Now(), jobs)
+
+	var dispatched []probeJob
+	for {
+		select {
+		case j := <-jobs:
+			dispatched = append(dispatched, j)
+			continue
+		default:
+		}
+		break
+	}
+
+	if len(dispatched) != 1 {
+		t.Fatalf("dispatched %d jobs, want exactly 1 (only the fresh probeable proxy)", len(dispatched))
+	}
+	j := dispatched[0]
+	if j.key.Name != "probeable" || j.uid != "uid-probeable" {
+		t.Errorf("dispatched job = %+v, want the recreated probeable proxy", j)
+	}
+	if want := "http://" + "10.0.0.1:" + strconv.Itoa(3128); j.proxyURL.String() != want {
+		t.Errorf("proxyURL = %s, want %s", j.proxyURL, want)
+	}
+
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if _, ok := e.states[types.NamespacedName{Namespace: "default", Name: "gone"}]; ok {
+		t.Error("state for a deleted proxy was not pruned")
+	}
+	if _, ok := e.states[types.NamespacedName{Namespace: "default", Name: "no-ip"}]; ok {
+		t.Error("state was created for a proxy with no IP")
+	}
+	st := e.states[types.NamespacedName{Namespace: "default", Name: "probeable"}]
+	if st == nil || st.uid != "uid-probeable" {
+		t.Fatalf("state for recreated proxy = %+v, want fresh state with the new UID", st)
+	}
+	if st.healthy != nil && !*st.healthy {
+		t.Error("recreated proxy inherited the previous incarnation's unhealthy verdict")
+	}
+	seededSt := e.states[types.NamespacedName{Namespace: "default", Name: "seeded"}]
+	if seededSt == nil {
+		t.Fatal("no state created for the seeded proxy")
+	}
+	if seededSt.healthy == nil || !*seededSt.healthy {
+		t.Error("seeded proxy did not inherit its Healthy condition")
+	}
+	if seededSt.reportedHealthy == nil || !*seededSt.reportedHealthy {
+		t.Error("seeded verdict must count as already reported, or restart would re-emit for the whole fleet")
+	}
+	if seededSt.reportedLatency != 42*time.Millisecond {
+		t.Errorf("seeded reportedLatency = %v, want 42ms", seededSt.reportedLatency)
+	}
+	if seededSt.consecOK != 0 || seededSt.consecFail != 0 {
+		t.Error("seeded counters must start at zero")
+	}
+}
+
+func TestEngine_StartEndToEnd(t *testing.T) {
+	t.Parallel()
+	s := runtime.NewScheme()
+	if err := crawlv1alpha1.AddToScheme(s); err != nil {
+		t.Fatalf("scheme: %v", err)
+	}
+
+	e := testEngine()
+	e.Tick = 5 * time.Millisecond
+	e.Reader = fake.NewClientBuilder().WithScheme(s).
+		WithObjects(externalProxy("p1", "192.0.2.1")).Build()
+	e.probeFn = func(context.Context, *url.URL, crawlv1alpha1.HealthCheckSpec, *tls.Config) probeResult {
+		return probeResult{ok: true, latency: 12 * time.Millisecond}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- e.Start(ctx) }()
+
+	select {
+	case evt := <-e.Events:
+		if evt.Object.GetName() != "p1" {
+			t.Errorf("event for %q, want p1", evt.Object.GetName())
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("no health event within 5s")
+	}
+	snap, ok := e.Snapshot(types.NamespacedName{Namespace: "default", Name: "p1"})
+	if !ok || !snap.Healthy || snap.Latency != 12*time.Millisecond {
+		t.Errorf("Snapshot = %+v, %v; want healthy at 12ms", snap, ok)
+	}
+
+	cancel()
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Errorf("Start returned %v, want nil on context cancel", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Start did not stop within 5s of cancel")
+	}
+}

internal/health/probe.go

@@ -0,0 +1,66 @@
+// Package health actively probes every proxy by fetching a URL through the
+// proxy itself, keeps per-proxy threshold state, and pushes status-affecting
+// transitions to the reconciler over a channel. The engine owns health
+// state; the reconciler owns its representation in the Proxy's status.
+package health
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"slices"
+	"time"
+
+	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
+)
+
+// probeResult is the outcome of a single through-the-proxy probe.
+type probeResult struct {
+	ok      bool
+	latency time.Duration
+	err     error
+}
+
+// probe fetches hc.ProbeURL through the proxy at proxyURL. For an https
+// probe URL the transport issues CONNECT to the proxy and TLS-handshakes
+// through the tunnel; a proxy that accepts TCP but cannot egress answers
+// CONNECT with a non-200, which client.Do surfaces as an error, not a
+// response — so success requires err == nil AND an expected status code.
+// tlsCfg is nil in production (system roots); tests and private-CA setups
+// inject their own.
+func probe(ctx context.Context, proxyURL *url.URL, hc crawlv1alpha1.HealthCheckSpec, tlsCfg *tls.Config) probeResult {
+	timeout := time.Duration(hc.TimeoutSeconds) * time.Second
+	transport := &http.Transport{
+		Proxy: http.ProxyURL(proxyURL),
+		// Load-bearing: with keep-alives on, net/http caches the established
+		// CONNECT tunnel and later probes would never re-exercise CONNECT —
+		// exactly the failure this probe exists to catch.
+		DisableKeepAlives:     true,
+		ForceAttemptHTTP2:     false,
+		TLSHandshakeTimeout:   timeout,
+		ResponseHeaderTimeout: timeout,
+		TLSClientConfig:       tlsCfg,
+		DialContext:           (&net.Dialer{Timeout: timeout}).DialContext,
+	}
+	defer transport.CloseIdleConnections()
+
+	client := &http.Client{Transport: transport, Timeout: timeout}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, hc.ProbeURL, nil)
+	if err != nil {
+		return probeResult{err: fmt.Errorf("building probe request: %w", err)}
+	}
+	start := time.Now()
+	resp, err := client.Do(req)
+	latency := time.Since(start)
+	if err != nil {
+		return probeResult{latency: latency, err: err}
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if !slices.Contains(hc.ExpectedStatusCodes, int32(resp.StatusCode)) {
+		return probeResult{latency: latency, err: fmt.Errorf("unexpected status %d", resp.StatusCode)}
+	}
+	return probeResult{ok: true, latency: latency}
+}

internal/health/probe_test.go

@@ -0,0 +1,169 @@
+package health
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
+)
+
+// startConnectProxy runs a minimal but real HTTP proxy: CONNECT tunneling
+// for https targets, absolute-URI forwarding for plain http ones. With
+// refuseConnect it answers CONNECT with 502 — the "accepts TCP but cannot
+// egress" failure mode the probe must classify as unhealthy.
+func startConnectProxy(t *testing.T, refuseConnect bool) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodConnect {
+			if refuseConnect {
+				http.Error(w, "no egress", http.StatusBadGateway)
+				return
+			}
+			dst, err := net.DialTimeout("tcp", r.Host, time.Second)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusBadGateway)
+				return
+			}
+			conn, bufrw, err := http.NewResponseController(w).Hijack()
+			if err != nil {
+				_ = dst.Close()
+				t.Errorf("hijack: %v", err)
+				return
+			}
+			_, _ = bufrw.WriteString("HTTP/1.1 200 Connection established\r\n\r\n")
+			_ = bufrw.Flush()
+			done := make(chan struct{}, 2)
+			go func() { _, _ = io.Copy(dst, bufrw); done <- struct{}{} }()
+			go func() { _, _ = io.Copy(conn, dst); done <- struct{}{} }()
+			<-done
+			_ = conn.Close()
+			_ = dst.Close()
+			return
+		}
+		out := r.Clone(r.Context())
+		out.RequestURI = ""
+		resp, err := http.DefaultTransport.RoundTrip(out)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadGateway)
+			return
+		}
+		defer func() { _ = resp.Body.Close() }()
+		for k, vv := range resp.Header {
+			for _, v := range vv {
+				w.Header().Add(k, v)
+			}
+		}
+		w.WriteHeader(resp.StatusCode)
+		_, _ = io.Copy(w, resp.Body)
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func startTLSTarget(t *testing.T, status int) (*httptest.Server, *tls.Config) {
+	t.Helper()
+	target := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(status)
+	}))
+	t.Cleanup(target.Close)
+	pool := x509.NewCertPool()
+	pool.AddCert(target.Certificate())
+	return target, &tls.Config{RootCAs: pool}
+}
+
+func proxyURL(t *testing.T, srv *httptest.Server) *url.URL {
+	t.Helper()
+	u, err := url.Parse(srv.URL)
+	if err != nil {
+		t.Fatalf("parsing proxy URL: %v", err)
+	}
+	return u
+}
+
+func testHC(probeTarget string) crawlv1alpha1.HealthCheckSpec {
+	return crawlv1alpha1.HealthCheckSpec{
+		ProbeURL:            probeTarget,
+		IntervalSeconds:     30,
+		TimeoutSeconds:      5,
+		FailureThreshold:    3,
+		SuccessThreshold:    1,
+		ExpectedStatusCodes: []int32{200, 204},
+	}
+}
+
+func TestProbe_connectTunnelSucceeds(t *testing.T) {
+	t.Parallel()
+	target, tlsCfg := startTLSTarget(t, http.StatusNoContent)
+	proxy := startConnectProxy(t, false)
+
+	res := probe(context.Background(), proxyURL(t, proxy), testHC(target.URL), tlsCfg)
+	if !res.ok {
+		t.Fatalf("probe failed through working proxy: %v", res.err)
+	}
+	if res.latency <= 0 {
+		t.Errorf("latency = %v, want > 0", res.latency)
+	}
+}
+
+func TestProbe_refusedConnectFails(t *testing.T) {
+	t.Parallel()
+	target, tlsCfg := startTLSTarget(t, http.StatusNoContent)
+	proxy := startConnectProxy(t, true)
+
+	res := probe(context.Background(), proxyURL(t, proxy), testHC(target.URL), tlsCfg)
+	if res.ok {
+		t.Fatal("probe succeeded through a proxy that refuses CONNECT")
+	}
+	if res.err == nil {
+		t.Error("expected an error from the refused CONNECT")
+	}
+}
+
+func TestProbe_unexpectedStatusFails(t *testing.T) {
+	t.Parallel()
+	target, tlsCfg := startTLSTarget(t, http.StatusInternalServerError)
+	proxy := startConnectProxy(t, false)
+
+	res := probe(context.Background(), proxyURL(t, proxy), testHC(target.URL), tlsCfg)
+	if res.ok {
+		t.Fatal("probe succeeded on a 500 response")
+	}
+}
+
+func TestProbe_unreachableProxyFails(t *testing.T) {
+	t.Parallel()
+	// A listener that is immediately closed: guaranteed-refused port.
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("reserving port: %v", err)
+	}
+	dead := &url.URL{Scheme: "http", Host: l.Addr().String()}
+	_ = l.Close()
+
+	res := probe(context.Background(), dead, testHC("https://example.invalid/"), nil)
+	if res.ok {
+		t.Fatal("probe succeeded against a dead proxy")
+	}
+}
+
+func TestProbe_plainHTTPForwardSucceeds(t *testing.T) {
+	t.Parallel()
+	target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	t.Cleanup(target.Close)
+	proxy := startConnectProxy(t, false)
+
+	res := probe(context.Background(), proxyURL(t, proxy), testHC(target.URL), nil)
+	if !res.ok {
+		t.Fatalf("plain-http probe failed: %v", res.err)
+	}
+}