egress-proxies-operator/internal/providerhealth/monitor.go

// Package providerhealth checks every configured provider on an interval
// and publishes the results: one ProviderHealth object per provider, a log
// line whenever a target goes down or comes back, and a snapshot the
// metrics collector reads at scrape time.
package providerhealth

import (
	"context"
	"fmt"
	"slices"
	"sync"
	"time"

	"github.com/go-logr/logr"
	"go.opentelemetry.io/otel/attribute"
	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/validation"
	"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"
	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
)

const (
	// labelManagedBy marks the ProviderHealth objects this operator owns;
	// cleanup never deletes an object without it.
	labelManagedBy = "app.kubernetes.io/managed-by"
	managedByValue = "egress-proxies-operator"

	minInterval    = 10 * time.Second
	defaultTimeout = 15 * time.Second
)

// Target is one configured provider. Checker is nil for a provider that
// cannot check itself.
type Target struct {
	Name    string
	Type    string
	Checker provider.Checker
}

// Result is the outcome of one provider's latest check.
type Result struct {
	Provider string
	Type     string
	// Supported is false when the provider has no Checker.
	Supported bool
	Targets   []provider.TargetStatus
	CheckedAt time.Time
}

// controller-gen reads RBAC markers only at package level, so they stay
// out of Monitor's doc comment.
// +kubebuilder:rbac:groups=crawl.example.com,resources=providerhealths,verbs=get;list;create;delete
// +kubebuilder:rbac:groups=crawl.example.com,resources=providerhealths/status,verbs=get;patch

// Monitor is the manager Runnable that checks the providers.
type Monitor struct {
	// Client writes the ProviderHealth objects.
	Client client.Client
	// Reader reads them; pass the manager's API reader. The objects are
	// cluster-scoped, and reading them through the cache would start a
	// cluster-wide informer needing watch rights, whatever --proxy-namespace
	// says, for a handful of reads a minute.
	Reader client.Reader
	// Targets are the configured providers, in config order.
	Targets []Target
	// Interval between rounds of checks; 0 turns checking off. The first
	// round runs at start: nothing races it, and early visibility is the
	// point.
	Interval time.Duration
	// Timeout bounds each round (default 15s).
	Timeout time.Duration

	now func() time.Time

	mu     sync.Mutex
	latest []Result

	// prev holds each target's previous verdict for transition logging.
	// Only the loop goroutine touches it.
	prev map[string]provider.TargetStatus
}

// NeedLeaderElection is true: only the leader dials the hypervisors and
// writes the objects, and only its gauges describe the checks.
func (m *Monitor) NeedLeaderElection() bool { return true }

// ValidateSchedule checks --provider-check-interval and
// --provider-check-timeout. The interval is 0 (off) or at least 10s, which
// is what keeps one status write per provider per round cheap; the timeout
// is positive and no longer than the interval.
func ValidateSchedule(interval, timeout time.Duration) error {
	switch {
	case interval == 0:
		return nil
	case interval < minInterval:
		return fmt.Errorf("--provider-check-interval must be 0 (off) or at least %s, got %s", minInterval, interval)
	case timeout <= 0 || timeout > interval:
		return fmt.Errorf("--provider-check-timeout must be positive and at most the interval (%s), got %s", interval, timeout)
	}
	return nil
}

// Start deletes the objects of providers that are no longer configured,
// then checks every Interval until ctx ends. The providers config is read
// once per process, so cleaning up once is enough. With checking off,
// every object goes, so none is left claiming a stale Ready.
func (m *Monitor) Start(ctx context.Context) error {
	if m.Timeout == 0 {
		m.Timeout = defaultTimeout
	}
	if m.now == nil {
		m.now = time.Now
	}
	log := logf.FromContext(ctx).WithName("provider-health")
	ctx = logf.IntoContext(ctx, log)

	keep := map[string]bool{}
	if m.Interval > 0 {
		for _, t := range m.Targets {
			keep[t.Name] = true
		}
	}
	m.cleanup(ctx, keep)
	if m.Interval == 0 {
		log.Info("Provider checks are off (--provider-check-interval=0)")
		return nil
	}
	for _, t := range m.Targets {
		if errs := validation.IsDNS1123Subdomain(t.Name); len(errs) > 0 {
			log.Info("WARNING: provider name is not a valid object name; its ProviderHealth object is not written, its metrics are",
				"provider", t.Name, "problems", errs)
		}
	}

	m.checkOnce(ctx)
	ticker := time.NewTicker(m.Interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			m.checkOnce(ctx)
		}
	}
}

// Snapshot returns the latest results in config order.
func (m *Monitor) Snapshot() []Result {
	m.mu.Lock()
	defer m.mu.Unlock()
	out := make([]Result, len(m.latest))
	for i, r := range m.latest {
		r.Targets = slices.Clone(r.Targets)
		out[i] = r
	}
	return out
}

