johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package component

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"maps"
	"strings"
	"sync"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	svc "github.com/johnandersen777/socialweb-computer-kcp/internal/component"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
)

const (
	Name = "component"

	Export = "applications"

	eventBuffer = 1024
)

type Options struct {
	Runtime *svc.Runtime

	Scope func(workspace string) bool

	Changed []func(svc.Key)

	Log *slog.Logger
}

type key struct {
	kind string

	cluster string

	name string
}

type Controller struct {
	runtime *svc.Runtime

	changed []func(svc.Key)

	log *slog.Logger

	mu sync.Mutex

	uids map[key]types.UID

	keys map[types.UID]key

	applied map[types.UID]string

	events map[string]chan ctlruntime.Request
}

func New(opts Options) (*Controller, error) {
	if opts.Runtime == nil {
		return nil, errors.New("component: a component runtime is required")
	}
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	return &Controller{runtime: opts.Runtime, changed: opts.Changed, log: log.With("component", "component-controller"), uids: map[key]types.UID{},
		keys: map[types.UID]key{}, applied: map[types.UID]string{}, events: map[string]chan ctlruntime.Request{}}, nil
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	c, err := New(opts)
	if err != nil {
		return nil, err
	}
	c.runtime.SetStatusCallback(c.statusChanged)
	errs := []error{
		register(m, c, opts.Scope, "Bidder", func(o *v1alpha1.Bidder) *v1alpha1.ComponentStatus { return &o.Status.ComponentStatus }, bidderStatus),
		register(m, c, opts.Scope, "GuestControlPlane", func(o *v1alpha1.GuestControlPlane) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "Pds", func(o *v1alpha1.Pds) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "Requester", func(o *v1alpha1.Requester) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "GatewayAPI", func(o *v1alpha1.GatewayAPI) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "SshServer", func(o *v1alpha1.SshServer) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "Plc", func(o *v1alpha1.Plc) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "Relay", func(o *v1alpha1.Relay) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "IngressProxy", func(o *v1alpha1.IngressProxy) *v1alpha1.ComponentStatus { return &o.Status }),
		register(m, c, opts.Scope, "DispatcherTLS", func(o *v1alpha1.DispatcherTLS) *v1alpha1.ComponentStatus { return &o.Status }),
	}
	if err := errors.Join(errs...); err != nil {
		return nil, err
	}
	return c, nil
}

func register[T interface {
	ctlruntime.Object
	DeepCopy() T
}](m *ctlruntime.Manager, c *Controller, scope func(string) bool, kind string, status func(T) *v1alpha1.ComponentStatus, kindStatus ...func(T)) error {
	events := make(chan ctlruntime.Request, eventBuffer)
	c.mu.Lock()
	c.events[kind] = events
	c.mu.Unlock()
	r := &reconciler[T]{c: c, kind: kind, status: status, kindStatus: kindStatus}
	opts := []ctlruntime.Option{ctlruntime.Export(Export), ctlruntime.Events(events), ctlruntime.Gone(r.gone)}
	if scope != nil {
		opts = append(opts, ctlruntime.InWorkspaces(scope))
	}
	if err := ctlruntime.For[T](m, Name+"-"+strings.ToLower(kind), r, opts...); err != nil {
		return fmt.Errorf("component: registering the %s controller: %w", kind, err)
	}
	return nil
}

type reconciler[T interface {
	ctlruntime.Object
	DeepCopy() T
}] struct {
	c *Controller

	kind string

	status func(T) *v1alpha1.ComponentStatus

	kindStatus []func(T)
}

func bidderStatus(o *v1alpha1.Bidder) {
	st := o.Spec.GuestNetwork.EffectiveFor(o.Spec.Policy)
	o.Status.GuestNetwork = &st
}

