johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package openbaoserver

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"

	"k8s.io/apimachinery/pkg/api/equality"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/openbaoctl"
)

const (
	Name = "openbaoserver"

	Export = "hostprocesses"

	GrantExport = "computevms"

	GrantEventBuffer = 1024
)

type Grants interface {
	List(ctx context.Context, workspace string) ([]v1alpha1.GuestIdentityGrant, error)
}

type Options struct {
	Servers *openbaoctl.Servers

	Grants Grants

	Scope func(workspace string) bool

	Log *slog.Logger
}

type grantKey struct {
	cluster string

	name string
}

type Controller struct {
	log *slog.Logger

	servers *openbaoctl.Servers

	grants Grants

	events chan ctlruntime.Request

	mu sync.Mutex

	identity map[string]map[string]struct{}

	outcomes map[grantKey]v1alpha1.GuestIdentityGrantStatus
}

func New(opts Options) (*Controller, error) {
	if opts.Servers == nil {
		return nil, errors.New("openbaoserver: the OpenBao servers backend is required")
	}
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	return &Controller{log: log.With("component", "openbaoserver-controller"), servers: opts.Servers, grants: opts.Grants,
		events: make(chan ctlruntime.Request, GrantEventBuffer), identity: map[string]map[string]struct{}{},
		outcomes: map[grantKey]v1alpha1.GuestIdentityGrantStatus{}}, nil
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	ctl, err := New(opts)
	if err != nil {
		return nil, err
	}
	options := []ctlruntime.Option{ctlruntime.Export(Export), ctlruntime.Owns[*v1alpha1.HostProcess](), ctlruntime.Gone(ctl.gone),
		ctlruntime.WatchesIn(GrantExport, func(_ context.Context, src ctlruntime.Request, _ *v1alpha1.GuestIdentityGrant) []ctlruntime.Request {
			return ctl.identityServers(src.Cluster)
		})}
	if opts.Scope != nil {
		options = append(options, ctlruntime.InWorkspaces(opts.Scope))
	}
	if err := ctlruntime.For[*v1alpha1.OpenBaoServer](m, Name, ctl, options...); err != nil {
		return nil, fmt.Errorf("openbaoserver: registering the OpenBaoServer controller: %w", err)
	}
	return ctl, nil
}

func (c *Controller) GrantEvents() <-chan ctlruntime.Request { return c.events }

func (c *Controller) Outcome(cluster, name string) (v1alpha1.GuestIdentityGrantStatus, bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	st, ok := c.outcomes[grantKey{cluster: cluster, name: name}]
	return st, ok
}

func (c *Controller) Forget(cluster, name string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	delete(c.outcomes, grantKey{cluster: cluster, name: name})
}

func (c *Controller) identityServers(cluster string) []ctlruntime.Request {
	c.mu.Lock()
	defer c.mu.Unlock()
	out := make([]ctlruntime.Request, 0, len(c.identity[cluster]))
	for name := range c.identity[cluster] {
		out = append(out, ctlruntime.Request{Cluster: cluster, Name: name})
	}
	return out
}

func (c *Controller) index(cluster, name string, identity bool) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if !identity {
		delete(c.identity[cluster], name)
		if len(c.identity[cluster]) == 0 {
			delete(c.identity, cluster)
		}
		return
	}
	if c.identity[cluster] == nil {
		c.identity[cluster] = map[string]struct{}{}
	}
	c.identity[cluster][name] = struct{}{}
}

