johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package exposure

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

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

	exposurev1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/exposure/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/component"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/exposure"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/binding"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/strs"
)

const (
	Name = "exposure"

	Export = "exposures"

	ConditionHandlerRegistered = "HandlerRegistered"

	ConditionReady = "Ready"

	ConditionIngressProvider = "IngressProvider"

	ServiceKind = "Service"

	IngressProxyKind = "IngressProxy"

	closedReportTimeout = 3 * time.Second
)

type Options struct {
	Publisher *exposure.Publisher

	Scope func(workspace string) bool

	Log *slog.Logger
}

type Controller struct {
	publisher *exposure.Publisher

	log *slog.Logger

	refs *binding.Table[struct{}]

	events *ctlruntime.Queue

	mu sync.Mutex

	subscribed map[string]ctlruntime.Request
}

func New(opts Options) (*Controller, error) {
	if opts.Publisher == nil {
		return nil, errors.New("exposure: a publisher is required")
	}
	log := opts.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	return &Controller{publisher: opts.Publisher, log: log.With("component", "exposure-controller"), refs: binding.New[struct{}](),
		events: ctlruntime.NewQueue(), subscribed: map[string]ctlruntime.Request{}}, nil
}

func Register(m *ctlruntime.Manager, opts Options) (*Controller, error) {
	if err := m.AddToScheme(exposurev1alpha1.AddToScheme); err != nil {
		return nil, err
	}
	ctl, err := New(opts)
	if err != nil {
		return nil, err
	}
	events, err := ctl.events.Run(m, Name+": components")
	if err != nil {
		return nil, fmt.Errorf("%s: forwarding component changes: %w", Name, err)
	}
	options := []ctlruntime.Option{ctlruntime.Export(Export), ctlruntime.Events(events), ctlruntime.Gone(ctl.gone),
		ctlruntime.ForeignStatus(exposure.SubscriberStatusPath),
		ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *exposurev1alpha1.Exposure) []ctlruntime.Request {
			if !obj.DeletionTimestamp.IsZero() || exposure.IngressRegisteredCurrent(obj) {
				return nil
			}
			return []ctlruntime.Request{src}
		}),
		ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *exposurev1alpha1.Service) []ctlruntime.Request {
			return ctl.users(binding.Key(src.Workspace, ServiceKind, obj.Name))
		}),
		ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj *exposurev1alpha1.ComponentLink) []ctlruntime.Request {
			to := obj.Spec.To
			return ctl.users(binding.Key(to.InWorkspace(src.Workspace), to.Kind, to.Name))
		})}
	if opts.Scope != nil {
		options = append(options, ctlruntime.InWorkspaces(opts.Scope))
	}
	if err := ctlruntime.For[*exposurev1alpha1.Exposure](m, Name, ctl, options...); err != nil {
		return nil, fmt.Errorf("exposure: registering the Exposure controller: %w", err)
	}
	return ctl, nil
}

func (c *Controller) Close() {
	c.publisher.Close()
	c.mu.Lock()
	subscribed := c.subscribed
	c.subscribed = map[string]ctlruntime.Request{}
	c.mu.Unlock()
	ctx, cancel := context.WithTimeout(context.Background(), closedReportTimeout)
	defer cancel()
	var wg sync.WaitGroup
	for _, req := range subscribed {
		state, ok := c.publisher.Subscriber(req.Workspace, req.Name)
		if !ok || state.State != exposurev1alpha1.IngressSubscriberClosed {
			continue
		}
		wg.Go(func() {
			if err := reportClosed(ctx, req, state); err != nil {
				c.log.Warn("reporting the closed ingress subscriber", "workspace", req.Workspace, "exposure", req.Name, "err", err)
			}
		})
	}
	wg.Wait()
}

func reportClosed(ctx context.Context, req ctlruntime.Request, state exposurev1alpha1.ExposureIngressSubscriber) error {
	if err := req.ApplyStatusAs(ctx, exposurev1alpha1.FieldManagerIngressSubscriber, exposureRef(req), exposure.SubscriberStatusFields(state)); err != nil {
		return err
	}
	var object exposurev1alpha1.Exposure
	if err := req.Get(ctx, "", req.Name, &object); err != nil {
		if apierrors.IsNotFound(err) {
			return nil
		}
		return err
	}
	status := *object.Status.DeepCopy()
	var route *exposurev1alpha1.ExposureIngress
	if status.Ingress != nil {
		status.Ingress.Subscriber = nil
		if status.Ingress.Host != "" || status.Ingress.URL != "" {
			route = status.Ingress
		} else {
			status.Ingress = nil
		}
	}
	exposure.SetIngressRegistered(&status.Conditions, object.Generation, route, &state)
	next := object.DeepCopy()
	next.Status = status
	return req.ApplyStatus(ctx, next)
}