func (r *reconciler[T]) Reconcile(ctx context.Context, req ctlruntime.Request, obj T) (ctlruntime.Result, error) {
	spec, err := specOf(r.kind, req, obj)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	k := key{kind: r.kind, cluster: req.Cluster, name: req.Name}
	r.c.remember(k, spec.UID)
	st, err := r.c.runtime.Reconcile(ctx, spec)
	if errors.Is(err, svc.ErrRuntimeNotStarted) {
		return ctlruntime.Result{}, err
	}
	if err != nil {
		r.c.log.Warn("the component did not start; it waits for a change of its spec", "kind", r.kind, "workspace", req.Workspace,
			"name", req.Name, "err", err.Error())
	}
	if current, ok := r.c.runtime.Status(spec.UID); ok {
		st = current
	}
	if spec.Deleting {
		r.c.forget(k)
		return ctlruntime.Result{}, nil
	}
	next := obj.DeepCopy()
	*r.status(next) = st.APIStatus()
	for _, fn := range r.kindStatus {
		fn(next)
	}
	if err := req.ApplyStatus(ctx, next); err != nil {
		return ctlruntime.Result{}, err
	}
	r.c.mu.Lock()
	r.c.applied[spec.UID] = fingerprint(st)
	r.c.mu.Unlock()
	return ctlruntime.Result{}, nil
}

func (r *reconciler[T]) gone(ctx context.Context, req ctlruntime.Request) {
	k := key{kind: r.kind, cluster: req.Cluster, name: req.Name}
	uid, ok := r.c.forget(k)
	if !ok {
		return
	}
	if err := r.c.runtime.Remove(ctx, uid); err != nil {
		r.c.log.Warn("stopping a deleted component", "kind", r.kind, "workspace", req.Workspace, "name", req.Name, "err", err.Error())
	}
}

func (c *Controller) remember(k key, uid types.UID) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if old, ok := c.uids[k]; ok && old != uid {
		delete(c.keys, old)
	}
	c.uids[k], c.keys[uid] = uid, k
}

func (c *Controller) forget(k key) (types.UID, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	uid, ok := c.uids[k]
	delete(c.uids, k)
	if ok {
		delete(c.keys, uid)
		delete(c.applied, uid)
	}
	return uid, ok
}

func (c *Controller) statusChanged(ctx context.Context, spec svc.ComponentSpec, status svc.ComponentStatus) error {
	for _, fn := range c.changed {
		fn(svc.Key{Workspace: spec.Workspace, Kind: spec.Kind, Name: spec.Name})
	}
	c.mu.Lock()
	k, ok := c.keys[spec.UID]
	events := c.events[spec.Kind]
	same := c.applied[spec.UID] == fingerprint(status)
	c.mu.Unlock()
	if !ok || events == nil || same {
		return nil
	}
	select {
	case events <- ctlruntime.Request{Cluster: k.cluster, Namespace: spec.Namespace, Name: k.name}:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func fingerprint(status svc.ComponentStatus) string {
	st := status.APIStatus()
	for i := range st.Conditions {
		st.Conditions[i].LastTransitionTime = metav1.Time{}
	}
	raw, err := json.Marshal(st)
	if err != nil {
		return ""
	}
	return string(raw)
}

func specOf[T ctlruntime.Object](kind string, req ctlruntime.Request, obj T) (svc.ComponentSpec, error) {
	spec := svc.ComponentSpec{
		UID:         obj.GetUID(),
		Group:       v1alpha1.GroupName,
		Kind:        kind,
		Name:        obj.GetName(),
		Namespace:   obj.GetNamespace(),
		Workspace:   req.Workspace,
		Generation:  obj.GetGeneration(),
		Deleting:    !obj.GetDeletionTimestamp().IsZero(),
		Annotations: maps.Clone(obj.GetAnnotations()),
		Labels:      maps.Clone(obj.GetLabels()),
	}
	raw, err := json.Marshal(obj)
	if err != nil {
		return spec, fmt.Errorf("component: encoding %s/%s: %w", kind, spec.Name, err)
	}
	var doc struct {
		Spec json.RawMessage `json:"spec"`
	}
	if err := json.Unmarshal(raw, &doc); err != nil {
		return spec, fmt.Errorf("component: the spec of %s/%s: %w", kind, spec.Name, err)
	}
	if len(doc.Spec) == 0 || string(doc.Spec) == "null" {
		return spec, nil
	}
	var common struct {
		Suspend       bool   `json:"suspend"`
		RestartPolicy string `json:"restartPolicy"`
		StateDir      string `json:"stateDir"`
	}
	if err := json.Unmarshal(doc.Spec, &common); err != nil {
		return spec, fmt.Errorf("component: the spec of %s/%s: %w", kind, spec.Name, err)
	}
	spec.Suspend, spec.RestartPolicy, spec.StateDir = common.Suspend, svc.RestartPolicy(common.RestartPolicy), common.StateDir
	spec.Spec = svc.RawSpec(doc.Spec)
	return spec, nil
}