func (c *Controller) record(cluster string, outcomes map[string]v1alpha1.GuestIdentityGrantStatus) []string {
	c.mu.Lock()
	defer c.mu.Unlock()
	var changed []string
	for name, st := range outcomes {
		k := grantKey{cluster: cluster, name: name}
		if held, ok := c.outcomes[k]; ok && equality.Semantic.DeepEqual(held, st) {
			continue
		}
		c.outcomes[k] = st
		changed = append(changed, name)
	}
	return changed
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *v1alpha1.OpenBaoServer) (ctlruntime.Result, error) {
	if !object.DeletionTimestamp.IsZero() {
		return ctlruntime.Result{}, nil
	}
	identity := openbaoctl.NeedsGrants(*object)
	c.index(req.Cluster, object.Name, identity)
	var grants []v1alpha1.GuestIdentityGrant
	if identity {
		if c.grants == nil {
			return ctlruntime.Result{}, c.apply(ctx, req, object, failed(object, "this core reads no GuestIdentityGrants, so it cannot hold spec.declared.identityOIDC"))
		}
		list, err := c.grants.List(ctx, req.Workspace)
		if err != nil && !apierrors.IsNotFound(err) {
			return ctlruntime.Result{}, errors.Join(fmt.Errorf("openbaoserver: listing the GuestIdentityGrants of %s: %w", req.Workspace, err),
				c.apply(ctx, req, object, failed(object, fmt.Sprintf("listing the GuestIdentityGrants: %v", err))))
		}
		grants = list
	}
	pctx, cancel := context.WithTimeout(ctx, openbaoctl.PassTimeout)
	out := c.servers.Reconcile(pctx, *object, grants, children{req: req, owner: object})
	cancel()
	if err := c.apply(ctx, req, object, out.Status); err != nil {
		return ctlruntime.Result{}, errors.Join(out.Err, err)
	}
	if changed := c.record(req.Cluster, out.Grants); len(changed) > 0 {
		for _, name := range changed {
			select {
			case c.events <- ctlruntime.Request{Cluster: req.Cluster, Name: name}:
			case <-ctx.Done():
				return ctlruntime.Result{}, ctx.Err()
			}
		}
	}
	if out.Err != nil {
		return ctlruntime.Result{}, fmt.Errorf("openbaoserver: %s: %w", object.Name, out.Err)
	}
	c.log.Debug("the OpenBaoServer is reconciled", "cluster", req.Cluster, "workspace", req.Workspace, "name", req.Name,
		"phase", out.Status.Phase, "requeueAfter", out.RequeueAfter.String())
	return ctlruntime.Result{RequeueAfter: out.RequeueAfter}, nil
}

func failed(object *v1alpha1.OpenBaoServer, message string) v1alpha1.OpenBaoServerStatus {
	st := *object.Status.DeepCopy()
	st.ObservedGeneration, st.Phase, st.Message, st.Ready = object.Generation, v1alpha1.OpenBaoServerPhaseFailed, message, false
	conds.Set(&st.Conditions, object.Generation, openbaoctl.ConditionReady, false, "Failed", message)
	return st
}

func (c *Controller) apply(ctx context.Context, req ctlruntime.Request, object *v1alpha1.OpenBaoServer, status v1alpha1.OpenBaoServerStatus) error {
	next := object.DeepCopy()
	next.Status = status
	return req.ApplyStatus(ctx, next)
}

func (c *Controller) gone(ctx context.Context, req ctlruntime.Request) {
	c.index(req.Cluster, req.Name, false)
	name := openbaoctl.ChildName(req.Name)
	var child v1alpha1.HostProcess
	if err := req.Get(ctx, "", name, &child); err != nil {
		return
	}
	if child.Labels[openbaoctl.OwnerLabel] != req.Name {
		return
	}
	if err := req.Delete(ctx, &child); err != nil {
		c.log.Warn("deleting the HostProcess of a gone OpenBaoServer", "workspace", req.Workspace, "openbaoserver", req.Name,
			"hostprocess", name, "err", err.Error())
		return
	}
	c.log.Info("the OpenBaoServer is gone; its HostProcess is deleted and the supervisor stops the unit; the storage stays",
		"workspace", req.Workspace, "openbaoserver", req.Name, "hostprocess", name)
}

var _ openbaoctl.Children = children{}

type children struct {
	req ctlruntime.Request

	owner *v1alpha1.OpenBaoServer
}

func (c children) Ensure(ctx context.Context, name string, spec v1alpha1.HostProcessSpec) (v1alpha1.HostProcess, error) {
	var cur v1alpha1.HostProcess
	err := c.req.Get(ctx, "", name, &cur)
	switch {
	case err == nil:
		if cur.Labels[openbaoctl.OwnerLabel] != c.owner.Name {
			return cur, fmt.Errorf("HostProcess %s exists and is not owned by OpenBaoServer %s", name, c.owner.Name)
		}
	case !apierrors.IsNotFound(err):
		return cur, err
	}
	want := &v1alpha1.HostProcess{ObjectMeta: metav1.ObjectMeta{Name: name, Labels: map[string]string{openbaoctl.OwnerLabel: c.owner.Name},
		OwnerReferences: []metav1.OwnerReference{{APIVersion: v1alpha1.GroupVersion.String(), Kind: "OpenBaoServer", Name: c.owner.Name,
			UID: c.owner.UID, Controller: new(true)}}}, Spec: spec}
	if err := c.req.Apply(ctx, want); err != nil {
		return cur, err
	}
	if err != nil {
		return *want, nil
	}
	cur.Spec = spec
	return cur, nil
}

func (c children) Delete(ctx context.Context, name string) error {
	return c.req.Delete(ctx, &v1alpha1.HostProcess{ObjectMeta: metav1.ObjectMeta{Name: name}})
}