// checkOnce runs every provider's check concurrently under one timeout,
// then records, logs and publishes the results provider by provider. A
// failed write is logged and retried by the next round.
func (m *Monitor) checkOnce(ctx context.Context) {
	ctx, span := tracing.Start(ctx, "providerhealth.check")
	defer span.End()
	log := logf.FromContext(ctx)

	checkCtx, cancel := context.WithTimeout(ctx, m.Timeout)
	defer cancel()
	results := make([]Result, len(m.Targets))
	var wg sync.WaitGroup
	for i, t := range m.Targets {
		wg.Go(func() { results[i] = m.run(checkCtx, t) })
	}
	wg.Wait()

	m.mu.Lock()
	m.latest = results
	m.mu.Unlock()

	notReady := 0
	for _, r := range results {
		m.logTransitions(log, r)
		for _, ts := range r.Targets {
			if !ts.Ready {
				notReady++
			}
		}
		if err := m.publish(ctx, r); err != nil {
			log.Error(err, "Writing provider health", "provider", r.Provider)
		}
	}
	span.SetAttributes(
		attribute.Int("providerhealth.providers", len(results)),
		attribute.Int("providerhealth.targets_not_ready", notReady),
	)
}

func (m *Monitor) run(ctx context.Context, t Target) Result {
	r := Result{Provider: t.Name, Type: t.Type, Supported: t.Checker != nil}
	if t.Checker != nil {
		r.Targets = t.Checker.Check(ctx)
	}
	r.CheckedAt = m.now()
	return r
}

// logTransitions logs a target going down, staying down for a different
// reason, or coming back. Steady states stay quiet; a target that is ready
// the first time it is seen only logs at V(1).
func (m *Monitor) logTransitions(log logr.Logger, r Result) {
	if m.prev == nil {
		m.prev = map[string]provider.TargetStatus{}
	}
	for _, ts := range r.Targets {
		key := r.Provider + "/" + ts.Name
		prev, seen := m.prev[key]
		m.prev[key] = ts
		kv := []any{"provider", r.Provider, "target", ts.Name, "address", ts.Address, "reason", ts.Reason}
		switch {
		case !ts.Ready && (!seen || prev.Ready || prev.Reason != ts.Reason):
			log.Info("WARNING: provider target not ready", append(kv, "message", ts.Message)...)
		case ts.Ready && seen && !prev.Ready:
			log.Info("Provider target ready", append(kv, "latency", ts.Latency.Round(time.Millisecond))...)
		case ts.Ready && !seen:
			log.V(1).Info("Provider target ready", append(kv, "latency", ts.Latency.Round(time.Millisecond))...)
		}
	}
}

// publish creates the provider's ProviderHealth object when missing and
// patches its status. The status changes on every round, since
// lastCheckTime moves; the comparison only skips the patch when a clock
// did not advance.
func (m *Monitor) publish(ctx context.Context, r Result) error {
	if len(validation.IsDNS1123Subdomain(r.Provider)) > 0 {
		return nil
	}
	var ph crawlv1alpha1.ProviderHealth
	err := m.Reader.Get(ctx, client.ObjectKey{Name: r.Provider}, &ph)
	switch {
	case apierrors.IsNotFound(err):
		ph = crawlv1alpha1.ProviderHealth{ObjectMeta: metav1.ObjectMeta{
			Name:   r.Provider,
			Labels: map[string]string{labelManagedBy: managedByValue},
		}}
		if err := m.Client.Create(ctx, &ph); err != nil {
			return fmt.Errorf("creating ProviderHealth %s: %w", r.Provider, err)
		}
	case err != nil:
		return fmt.Errorf("reading ProviderHealth %s: %w", r.Provider, err)
	}
	base := ph.DeepCopy()
	ph.Status = buildStatus(ph.Status, r)
	if equality.Semantic.DeepEqual(base.Status, ph.Status) {
		return nil
	}
	if err := m.Client.Status().Patch(ctx, &ph, client.MergeFrom(base)); err != nil {
		return fmt.Errorf("patching ProviderHealth %s status: %w", r.Provider, err)
	}
	return nil
}

// cleanup deletes this operator's ProviderHealth objects whose name is not
// in keep.
func (m *Monitor) cleanup(ctx context.Context, keep map[string]bool) {
	log := logf.FromContext(ctx)
	var list crawlv1alpha1.ProviderHealthList
	if err := m.Reader.List(ctx, &list, client.MatchingLabels{labelManagedBy: managedByValue}); err != nil {
		log.Error(err, "Listing provider health objects; stale ones stay until the next start")
		return
	}
	for i := range list.Items {
		ph := &list.Items[i]
		if keep[ph.Name] {
			continue
		}
		if err := m.Client.Delete(ctx, ph); client.IgnoreNotFound(err) != nil {
			log.Error(err, "Deleting provider health of an unconfigured provider", "name", ph.Name)
			continue
		}
		log.Info("Deleted provider health of an unconfigured provider", "name", ph.Name)
	}
}