func exposureRef(req ctlruntime.Request) *exposurev1alpha1.Exposure {
	return &exposurev1alpha1.Exposure{ObjectMeta: metav1.ObjectMeta{Name: req.Name}}
}

func (c *Controller) remember(req ctlruntime.Request, subscribed bool) {
	key := req.Workspace + "/" + req.Name
	c.mu.Lock()
	defer c.mu.Unlock()
	if subscribed {
		c.subscribed[key] = req
		return
	}
	delete(c.subscribed, key)
}

func (c *Controller) ComponentChanged(key component.Key) {
	seen := map[string]bool{}
	for _, ref := range []string{binding.Key(key.Workspace, key.Kind, key.Name), binding.Key("", key.Kind, key.Name)} {
		for _, r := range c.users(ref) {
			id := r.Cluster + "/" + r.Name
			if seen[id] {
				continue
			}
			seen[id] = true
			c.events.Add(r)
		}
	}
}

func (c *Controller) users(ref string) []ctlruntime.Request {
	var out []ctlruntime.Request
	for _, t := range c.refs.Users(ref) {
		out = append(out, ctlruntime.Request{Cluster: t.Cluster, Workspace: t.Workspace, Name: t.Name})
	}
	return out
}

func (c *Controller) gone(_ context.Context, req ctlruntime.Request) {
	c.remember(req, false)
	c.publisher.Withdraw(req.Workspace, req.Name)
	c.refs.Set(binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}, nil)
}

func (c *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, object *exposurev1alpha1.Exposure) (ctlruntime.Result, error) {
	self := binding.Target{Cluster: req.Cluster, Workspace: req.Workspace, Name: req.Name}
	if !object.DeletionTimestamp.IsZero() {
		c.remember(req, false)
		c.publisher.Withdraw(req.Workspace, req.Name)
		c.refs.Set(self, nil)
		return ctlruntime.Result{}, nil
	}
	spec := object.Spec
	serviceName := strs.LastSegment(spec.ServiceRef)
	var refs []string
	if serviceName != "" {
		refs = append(refs, binding.Key(req.Workspace, ServiceKind, serviceName))
	}
	if spec.Mode == exposurev1alpha1.ExposureModeIngress {
		ingress := cmp.Or(strings.TrimSpace(spec.IngressRef), exposure.DefaultIngressName)
		refs = append(refs, binding.Key(req.Workspace, IngressProxyKind, ingress), binding.Key("", IngressProxyKind, ingress))
	}
	var svc *exposurev1alpha1.Service
	if serviceName != "" {
		var found exposurev1alpha1.Service
		if err := req.Get(ctx, "", serviceName, &found); err == nil {
			svc = &found
		} else if !apierrors.IsNotFound(err) {
			c.refs.Set(self, &struct{}{}, refs...)
			return ctlruntime.Result{}, err
		}
	}
	var target component.Key
	var targetErr error
	if svc != nil {
		if target, targetErr = exposure.Target(svc.Spec.TargetRef, req.Workspace); targetErr == nil {
			refs = append(refs, binding.Key(target.Workspace, target.Kind, target.Name))
		}
	}
	c.refs.Set(self, &struct{}{}, refs...)
	switch {
	case spec.Suspend:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "Suspended", "the Exposure is suspended")
	case spec.Mode != exposurev1alpha1.ExposureModeLocal && spec.Mode != exposurev1alpha1.ExposureModeIngress:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposureFailed, "InvalidMode", fmt.Sprintf("unknown mode %q", spec.Mode))
	case serviceName == "":
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "ServiceMissing", "the Exposure has no serviceRef")
	case svc == nil:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "ServiceMissing",
			"the Service "+serviceName+" is not present in workspace "+req.Workspace)
	case svc.Spec.Suspend:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "ServiceSuspended", "the Service "+serviceName+" is suspended")
	case targetErr != nil:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposureFailed, "InvalidTarget", targetErr.Error())
	}
	_, found, err := c.publisher.Handler(ctx, target, svc.Spec.Protocol, svc.Spec.Handler)
	switch {
	case errors.Is(err, bus.ErrLinkDenied):
		res, applyErr := c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "LinkDenied", err.Error())
		return res, errors.Join(err, applyErr)
	case errors.Is(err, exposure.ErrUnknownProtocol):
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposureFailed, "InvalidProtocol", err.Error())
	case err != nil:
		res, applyErr := c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "HandlerUnreachable", err.Error())
		return res, errors.Join(err, applyErr)
	case !found:
		return c.withdraw(ctx, req, object, exposurev1alpha1.ExposurePending, "HandlerMissing",
			fmt.Sprintf("no handler %q is registered for %s/%s in workspace %s", exposure.HandlerName(svc.Spec.Handler), target.Kind, target.Name, target.Workspace))
	}
	published, err := c.publisher.Publish(ctx, exposure.Binding{Workspace: req.Workspace, Name: req.Name, Spec: spec, Service: svc.Spec,
		ServiceGeneration: svc.Generation, Target: target, SubscriberChanged: func() { c.events.Add(req) }})
	generation := object.Generation
	status := object.Status
	if err != nil {
		status.Phase = exposurev1alpha1.ExposurePending
		status.Ready = false
		status.Address = ""
		status.Endpoint = ""
		reason := "Pending"
		var retry error
		switch {
		case errors.Is(err, exposure.ErrStableHostMismatch):
			reason = "StableHostMismatch"
			status.Phase = exposurev1alpha1.ExposureFailed
		case errors.Is(err, exposure.ErrIngressNotReady):
			reason = "IngressNotReady"
		default:
			retry = err
		}
		conds.Set(&status.Conditions, generation, ConditionHandlerRegistered, true, "HandlerRegistered", "the target handler is registered")
		conds.Set(&status.Conditions, generation, ConditionReady, false, reason, err.Error())
		return ctlruntime.Result{}, errors.Join(retry, c.apply(ctx, req, object, status, nil))
	}
	status.Phase = exposurev1alpha1.ExposureReady
	status.Ready = true
	status.Address = published.Address
	status.Endpoint = published.Endpoint
	conds.Set(&status.Conditions, generation, ConditionHandlerRegistered, true, "HandlerRegistered", "the target handler is registered")
	message := "the Exposure is bound"
	if published.Provider != "" {
		conds.Set(&status.Conditions, generation, ConditionIngressProvider, true, "ProviderReady", published.Provider)
		message = "the Exposure is bound by the " + published.Provider + " ingress provider"
	}
	conds.Set(&status.Conditions, generation, ConditionReady, true, "Ready", message)
	return ctlruntime.Result{}, c.apply(ctx, req, object, status, published.Ingress)
}

