johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package cluster

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"sync"

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

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

const (
	Name = "cluster"

	Export = "clusters"

	LeaseExport = "applications"

	FrontHolder = "/front"
)

type Options struct {
	Ports local.Ports

	Front local.Front

	Changes <-chan struct{}

	Scope func(workspace string) bool
}

type Controller struct {
	ports local.Ports

	front local.Front

	mu sync.Mutex

	known map[ctlruntime.Request]bool
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	if opts.Ports == nil {
		return nil, errors.New("cluster: a port registry is required")
	}
	if opts.Front == nil {
		return nil, errors.New("cluster: an apiserver front is required")
	}
	if err := m.AddToScheme(v1alpha1.AddToScheme); err != nil {
		return nil, err
	}
	ctl := &Controller{ports: opts.Ports, front: opts.Front, known: map[ctlruntime.Request]bool{}}
	options := []ctlruntime.Option{ctlruntime.Export(Export), ctlruntime.SharedStatus(), ctlruntime.FieldManager(v1alpha1.FieldManagerClusterFront),
		ctlruntime.Gone(ctl.gone),
		ctlruntime.WatchesIn(LeaseExport, ctl.holderOf)}
	if opts.Scope != nil {
		options = append(options, ctlruntime.InWorkspaces(opts.Scope))
	}
	if opts.Changes != nil {
		events, err := ctlruntime.Forward(m, Name+": changes", opts.Changes, func(struct{}) []ctlruntime.Request { return ctl.all() })
		if err != nil {
			return nil, fmt.Errorf("cluster: forwarding changes: %w", err)
		}
		options = append(options, ctlruntime.Events(events))
	}
	if err := ctlruntime.For[*v1alpha1.Cluster](m, Name, ctl, options...); err != nil {
		return nil, fmt.Errorf("cluster: registering the Cluster controller: %w", err)
	}
	return ctl, nil
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *v1alpha1.Cluster) (ctlruntime.Result, error) {
	self := ctlruntime.Request{Cluster: req.Cluster, Namespace: req.Namespace, Name: req.Name}
	if !object.DeletionTimestamp.IsZero() {
		c.forget(self)
		return ctlruntime.Result{}, nil
	}
	c.remember(self)
	name := strings.TrimSpace(object.Name)
	registry := registryHeldPorts(c.ports.Leases(), req.Workspace, name)
	front := object.Status.Front
	if front == nil {
		front = &v1alpha1.ClusterFrontStatus{}
	}
	state, err := c.front.Status(ctx)
	if err != nil {
		var kept []v1alpha1.ClusterHeldPort
		for _, held := range front.HeldPorts {
			if held.HolderRef == name+FrontHolder {
				kept = append(kept, held)
			}
		}
		next := v1alpha1.ClusterFrontStatus{Address: front.Address, Message: "reading the apiserver front: " + err.Error(),
			HeldPorts: mergeHeldPorts(nil, name, kept, registry)}
		applyErr := c.apply(ctx, req, object, next)
		return ctlruntime.Result{}, errors.Join(fmt.Errorf("cluster: reading the apiserver front of cluster %s: %w", name, err), applyErr)
	}
	address := strings.TrimSpace(state.Address)
	next := v1alpha1.ClusterFrontStatus{Address: address, Ready: address != "", HeldPorts: mergeHeldPorts(state.Holds, name, registry)}
	if !next.Ready {
		next.Message = "the apiserver front reports no address"
	}
	return ctlruntime.Result{}, c.apply(ctx, req, object, next)
}

func (c *Controller) apply(ctx context.Context, req ctlruntime.Request, object *v1alpha1.Cluster, front v1alpha1.ClusterFrontStatus) error {
	if object.Status.Front != nil && reflect.DeepEqual(*object.Status.Front, front) {
		return nil
	}
	next := &v1alpha1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: object.Name, Namespace: object.Namespace},
		Status: v1alpha1.ClusterStatus{Front: &front}}
	return req.ApplyStatus(ctx, next)
}

func (c *Controller) gone(_ context.Context, req ctlruntime.Request) {
	c.forget(ctlruntime.Request{Cluster: req.Cluster, Namespace: req.Namespace, Name: req.Name})
}

func (c *Controller) remember(r ctlruntime.Request) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.known[r] = true
}

func (c *Controller) forget(r ctlruntime.Request) {
	c.mu.Lock()
	defer c.mu.Unlock()
	delete(c.known, r)
}

func (c *Controller) all() []ctlruntime.Request {
	c.mu.Lock()
	defer c.mu.Unlock()
	out := make([]ctlruntime.Request, 0, len(c.known))
	for r := range c.known {
		out = append(out, r)
	}
	return out
}

func (c *Controller) holderOf(_ context.Context, src ctlruntime.Request, lease *v1alpha1.PortLease) []ctlruntime.Request {
	ref := lease.Spec.ClusterRef
	name := strings.TrimSpace(ref.Name)
	if name == "" {
		return nil
	}
	if ws := strings.TrimSpace(ref.Workspace); ws != "" && ws != src.Workspace {
		return nil
	}
	target := ctlruntime.Request{Cluster: src.Cluster, Name: name}
	c.mu.Lock()
	defer c.mu.Unlock()
	if !c.known[target] {
		return nil
	}
	return []ctlruntime.Request{target}
}

func mergeHeldPorts(front map[string]int, cluster string, extra ...[]v1alpha1.ClusterHeldPort) []v1alpha1.ClusterHeldPort {
	byPort := make(map[int32]v1alpha1.ClusterHeldPort, len(front))
	for purpose, port := range front {
		if port <= 0 || port > 65535 {
			continue
		}
		byPort[int32(port)] = v1alpha1.ClusterHeldPort{Port: int32(port), Purpose: purpose, HolderRef: cluster + FrontHolder}
	}
	for _, group := range extra {
		for _, held := range group {
			if held.Port <= 0 || held.Port > 65535 {
				continue
			}
			if _, ok := byPort[held.Port]; !ok {
				byPort[held.Port] = held
			}
		}
	}
	if len(byPort) == 0 {
		return nil
	}
	out := make([]v1alpha1.ClusterHeldPort, 0, len(byPort))
	for _, held := range byPort {
		out = append(out, held)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Port < out[j].Port })
	return out
}

func registryHeldPorts(leases []ports.Lease, workspace, cluster string) []v1alpha1.ClusterHeldPort {
	out := make([]v1alpha1.ClusterHeldPort, 0, len(leases))
	for _, lease := range leases {
		if !local.HeldBy(lease.Name, workspace, cluster) || lease.Port <= 0 || lease.Port > 65535 {
			continue
		}
		purpose := lease.Name
		if i := strings.LastIndex(lease.Name, "/"); i >= 0 && i+1 < len(lease.Name) {
			purpose = lease.Name[i+1:]
		}
		out = append(out, v1alpha1.ClusterHeldPort{Port: int32(lease.Port), Purpose: purpose, HolderRef: lease.Name})
	}
	return out
}