johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package runtime

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"reflect"
	"strings"
	"time"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	kruntime "k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/util/workqueue"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
	"sigs.k8s.io/controller-runtime/pkg/cluster"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	"sigs.k8s.io/controller-runtime/pkg/source"
	mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"
	mchandler "sigs.k8s.io/multicluster-runtime/pkg/handler"
	"sigs.k8s.io/multicluster-runtime/pkg/multicluster"
	mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile"
)

const scopeMissRetries = 10

type Object interface {
	metav1.Object
	kruntime.Object
}

type Result struct {
	RequeueAfter time.Duration

	Waiting bool

	Backoff bool
}

type Reconciler[T Object] interface {
	Reconcile(ctx context.Context, req Request, obj T) (Result, error)
}

type ReconcilerFunc[T Object] func(ctx context.Context, req Request, obj T) (Result, error)

func (f ReconcilerFunc[T]) Reconcile(ctx context.Context, req Request, obj T) (Result, error) {
	return f(ctx, req, obj)
}

type Request struct {
	Cluster string

	Workspace string

	Namespace string

	Name string

	scope *scope
}

type scope struct {
	client client.Client

	scheme *kruntime.Scheme

	fieldManager string

	finalizer string

	sharedStatus bool

	foreignStatus []string

	log *slog.Logger
}

type Option func(*options)

type options struct {
	export string

	fieldManager string

	finalizer string

	owns []client.Object

	watches []watchInput

	only func(client.Object) bool

	inWorkspace func(context.Context, string) bool

	gone func(context.Context, Request)

	events []<-chan Request

	sharedStatus bool

	foreignStatus []string

	backoffBase time.Duration

	backoffMax time.Duration

	workers int

	err error
}

type watchInput struct {
	object client.Object

	export string

	mapFunc func(ctx context.Context, src Request, obj client.Object) []Request

	changed func(old, new client.Object) bool

	observe func(obj client.Object, gone bool)
}

type WatchFilter[R Object] struct {
	Changed func(old, new R) bool

	Observe func(obj R, gone bool)
}

func Export(name string) Option {
	return func(o *options) { o.export = strings.TrimSpace(name) }
}

func FieldManager(name string) Option {
	return func(o *options) { o.fieldManager = strings.TrimSpace(name) }
}

func Finalizer(name string) Option {
	return func(o *options) { o.finalizer = strings.TrimSpace(name) }
}

func Backoff(base, max time.Duration) Option {
	return func(o *options) { o.backoffBase, o.backoffMax = base, max }
}

func Workers(n int) Option {
	return func(o *options) { o.workers = n }
}

func SharedStatus() Option {
	return func(o *options) { o.sharedStatus = true }
}

func ForeignStatus(paths ...string) Option {
	return func(o *options) { o.foreignStatus = append(o.foreignStatus, paths...) }
}

func Events(ch <-chan Request) Option {
	return func(o *options) {
		if ch != nil {
			o.events = append(o.events, ch)
		}
	}
}

func InWorkspaces(keep func(workspace string) bool) Option {
	return func(o *options) { o.inWorkspace = inWorkspaces(keep) }
}

func InWorkspacesContext(keep func(ctx context.Context, workspace string) bool) Option {
	return func(o *options) { o.inWorkspace = keep }
}

func inWorkspaces(keep func(workspace string) bool) func(context.Context, string) bool {
	if keep == nil {
		return nil
	}
	return func(_ context.Context, workspace string) bool { return keep(workspace) }
}

func Gone(fn func(ctx context.Context, req Request)) Option {
	return func(o *options) { o.gone = fn }
}

func Only[T Object](keep func(T) bool) Option {
	return func(o *options) {
		if keep == nil {
			return
		}
		o.only = func(obj client.Object) bool {
			typed, ok := obj.(T)
			return ok && keep(typed)
		}
	}
}

func Owns[C Object]() Option {
	return func(o *options) {
		obj, err := newObject[C]()
		if err != nil {
			o.err = errors.Join(o.err, err)
			return
		}
		o.owns = append(o.owns, obj)
	}
}

func Watches[R Object](mapFunc func(ctx context.Context, src Request, obj R) []Request) Option {
	return WatchesIn("", mapFunc)
}

func WatchesIn[R Object](export string, mapFunc func(ctx context.Context, src Request, obj R) []Request) Option {
	return WatchesFiltered(export, WatchFilter[R]{}, mapFunc)
}

