johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package compose

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

	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/rest"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	pdsv1alpha1 "github.com/johnandersen777/socialweb-computer-kcp/api/pds/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver"
)

const (
	ManifestSet        = "applications"
	Finalizer          = "apps.socialweb.computer/components"
	ComponentWorkspace = "app"
	DefaultWorkspace   = "root"
	ApplyTimeout       = time.Minute
	ComponentsDir      = "components"
	AppsDir            = "apps"
)

type Config struct {
	Kubeconfig string

	Workspace string

	Layout devinstance.Layout

	Manager hostsvc.Manager

	Record func() (devinstance.Record, error)

	ReadyTimeout time.Duration

	Listeners func(owner string) []v1alpha1.ApplicationListener

	OnComponents func(owner string, listeners []v1alpha1.ApplicationListener)

	StoreUsage func() *v1alpha1.StoreUsage

	TTLReport func() *v1alpha1.TTLStatus

	Kube *kube.Client

	Bus *bus.Bus

	RemotePDS *pdsv1alpha1.RemotePDS

	Rest *rest.Config

	Log *slog.Logger
}

type Composer struct {
	cfg Config

	restCfg *rest.Config

	log *slog.Logger

	mu sync.Mutex

	clients map[string]*kube.Client

	applied map[string]int64

	forgotten map[string]bool

	samples samples

	adoptions adoptions

	parking parking

	roots rootIndex

	drift map[string]*driftState

	trusts Trusts

	requesterSets sync.Map

	requesters requesterIndex

	applies applyCache

	linkSets linkSets
}

func New(cfg Config) (*Composer, error) {
	var restCfg *rest.Config
	if cfg.Rest != nil {
		restCfg = rest.CopyConfig(cfg.Rest)
		if restCfg.QPS == 0 {
			restCfg.QPS, restCfg.Burst = kcpserver.ClientQPS, kcpserver.ClientBurst
		}
	} else if cfg.Kube == nil {
		var err error
		restCfg, err = kcpserver.FrontRestConfig(cfg.Kubeconfig)
		if err != nil {
			return nil, err
		}
	}
	log := cfg.Log
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	return &Composer{cfg: cfg, restCfg: restCfg, log: log.With("component", "app-compose"), clients: map[string]*kube.Client{},
		applied: map[string]int64{}, forgotten: map[string]bool{}, roots: rootIndex{byCluster: map[string]map[string]map[string]bool{}}}, nil
}

func (c *Composer) Config() Config { return c.cfg }

func (c *Composer) Log() *slog.Logger { return c.log }

func (c *Composer) Bus() *bus.Bus { return c.cfg.Bus }

func (c *Composer) ProvideTrust(gk schema.GroupKind) { c.trusts.Provide(c.cfg.Bus, gk) }

func (c *Composer) PublishTrust(ref commonv1.ObjectRef, bundle []byte) { c.trusts.Publish(ref, bundle) }

func (c *Composer) Workspace() string {
	return cmp.Or(strings.TrimSpace(c.cfg.Workspace), DefaultWorkspace)
}

func (c *Composer) StateRoot() string { return c.cfg.Layout.StateDir }

func (c *Composer) RestConfig() *rest.Config { return c.restCfg }

func (c *Composer) KcpServer() string {
	if c.restCfg == nil {
		return ""
	}
	server, _, _ := strings.Cut(c.restCfg.Host, "/clusters/")
	return server
}

func (c *Composer) Kube(ws string) (*kube.Client, error) {
	if c.cfg.Kube != nil {
		return c.cfg.Kube, nil
	}
	ws = cmp.Or(strings.TrimSpace(ws), c.Workspace())
	c.mu.Lock()
	kc, ok := c.clients[ws]
	c.mu.Unlock()
	if !ok {
		var err error
		kc, err = kube.ForWorkspace(c.restCfg, ws)
		if err != nil {
			return nil, err
		}
		c.mu.Lock()
		c.clients[ws] = kc
		c.mu.Unlock()
	}
	return kc, nil
}

func (c *Composer) Client(ws string, gvr schema.GroupVersionResource) (kube.Resource, error) {
	kc, err := c.Kube(ws)
	if err != nil {
		return kube.Resource{}, err
	}
	return kc.Resource(gvr), nil
}

func (c *Composer) Applied(r Root) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.applied[string(r.UID)] == r.Generation
}

func (c *Composer) MarkApplied(r Root) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.applied[string(r.UID)] = r.Generation
	delete(c.forgotten, string(r.UID))
}

func (c *Composer) AppliedBefore(r Root, recorded *v1alpha1.AppliedSet, digest string) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return !c.forgotten[string(r.UID)] && recorded != nil && digest != "" &&
		recorded.Generation == r.Generation && recorded.Digest == digest
}

func (c *Composer) Forget(r Root) {
	c.mu.Lock()
	delete(c.applied, string(r.UID))
	c.forgotten[string(r.UID)] = true
	c.mu.Unlock()
	c.adoptions.forget(r)
}

func ComponentStateDir(instanceState, component string) string {
	return filepath.Join(instanceState, ComponentsDir, component)
}

func ComponentCRStateDir(instanceState, kind, name string) string {
	return filepath.Join(instanceState, ComponentsDir, kind+"-"+name)
}

func failf(format string, args ...any) error {
	return fmt.Errorf("compose: "+format, args...)
}

type contextKey struct{}

func WithRoot(ctx context.Context, r Root) context.Context {
	return context.WithValue(ctx, contextKey{}, r)
}

func RootFrom(ctx context.Context) (Root, bool) {
	r, ok := ctx.Value(contextKey{}).(Root)
	return r, ok
}