func (c *Controller) withdraw(ctx context.Context, req ctlruntime.Request, object *exposurev1alpha1.Exposure, phase exposurev1alpha1.ExposurePhase,
	reason, message string) (ctlruntime.Result, error) {
	c.publisher.Pause(req.Workspace, req.Name)
	generation := object.Generation
	status := object.Status
	status.Phase = phase
	status.Ready = false
	status.Address = ""
	status.Endpoint = ""
	conds.Set(&status.Conditions, generation, ConditionHandlerRegistered, false, reason, message)
	conds.Set(&status.Conditions, generation, ConditionReady, false, reason, message)
	return ctlruntime.Result{}, c.apply(ctx, req, object, status, nil)
}

func (c *Controller) apply(ctx context.Context, req ctlruntime.Request, object *exposurev1alpha1.Exposure, status exposurev1alpha1.ExposureStatus,
	route *exposurev1alpha1.ExposureIngress) error {
	var reported *exposurev1alpha1.ExposureIngressSubscriber
	if object.Status.Ingress != nil {
		reported = object.Status.Ingress.Subscriber
	}
	sub := reported
	local, ok := c.publisher.Subscriber(req.Workspace, req.Name)
	if ok {
		sub = &local
	}
	c.remember(req, ok)
	if object.Spec.Mode != exposurev1alpha1.ExposureModeIngress {
		route, sub = nil, nil
	}
	status.Ingress = nil
	if route != nil {
		served := *route
		served.Subscriber = nil
		status.Ingress = &served
	}
	exposure.SetIngressRegistered(&status.Conditions, object.Generation, route, sub)
	next := object.DeepCopy()
	next.Status = status
	if err := req.ApplyStatus(ctx, next); err != nil {
		return err
	}
	if !ok || exposure.SubscriberStateEqual(reported, &local) {
		return nil
	}
	return req.ApplyStatusAs(ctx, exposurev1alpha1.FieldManagerIngressSubscriber, exposureRef(req), exposure.SubscriberStatusFields(local))
}