func WatchesFiltered[R Object](export string, f WatchFilter[R], mapFunc func(ctx context.Context, src Request, obj R) []Request) Option {
	return func(o *options) {
		obj, err := newObject[R]()
		if err != nil {
			o.err = errors.Join(o.err, err)
			return
		}
		if mapFunc == nil {
			o.err = errors.Join(o.err, fmt.Errorf("runtime: the watch of %T has no map func", obj))
			return
		}
		w := watchInput{object: obj, export: strings.TrimSpace(export), mapFunc: func(ctx context.Context, src Request, obj client.Object) []Request {
			typed, ok := obj.(R)
			if !ok {
				return nil
			}
			return mapFunc(ctx, src, typed)
		}}
		if f.Changed != nil {
			w.changed = func(old, new client.Object) bool {
				o, ok1 := old.(R)
				n, ok2 := new.(R)
				return !ok1 || !ok2 || f.Changed(o, n)
			}
		}
		if f.Observe != nil {
			w.observe = func(obj client.Object, gone bool) {
				if typed, ok := obj.(R); ok {
					f.Observe(typed, gone)
				}
			}
		}
		o.watches = append(o.watches, w)
	}
}

func filtered(w watchInput, inner mchandler.TypedEventHandlerFunc[client.Object, mcreconcile.Request]) mchandler.TypedEventHandlerFunc[client.Object, mcreconcile.Request] {
	if w.changed == nil && w.observe == nil {
		return inner
	}
	observe := func(obj client.Object, gone bool) {
		if w.observe != nil && obj != nil {
			w.observe(obj, gone)
		}
	}
	return func(cn multicluster.ClusterName, cl cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
		h := inner(cn, cl)
		return handler.TypedFuncs[client.Object, mcreconcile.Request]{
			CreateFunc: func(ctx context.Context, e event.TypedCreateEvent[client.Object], q workqueue.TypedRateLimitingInterface[mcreconcile.Request]) {
				observe(e.Object, false)
				h.Create(ctx, e, q)
			},
			UpdateFunc: func(ctx context.Context, e event.TypedUpdateEvent[client.Object], q workqueue.TypedRateLimitingInterface[mcreconcile.Request]) {
				observe(e.ObjectNew, false)
				if w.changed != nil && e.ObjectOld != nil && e.ObjectNew != nil && !w.changed(e.ObjectOld, e.ObjectNew) {
					return
				}
				h.Update(ctx, e, q)
			},
			DeleteFunc: func(ctx context.Context, e event.TypedDeleteEvent[client.Object], q workqueue.TypedRateLimitingInterface[mcreconcile.Request]) {
				observe(e.Object, true)
				h.Delete(ctx, e, q)
			},
			GenericFunc: func(ctx context.Context, e event.TypedGenericEvent[client.Object], q workqueue.TypedRateLimitingInterface[mcreconcile.Request]) {
				h.Generic(ctx, e, q)
			},
		}
	}
}

func For[T Object](m *Manager, name string, r Reconciler[T], opts ...Option) error {
	name = strings.TrimSpace(name)
	if m == nil {
		return errors.New("runtime: a manager is required")
	}
	if name == "" {
		return errors.New("runtime: a controller needs a name")
	}
	if r == nil {
		return fmt.Errorf("runtime: the controller %s has no reconciler", name)
	}
	o := options{fieldManager: FieldManagerPrefix + name, backoffBase: DefaultBackoffBase, backoffMax: DefaultBackoffMax, workers: 1}
	for _, opt := range opts {
		if opt != nil {
			opt(&o)
		}
	}
	if o.err != nil {
		return fmt.Errorf("runtime: the controller %s: %w", name, o.err)
	}
	if o.export == "" {
		return fmt.Errorf("runtime: the controller %s names no APIExport", name)
	}
	o.workers = opt.Positive(o.workers, 1)
	if o.inWorkspace == nil {
		o.inWorkspace = inWorkspaces(m.workspaces)
	}
	proto, err := newObject[T]()
	if err != nil {
		return fmt.Errorf("runtime: the controller %s: %w", name, err)
	}
	m.addExport(o.export)
	inExport := exportFilter(o.export)
	limiter := newBackoff(o.backoffBase, o.backoffMax)
	log := m.log.With("controller", name)
	b := mcbuilder.ControllerManagedBy(m.mgr).Named(name).
		For(proto, inExport, mcbuilder.WithPredicates(specOrDeletion())).
		WithOptions(controller.TypedOptions[mcreconcile.Request]{MaxConcurrentReconciles: o.workers, RateLimiter: limiter,
			SkipNameValidation: new(true)})
	for _, owned := range o.owns {
		b = b.Owns(owned, inExport)
	}
	for _, w := range o.watches {
		source := inExport
		if w.export != "" && w.export != o.export {
			m.addExport(w.export)
			source = exportFilter(w.export)
		}
		b = b.Watches(w.object, filtered(w, m.mapHandler(o.export, w.mapFunc)), source)
	}
	for _, ch := range o.events {
		b = b.WatchesRawSource(m.eventSource(o.export, ch))
	}
	a := &adapter[T]{m: m, name: name, r: r, limiter: limiter, log: log, only: o.only, inWorkspace: o.inWorkspace, gone: o.gone,
		scope: scope{scheme: m.scheme, fieldManager: o.fieldManager, finalizer: o.finalizer, sharedStatus: o.sharedStatus,
			foreignStatus: o.foreignStatus, log: log}}
	if err := b.Complete(a); err != nil {
		return fmt.Errorf("runtime: building the controller %s: %w", name, err)
	}
	log.Info("the controller is registered", "kind", fmt.Sprintf("%T", proto), "export", o.export, "fieldManager", o.fieldManager)
	return nil
}

