egress-proxies-operator/internal/health/engine.go

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
}