Add the Proxy reconciler state machine with action-table, phase, and envtest suites

Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-09 14:03:02 +02:00
parent 5c408cc284
commit 1125f74221
11 changed files with 1566 additions and 74 deletions
--- a/internal/controller/proxy_controller.go
+++ b/internal/controller/proxy_controller.go
@@ -18,46 +18,381 @@ package controller

 import (
 	"context"
+	"errors"
+	"fmt"
+	"time"

+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"

 	crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
+	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
 )

-// ProxyReconciler reconciles a Proxy object
+// ProxyReconciler reconciles Proxy objects as a state machine: every
+// reconcile derives exactly one action from (spec, status, provider Get),
+// performs it, and requeues. Status is written at most once per reconcile,
+// by the deferred patch in Reconcile.
 type ProxyReconciler struct {
 	client.Client
 	Scheme *runtime.Scheme
+
+	// Providers maps spec.provider values to configured backends.
+	Providers map[string]provider.Provider
+
+	// Poll intervals are struct fields, never consts, so tests can shrink
+	// them to milliseconds.
+	ProvisioningPoll time.Duration // while waiting for an instance to reach Running
+	DriftPoll        time.Duration // between re-checks of a Running instance
+	DeletionPoll     time.Duration // while waiting for an instance to disappear
+	QuotaRetry       time.Duration // after ErrQuotaExceeded; slow, off the backoff curve
+	RequeueNow       time.Duration // "process the next state promptly" (Result.Requeue is deprecated)
 }

 // +kubebuilder:rbac:groups=crawl.example.com,resources=proxies,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=crawl.example.com,resources=proxies/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=crawl.example.com,resources=proxies/finalizers,verbs=update
+// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch

-// Reconcile is part of the main kubernetes reconciliation loop which aims to
-// move the current state of the cluster closer to the desired state.
-// TODO(user): Modify the Reconcile function to compare the state specified by
-// the Proxy object against the actual cluster state, and then
-// perform operations to make the cluster state reflect the state specified by
-// the user.
-//
-// For more details, check Reconcile and its Result here:
-// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.24.1/pkg/reconcile
-func (r *ProxyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	_ = logf.FromContext(ctx)
+// Reconcile fetches the Proxy named by req into p (r.Get fills the struct
+// through the pointer), dispatches to the delete/external/managed state
+// machines, and flushes any status change exactly once on the way out.
+func (r *ProxyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (res ctrl.Result, err error) {
+	var p crawlv1alpha1.Proxy
+	if err := r.Get(ctx, req.NamespacedName, &p); err != nil {
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+	base := p.DeepCopy()
+	defer func() {
+		// NotFound is expected when this reconcile just removed the last
+		// finalizer and the object is already gone.
+		if perr := r.patchStatusIfChanged(ctx, base, &p); perr != nil && !apierrors.IsNotFound(perr) {
+			err = errors.Join(err, perr)
+		}
+	}()

-	// TODO(user): your logic here
+	switch {
+	case !p.DeletionTimestamp.IsZero():
+		return r.reconcileDelete(ctx, &p)
+	case p.Spec.Mode == crawlv1alpha1.ModeExternal:
+		return r.reconcileExternal(ctx, &p)
+	default:
+		return r.reconcileManaged(ctx, &p)
+	}
+}

+func (r *ProxyReconciler) reconcileManaged(ctx context.Context, p *crawlv1alpha1.Proxy) (ctrl.Result, error) {
+	log := logf.FromContext(ctx)
+
+	if controllerutil.AddFinalizer(p, crawlv1alpha1.FinalizerName) {
+		// The Update event re-triggers reconciliation; provisioning starts
+		// on the next pass, with the finalizer safely persisted first.
+		return ctrl.Result{}, r.Update(ctx, p)
+	}
+
+	// Permanent-failure latch: once this generation has failed permanently,
+	// stop calling the provider until the spec changes.
+	if cond := apimeta.FindStatusCondition(p.Status.Conditions, crawlv1alpha1.ConditionProvisioned); cond != nil &&
+		cond.Status == metav1.ConditionFalse && cond.Reason == ReasonPermanentError &&
+		cond.ObservedGeneration == p.Generation {
+		return ctrl.Result{}, nil
+	}
+
+	prov, ok := r.Providers[p.Spec.Provider]
+	if !ok {
+		setProvisioned(p, metav1.ConditionFalse, ReasonPermanentError,
+			fmt.Sprintf("provider %q is not configured", p.Spec.Provider))
+		return ctrl.Result{}, nil
+	}
+
+	cloudInit, err := r.resolveCloudInit(ctx, p)
+	if err != nil {
+		setProvisioned(p, metav1.ConditionFalse, ReasonCloudInitError, err.Error())
+		return ctrl.Result{}, err
+	}
+	hash := specHash(p, cloudInit)
+
+	if p.Status.ProviderID == "" {
+		id, err := prov.Create(ctx, provider.CreateRequest{
+			Name:      provider.NameFromUID(p.UID),
+			UID:       string(p.UID),
+			Namespace: p.Namespace,
+			ProxyName: p.Name,
+			Placement: placementFrom(p.Spec.Placement),
+			CloudInit: cloudInit,
+			Port:      p.EffectivePort(),
+		})
+		if err != nil {
+			return r.providerFailure(p, err)
+		}
+		log.Info("created instance", "provider", p.Spec.Provider, "providerID", id)
+		if err := r.setSpecHash(ctx, p, hash); err != nil {
+			return ctrl.Result{}, err
+		}
+		p.Status.ProviderID = id
+		p.Status.IP = ""
+		setProvisioned(p, metav1.ConditionFalse, ReasonProvisioning, "instance created; waiting for it to run")
+		return ctrl.Result{RequeueAfter: r.ProvisioningPoll}, nil
+	}
+
+	if ann := p.Annotations[crawlv1alpha1.AnnotationSpecHash]; ann != hash {
+		if ann == "" {
+			// Adopt: an instance provisioned before the hash-input struct
+			// gained a field (or by an older operator version) keeps its
+			// instance; replacing the whole fleet on upgrade would be wrong.
+			if err := r.setSpecHash(ctx, p, hash); err != nil {
+				return ctrl.Result{}, err
+			}
+			return ctrl.Result{RequeueAfter: r.RequeueNow}, nil
+		}
+		return r.replaceInstance(ctx, p, prov, hash)
+	}
+
+	inst, err := prov.Get(ctx, p.Status.ProviderID)
+	if provider.Class(err) == provider.ErrNotFound {
+		p.Status.ProviderID = ""
+		p.Status.IP = ""
+		return ctrl.Result{RequeueAfter: r.RequeueNow}, nil
+	}
+	if err != nil {
+		return r.providerFailure(p, err)
+	}
+
+	switch inst.State {
+	case provider.StateProvisioning:
+		p.Status.IP = ""
+		setProvisioned(p, metav1.ConditionFalse, ReasonProvisioning, "waiting for the instance to run")
+		return ctrl.Result{RequeueAfter: r.ProvisioningPoll}, nil
+	case provider.StateRunning:
+		p.Status.IP = inst.IP
+		setProvisioned(p, metav1.ConditionTrue, ReasonCreated, "instance is running")
+		return ctrl.Result{RequeueAfter: r.DriftPoll}, nil
+	default: // Stopped, Terminated: cattle, not pets — delete and recreate.
+		if err := prov.Delete(ctx, p.Status.ProviderID); err != nil {
+			return r.providerFailure(p, err)
+		}
+		log.Info("deleting instance for recreation", "providerID", p.Status.ProviderID, "state", inst.State)
+		p.Status.IP = ""
+		setProvisioned(p, metav1.ConditionFalse, ReasonRecreating,
+			fmt.Sprintf("instance is %s; deleting it for recreation", inst.State))
+		return ctrl.Result{RequeueAfter: r.DeletionPoll}, nil
+	}
+}
+
+// replaceInstance handles a spec-hash mismatch. The replacement instance has
+// the same deterministic name as the old one (both derive from the CR UID),
+// so recreating before the old instance is fully gone would hit "already
+// exists" — hence: delete, poll to NotFound, only then advance the hash and
+// let the create branch run.
+func (r *ProxyReconciler) replaceInstance(ctx context.Context, p *crawlv1alpha1.Proxy, prov provider.Provider, hash string) (ctrl.Result, error) {
+	_, err := prov.Get(ctx, p.Status.ProviderID)
+	if provider.Class(err) == provider.ErrNotFound {
+		// Old instance is gone. The Update inside setSpecHash refreshes p
+		// from the server — including status — so the status clear must be
+		// staged after it, or it would be silently overwritten. A crash
+		// between the two writes recovers either way: the create branch's
+		// Create is idempotent by name, and a stale ID resolves to NotFound
+		// again.
+		if err := r.setSpecHash(ctx, p, hash); err != nil {
+			return ctrl.Result{}, err
+		}
+		p.Status.ProviderID = ""
+		p.Status.IP = ""
+		return ctrl.Result{RequeueAfter: r.RequeueNow}, nil
+	}
+	if err != nil {
+		return r.providerFailure(p, err)
+	}
+	if err := prov.Delete(ctx, p.Status.ProviderID); err != nil {
+		return r.providerFailure(p, err)
+	}
+	logf.FromContext(ctx).Info("replacing instance after spec change", "providerID", p.Status.ProviderID)
+	p.Status.IP = ""
+	setProvisioned(p, metav1.ConditionFalse, ReasonReplacing, "spec changed; deleting the old instance before recreating")
+	return ctrl.Result{RequeueAfter: r.DeletionPoll}, nil
+}
+
+func (r *ProxyReconciler) reconcileDelete(ctx context.Context, p *crawlv1alpha1.Proxy) (ctrl.Result, error) {
+	if !controllerutil.ContainsFinalizer(p, crawlv1alpha1.FinalizerName) {
+		return ctrl.Result{}, nil
+	}
+	if p.Status.ProviderID == "" {
+		// Nothing was ever recorded as created; orphan GC reaps any stray
+		// instance a crashed create might have left behind.
+		controllerutil.RemoveFinalizer(p, crawlv1alpha1.FinalizerName)
+		return ctrl.Result{}, r.Update(ctx, p)
+	}
+	prov, ok := r.Providers[p.Spec.Provider]
+	if !ok {
+		return ctrl.Result{}, fmt.Errorf(
+			"provider %q is not configured; cannot clean up instance %s", p.Spec.Provider, p.Status.ProviderID)
+	}
+	_, err := prov.Get(ctx, p.Status.ProviderID)
+	if provider.Class(err) == provider.ErrNotFound {
+		controllerutil.RemoveFinalizer(p, crawlv1alpha1.FinalizerName)
+		return ctrl.Result{}, r.Update(ctx, p)
+	}
+	if err != nil {
+		return r.deletionFailure(err)
+	}
+	if err := prov.Delete(ctx, p.Status.ProviderID); err != nil {
+		return r.deletionFailure(err)
+	}
+	logf.FromContext(ctx).Info("deleting instance", "providerID", p.Status.ProviderID)
+	setProvisioned(p, metav1.ConditionFalse, ReasonDeleting, "deleting the instance before removing the finalizer")
+	return ctrl.Result{RequeueAfter: r.DeletionPoll}, nil
+}
+
+func (r *ProxyReconciler) reconcileExternal(_ context.Context, p *crawlv1alpha1.Proxy) (ctrl.Result, error) {
+	if p.Spec.Endpoint == nil {
+		// CEL guarantees an endpoint on any object that went through the API
+		// server; tolerate its absence instead of panicking.
+		setProvisioned(p, metav1.ConditionFalse, ReasonPermanentError, "external proxy has no endpoint")
+		return ctrl.Result{}, nil
+	}
+	p.Status.IP = p.Spec.Endpoint.Host
+	setProvisioned(p, metav1.ConditionTrue, ReasonExternalEndpoint, "tracking an external endpoint")
 	return ctrl.Result{}, nil
 }

-// SetupWithManager sets up the controller with the Manager.
+// providerFailure translates a classified provider error into the
+// state-machine's reaction: transient errors ride the workqueue's
+// exponential backoff, quota errors back off slowly without counting as
+// errors, and permanent errors latch Failed and stop retrying.
+func (r *ProxyReconciler) providerFailure(p *crawlv1alpha1.Proxy, err error) (ctrl.Result, error) {
+	switch provider.Class(err) {
+	case provider.ErrQuotaExceeded:
+		setProvisioned(p, metav1.ConditionFalse, ReasonQuotaExceeded, err.Error())
+		return ctrl.Result{RequeueAfter: r.QuotaRetry}, nil
+	case provider.ErrPermanent:
+		setProvisioned(p, metav1.ConditionFalse, ReasonPermanentError, err.Error())
+		return ctrl.Result{}, nil
+	default:
+		return ctrl.Result{}, err
+	}
+}
+
+// deletionFailure is providerFailure for the finalizer path, where latching
+// a permanent failure would wedge the object forever with no retry — keep
+// retrying instead, visibly, until cleanup succeeds or an operator
+// intervenes.
+func (r *ProxyReconciler) deletionFailure(err error) (ctrl.Result, error) {
+	if provider.Class(err) == provider.ErrQuotaExceeded {
+		return ctrl.Result{RequeueAfter: r.QuotaRetry}, nil
+	}
+	return ctrl.Result{}, err
+}
+
+// resolveCloudInit returns the effective cloud-init user-data, reading the
+// referenced Secret if one is used. Both the spec hash and CreateRequest see
+// only resolved content, so rotating a Secret triggers replacement.
+func (r *ProxyReconciler) resolveCloudInit(ctx context.Context, p *crawlv1alpha1.Proxy) (string, error) {
+	ci := p.Spec.CloudInit
+	if ci == nil {
+		return "", nil
+	}
+	if ci.Inline != "" {
+		return ci.Inline, nil
+	}
+	if ci.SecretRef == nil {
+		return "", nil
+	}
+	key := ci.SecretRef.Key
+	if key == "" {
+		key = crawlv1alpha1.DefaultCloudInitSecretKey
+	}
+	var sec corev1.Secret
+	if err := r.Get(ctx, client.ObjectKey{Namespace: p.Namespace, Name: ci.SecretRef.Name}, &sec); err != nil {
+		return "", fmt.Errorf("resolving cloudInit secret %q: %w", ci.SecretRef.Name, err)
+	}
+	data, ok := sec.Data[key]
+	if !ok {
+		return "", fmt.Errorf("cloudInit secret %q has no key %q", ci.SecretRef.Name, key)
+	}
+	return string(data), nil
+}
+
+// setSpecHash persists the spec-hash annotation. Status changes staged on p
+// are untouched by the Update (they live on the status subresource) and are
+// flushed by the deferred patch in Reconcile.
+func (r *ProxyReconciler) setSpecHash(ctx context.Context, p *crawlv1alpha1.Proxy, hash string) error {
+	if p.Annotations[crawlv1alpha1.AnnotationSpecHash] == hash {
+		return nil
+	}
+	if p.Annotations == nil {
+		p.Annotations = map[string]string{}
+	}
+	p.Annotations[crawlv1alpha1.AnnotationSpecHash] = hash
+	return r.Update(ctx, p)
+}
+
+func placementFrom(ps *crawlv1alpha1.PlacementSpec) provider.Placement {
+	if ps == nil {
+		return provider.Placement{}
+	}
+	return provider.Placement{
+		Region:      ps.Region,
+		Zone:        ps.Zone,
+		MachineType: ps.MachineType,
+		Image:       ps.Image,
+	}
+}
+
+// proxiesForSecret maps a Secret event to the Proxies whose cloudInit
+// references it, so rotating a Secret re-triggers the replacement check.
+func (r *ProxyReconciler) proxiesForSecret(ctx context.Context, obj client.Object) []reconcile.Request {
+	var list crawlv1alpha1.ProxyList
+	if err := r.List(ctx, &list, client.InNamespace(obj.GetNamespace())); err != nil {
+		logf.FromContext(ctx).Error(err, "listing proxies for secret event", "secret", obj.GetName())
+		return nil
+	}
+	var reqs []reconcile.Request
+	for i := range list.Items {
+		p := &list.Items[i]
+		if ci := p.Spec.CloudInit; ci != nil && ci.SecretRef != nil && ci.SecretRef.Name == obj.GetName() {
+			reqs = append(reqs, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(p)})
+		}
+	}
+	return reqs
+}
+
+// SetupWithManager sets up the controller with the Manager. The Secret watch
+// only fires for Secrets the manager's cache holds; the composition root
+// (cmd/main.go) restricts that cache to labelled cloud-init Secrets.
 func (r *ProxyReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	r.applyDefaults()
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&crawlv1alpha1.Proxy{}).
 		Named("proxy").
+		Watches(&corev1.Secret{}, handler.EnqueueRequestsFromMapFunc(r.proxiesForSecret)).
+		WithOptions(controller.Options{MaxConcurrentReconciles: 3}).
 		Complete(r)
 }
+
+func (r *ProxyReconciler) applyDefaults() {
+	if r.ProvisioningPoll == 0 {
+		r.ProvisioningPoll = 10 * time.Second
+	}
+	if r.DriftPoll == 0 {
+		r.DriftPoll = 2 * time.Minute
+	}
+	if r.DeletionPoll == 0 {
+		r.DeletionPoll = 10 * time.Second
+	}
+	if r.QuotaRetry == 0 {
+		r.QuotaRetry = 5 * time.Minute
+	}
+	if r.RequeueNow == 0 {
+		r.RequeueNow = time.Second
+	}
+}