func (m *Manager) mapHandler(export string, mapFunc func(ctx context.Context, src Request, obj client.Object) []Request) mchandler.TypedEventHandlerFunc[client.Object, mcreconcile.Request] {
	return func(cn multicluster.ClusterName, _ cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] {
		_, lc := splitCluster(string(cn))
		return handler.TypedEnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []mcreconcile.Request {
			workspace, err := m.path(ctx, lc)
			if err != nil {
				m.log.Warn("a watched object is not mapped: its workspace path does not resolve", "export", export, "cluster", lc,
					"name", obj.GetName(), "err", err.Error())
				return nil
			}
			src := Request{Cluster: lc, Workspace: workspace, Namespace: obj.GetNamespace(), Name: obj.GetName()}
			var out []mcreconcile.Request
			for _, r := range mapFunc(ctx, src, obj) {
				cluster := lc
				if r.Cluster != "" {
					cluster = r.Cluster
				}
				target := multicluster.ClusterName(export + clusterSeparator + cluster)
				out = append(out, mcreconcile.Request{ClusterName: target,
					Request: reconcile.Request{NamespacedName: types.NamespacedName{Namespace: r.Namespace, Name: r.Name}}})
			}
			return out
		})
	}
}

func (m *Manager) eventSource(export string, ch <-chan Request) source.TypedSource[mcreconcile.Request] {
	return source.TypedFunc[mcreconcile.Request](func(ctx context.Context, q workqueue.TypedRateLimitingInterface[mcreconcile.Request]) error {
		m.sourcesWG.Go(func() {
			for {
				select {
				case <-ctx.Done():
					return
				case r, ok := <-ch:
					if !ok {
						return
					}
					q.Add(mcreconcile.Request{ClusterName: multicluster.ClusterName(export + clusterSeparator + r.Cluster),
						Request: reconcile.Request{NamespacedName: types.NamespacedName{Namespace: r.Namespace, Name: r.Name}}})
				}
			}
		})
		return nil
	})
}

func exportFilter(export string) mcbuilder.EngageOptions {
	prefix := export + clusterSeparator
	return mcbuilder.WithClusterFilter(func(cn multicluster.ClusterName, _ cluster.Cluster) bool {
		return strings.HasPrefix(string(cn), prefix)
	})
}

func specOrDeletion() predicate.Predicate {
	return predicate.Or(predicate.GenerationChangedPredicate{}, predicate.Funcs{
		UpdateFunc: func(e event.UpdateEvent) bool {
			if e.ObjectOld == nil || e.ObjectNew == nil {
				return false
			}
			return e.ObjectOld.GetDeletionTimestamp().IsZero() != e.ObjectNew.GetDeletionTimestamp().IsZero()
		},
	})
}

type adapter[T Object] struct {
	m *Manager

	name string

	r Reconciler[T]

	limiter *backoff

	log *slog.Logger

	only func(client.Object) bool

	inWorkspace func(context.Context, string) bool

	gone func(context.Context, Request)

	scope scope
}

func (a *adapter[T]) Reconcile(ctx context.Context, mcreq mcreconcile.Request) (reconcile.Result, error) {
	res, err := a.reconcile(ctx, mcreq)
	if err != nil && a.m.stopping(ctx) {
		a.log.Debug("the reconcile ended with the shutdown", "cluster", mcreq.ClusterName, "name", mcreq.Name, "err", err.Error())
		return reconcile.Result{}, nil
	}
	return res, err
}

func (a *adapter[T]) reconcile(ctx context.Context, mcreq mcreconcile.Request) (reconcile.Result, error) {
	_, lc := splitCluster(string(mcreq.ClusterName))
	cl, err := a.m.mgr.GetCluster(ctx, mcreq.ClusterName)
	if err != nil {
		return reconcile.Result{}, err
	}
	workspace, err := a.m.path(ctx, lc)
	if err != nil {
		return reconcile.Result{}, a.failed(ctx, mcreq, lc, mcreq.Name, err)
	}
	if a.inWorkspace != nil && !a.inWorkspace(ctx, workspace) {
		return a.outOfScope(mcreq, workspace), nil
	}
	obj, err := newObject[T]()
	if err != nil {
		return reconcile.Result{}, reconcile.TerminalError(err)
	}
	s := a.scope
	s.client = cl.GetClient()
	req := Request{Cluster: lc, Workspace: workspace, Namespace: mcreq.Namespace, Name: mcreq.Name, scope: &s}
	if err := cl.GetClient().Get(ctx, mcreq.NamespacedName, obj); err != nil {
		if apierrors.IsNotFound(err) {
			if a.gone != nil {
				a.gone(ctx, req)
			}
			return reconcile.Result{}, nil
		}
		return reconcile.Result{}, a.failed(ctx, mcreq, lc, mcreq.Name, err)
	}
	if a.only != nil && !a.only(obj) {
		return reconcile.Result{}, nil
	}
	log := a.log.With("cluster", lc, "workspace", req.Workspace, "name", req.Name)
	s.log = log
	if s.finalizer != "" && obj.GetDeletionTimestamp().IsZero() && !controllerutil.ContainsFinalizer(obj, s.finalizer) {
		if err := req.applyFinalizer(ctx, obj, true); err != nil {
			return reconcile.Result{}, a.failed(ctx, mcreq, lc, req.Name, err)
		}
	}
	gvk, err := apiutil.GVKForObject(obj, cl.GetScheme())
	if err != nil {
		return reconcile.Result{}, reconcile.TerminalError(err)
	}
	ctx = bus.WithCallerLabels(ctx, commonv1.ObjectRef{Group: gvk.Group, Kind: gvk.Kind, Name: obj.GetName(), Workspace: req.Workspace}, obj.GetLabels())
	log.Debug("reconciling")
	res, err := a.r.Reconcile(ctx, req, obj)
	if err != nil {
		return reconcile.Result{}, a.failed(ctx, mcreq, req.Workspace, req.Name, err)
	}
	if s.finalizer != "" && !res.Waiting && !obj.GetDeletionTimestamp().IsZero() && controllerutil.ContainsFinalizer(obj, s.finalizer) {
		if err := req.applyFinalizer(ctx, obj, false); err != nil {
			return reconcile.Result{}, a.failed(ctx, mcreq, req.Workspace, req.Name, err)
		}
	}
	if res.Backoff {
		if _, next := a.limiter.peek(mcreq); res.RequeueAfter <= 0 || res.RequeueAfter > next {
			return reconcile.Result{Requeue: true}, nil //nolint:staticcheck
		}
	}
	return reconcile.Result{RequeueAfter: res.RequeueAfter}, nil
}

func (a *adapter[T]) outOfScope(item mcreconcile.Request, workspace string) reconcile.Result {
	misses := a.limiter.NumRequeues(item)
	if misses >= scopeMissRetries {
		a.log.Debug("the workspace stays outside the controller's scope; the request is dropped", "workspace", workspace,
			"name", item.Name, "misses", misses)
		return reconcile.Result{}
	}
	a.log.Debug("the workspace is not in the controller's scope yet; it requeues with backoff", "workspace", workspace,
		"name", item.Name, "misses", misses+1)
	return reconcile.Result{Requeue: true} //nolint:staticcheck
}

func (a *adapter[T]) failed(ctx context.Context, item mcreconcile.Request, workspace, name string, err error) error {
	if a.m.stopping(ctx) {
		return err
	}
	failures, retryIn := a.limiter.peek(item)
	a.log.Warn("the reconcile failed; it retries with backoff", "workspace", workspace, "name", name,
		"failures", failures, "retryIn", retryIn.String(), "err", err.Error())
	return err
}

func newObject[T Object]() (T, error) {
	var zero T
	t := reflect.TypeOf(zero)
	if t == nil || t.Kind() != reflect.Pointer || t.Elem().Kind() != reflect.Struct {
		return zero, fmt.Errorf("runtime: %v is not a pointer to a struct", t)
	}
	obj, ok := reflect.New(t.Elem()).Interface().(T)
	if !ok {
		return zero, fmt.Errorf("runtime: %v does not implement the object interface", t)
	}
	return obj, nil
}