johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package devinstance

import (
	"context"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"sync"
	"time"

	apisv1alpha2 "github.com/kcp-dev/sdk/apis/apis/v1alpha2"
	tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"

	"github.com/johnandersen777/socialweb-computer-kcp/api/kcpgvr"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap/table"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kcpapply"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostprocess"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/clock"
)

const (
	HostID              = "dev-host"
	HostSet             = "dev-host"
	DevWorkspace        = "root:dev"
	ChildWorkspaceType  = "dev-instance"
	HostInterval        = time.Second
	HostApplyTimeout    = time.Minute
	ChildWorkspaceReady = 30 * time.Second
	ChildWorkspaceGone  = 2 * time.Minute
)

const hostProcessExport = "hostprocesses"

var sweptKinds = []string{hostsvc.NamePrefix + "kcp-", hostsvc.NamePrefix + "core-", hostsvc.NamePrefix + "hp-"}

type Parent struct {
	ID         string `json:"id"`
	Socket     string `json:"socket"`
	Kubeconfig string `json:"kubeconfig"`
	Workspace  string `json:"workspace"`
}

func ChildWorkspace(id string) string { return DevWorkspace + ":" + id }

func Prefixes(id string) []string {
	n := table.DevNames(Short(id))
	return []string{n.UnitPrefix(), n.CorePrefix(), n.HostProcessPrefix()}
}

func clusterClient(kubeconfig, workspace string) (*kube.Client, error) {
	restCfg, err := kcpserver.FrontRestConfig(kubeconfig)
	if err != nil {
		return nil, err
	}
	cfg := rest.CopyConfig(restCfg)
	cfg.Host = strings.TrimSuffix(cfg.Host, "/") + "/clusters/" + workspace
	return kube.New(cfg)
}

func workspacePhase(w *tenancyv1alpha1.Workspace) string {
	return string(w.Status.Phase)
}

func WaitHost(ctx context.Context, kubeconfig string, timeout time.Duration) error {
	parent, name, _ := strings.Cut(DevWorkspace, ":")
	root, err := clusterClient(kubeconfig, parent)
	if err != nil {
		return err
	}
	dc, err := clusterClient(kubeconfig, DevWorkspace)
	if err != nil {
		return err
	}
	_, err = kube.Await(ctx, root.Resource(kcpgvr.Workspaces), name, timeout, func(w *tenancyv1alpha1.Workspace) (bool, error) {
		return w != nil && workspacePhase(w) == "Ready", nil
	})
	if err == nil {
		_, err = kube.List[tenancyv1alpha1.Workspace](ctx, dc.Resource(kcpgvr.Workspaces), metav1.ListOptions{Limit: 1})
	}
	if err != nil {
		return fmt.Errorf("devinstance: the dev host's %s did not serve workspaces within %s: %w", DevWorkspace, timeout, err)
	}
	return nil
}

func awaitChildGone(ctx context.Context, ws kube.Resource, id string, log *slog.Logger) error {
	started := time.Now()
	waited := false
	_, err := kube.Await(ctx, ws, id, ChildWorkspaceGone, func(w *tenancyv1alpha1.Workspace) (bool, error) {
		if w == nil || w.DeletionTimestamp == nil {
			return true, nil
		}
		if !waited {
			waited = true
			log.Info("the child workspace is being deleted in the parent; waiting for it to go before placing it again", "workspace", ChildWorkspace(id))
		}
		return false, nil
	})
	if err != nil {
		return fmt.Errorf("devinstance: workspace %s was being deleted in the parent and did not go within %s: %w", ChildWorkspace(id), ChildWorkspaceGone, err)
	}
	if waited {
		log.Info("the deleted child workspace is gone; placing it again", "workspace", ChildWorkspace(id), "seconds", time.Since(started).Seconds())
	}
	return nil
}

func EnsureChildWorkspace(ctx context.Context, kubeconfig, id string, timeout time.Duration, log *slog.Logger) error {
	if err := WaitHost(ctx, kubeconfig, timeout); err != nil {
		return err
	}
	dc, err := clusterClient(kubeconfig, DevWorkspace)
	if err != nil {
		return err
	}
	ws := dc.Resource(kcpgvr.Workspaces)
	if err := awaitChildGone(ctx, ws, id, log); err != nil {
		return err
	}
	o := kube.Doc{
		"apiVersion": "tenancy.kcp.io/v1alpha1", "kind": "Workspace",
		"metadata": map[string]any{"name": id, "annotations": map[string]any{kcpserver.WorkspaceOwnerAnnotation: kcpserver.ShardAdminOwner}},
		"spec":     map[string]any{"type": map[string]any{"name": ChildWorkspaceType, "path": "root"}},
	}
	if _, err := kube.Create[kube.Doc](ctx, ws, o, metav1.CreateOptions{}); err != nil && !apierrors.IsAlreadyExists(err) {
		return fmt.Errorf("devinstance: creating workspace %s in the parent: %w", ChildWorkspace(id), err)
	}
	child, err := clusterClient(kubeconfig, ChildWorkspace(id))
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	cur, err := kube.Await(ctx, ws, id, 0, func(w *tenancyv1alpha1.Workspace) (bool, error) {
		if w != nil && w.DeletionTimestamp != nil {
			return false, fmt.Errorf("devinstance: workspace %s is being deleted in the parent; wait for it to go, then dev up again", ChildWorkspace(id))
		}
		return w != nil && workspacePhase(w) == "Ready", nil
	})
	phase := ""
	if cur != nil {
		phase = workspacePhase(cur)
	}
	if err == nil {
		err = awaitBound(ctx, child.Resource(kcpgvr.APIBindings), hostProcessExport)
	}
	if err == nil {
		_, err = kube.List[v1alpha1.HostProcess](ctx, child.Resource(hostprocess.GVR), metav1.ListOptions{Limit: 1})
	}
	if err != nil {
		return fmt.Errorf("devinstance: workspace %s did not become Ready with the HostProcess API bound within %s (phase %q): %w", ChildWorkspace(id), timeout, phase, err)
	}
	return nil
}

func awaitBound(ctx context.Context, bindings kube.Resource, export string) error {
	_, err := kube.AwaitList(ctx, bindings, metav1.ListOptions{}, 0, func(items []apisv1alpha2.APIBinding) (bool, error) {
		for _, b := range items {
			ref := b.Spec.Reference.Export
			if ref != nil && ref.Name == export && b.Status.Phase == apisv1alpha2.APIBindingPhaseBound {
				return true, nil
			}
		}
		return false, nil
	})
	return err
}

func DeleteChildWorkspace(ctx context.Context, kubeconfig, id string) error {
	dc, err := clusterClient(kubeconfig, DevWorkspace)
	if err != nil {
		return err
	}
	err = dc.Resource(kcpgvr.Workspaces).Delete(ctx, id, metav1.DeleteOptions{})
	if err != nil && !apierrors.IsNotFound(err) {
		return fmt.Errorf("devinstance: deleting workspace %s in the parent: %w", ChildWorkspace(id), err)
	}
	return nil
}

func ChildUnits(ctx context.Context, m hostsvc.Manager, id string) ([]hostsvc.UnitStatus, error) {
	var out []hostsvc.UnitStatus
	for _, p := range Prefixes(id) {
		units, err := m.List(ctx, p)
		if err != nil {
			return nil, err
		}
		out = append(out, units...)
	}
	return out, nil
}

func OpenRemote(ctx context.Context, p Parent, log *slog.Logger) (*hostprocess.Remote, error) {
	reader, err := hostsvc.Open(ctx, hostsvc.Config{Backend: hostsvc.BackendProc, ProcSocket: p.Socket})
	if err != nil {
		return nil, fmt.Errorf("devinstance: the parent %s devsup on %s: %w", p.ID, p.Socket, err)
	}
	r, err := hostprocess.NewRemote(hostprocess.RemoteConfig{Kubeconfig: p.Kubeconfig, Workspace: p.Workspace, Reader: reader, Log: log})
	if err != nil {
		reader.Close()
		return nil, err
	}
	return r, nil
}

type HostConfig struct {
	Kubeconfig string
	Manager    hostsvc.Manager
	Own        []string
	Resolve    func(v1alpha1.HostProcessAsset) (string, error)
	Log        *slog.Logger
}

type host struct {
	c        HostConfig
	store    *kube.Cache[tenancyv1alpha1.Workspace]
	children map[string]context.CancelFunc
	wg       sync.WaitGroup
}

func applyHostSet(ctx context.Context, c HostConfig) error {
	restCfg, err := kcpserver.FrontRestConfig(c.Kubeconfig)
	if err != nil {
		return err
	}
	ap, err := kcpapply.New(restCfg, kcpapply.Options{WaitTimeout: HostApplyTimeout, Log: c.Log})
	if err != nil {
		return err
	}
	ix, err := manifests.Load()
	if err != nil {
		return err
	}
	set, ok := ix.Set(HostSet)
	if !ok {
		return fmt.Errorf("devinstance: deploy/sets.yaml names no %s set", HostSet)
	}
	values, err := manifests.DecodeValues(set.Values, nil)
	if err != nil {
		return err
	}
	_, err = ap.Apply(ctx, set, values)
	return err
}

func RunHost(ctx context.Context, c HostConfig) {
	log := c.Log.With("component", "dev-host")
	for {
		err := applyHostSet(ctx, c)
		if err == nil {
			break
		}
		if ctx.Err() != nil {
			return
		}
		log.Warn("applying the dev-host set; retrying", "err", err.Error())
		select {
		case <-ctx.Done():
			return
		case <-time.After(hostprocess.RetryAfter):
		}
	}
	dc, err := clusterClient(c.Kubeconfig, DevWorkspace)
	if err != nil {
		log.Error("the root:dev client", "err", err.Error())
		return
	}
	kick := make(chan struct{}, 1)
	nudge := func() {
		select {
		case kick <- struct{}{}:
		default:
		}
	}
	informer, err := kube.NewCache[tenancyv1alpha1.Workspace](dc.Resource(kcpgvr.Workspaces), metav1.ListOptions{}, nudge)
	if err != nil {
		log.Error("the root:dev workspace watch", "err", err.Error())
		return
	}
	h := &host{c: c, store: informer, children: map[string]context.CancelFunc{}}
	var watches sync.WaitGroup
	wctx, stopWatches := context.WithCancel(ctx)
	watches.Go(func() { informer.Run(wctx) })
	for _, k := range sweptKinds {
		watches.Go(func() { followUnits(wctx, c.Manager, k, nudge, log) })
	}
	defer func() {
		stopWatches()
		watches.Wait()
	}()
	if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {
		return
	}
	log.Info("the dev host places its children's HostProcess objects", "workspace", DevWorkspace)
	retry := time.NewTimer(0)
	defer retry.Stop()
	for {
		select {
		case <-ctx.Done():
			for _, cancel := range h.children {
				cancel()
			}
			h.wg.Wait()
			return
		case <-kick:
		case <-retry.C:
		}
		if err := h.pass(ctx, log); err != nil {
			log.Warn("a dev-host pass failed; children keep running", "err", err.Error())
			retry.Reset(HostInterval)
		}
	}
}

func followUnits(ctx context.Context, m hostsvc.Manager, prefix string, changed func(), log *slog.Logger) {
	for ctx.Err() == nil {
		events, err := m.Watch(ctx, prefix)
		if err != nil {
			log.Warn("watching the units a dev host sweeps", "prefix", prefix, "err", err.Error())
		} else {
			changed()
			for range events {
				changed()
			}
		}
		if !clock.Sleep(ctx, hostprocess.RetryAfter) {
			return
		}
	}
}

func (h *host) pass(ctx context.Context, log *slog.Logger) error {
	keep := append([]string(nil), h.c.Own...)
	live := map[string]bool{}
	for _, w := range h.store.List() {
		id := w.Name
		if w.DeletionTimestamp != nil {
			continue
		}
		live[id] = true
		keep = append(keep, Prefixes(id)...)
		if _, ok := h.children[id]; ok || workspacePhase(&w) != "Ready" {
			continue
		}
		cctx, cancel := context.WithCancel(ctx)
		h.children[id] = cancel
		childCfg := hostprocess.Config{Kubeconfig: h.c.Kubeconfig, Workspace: ChildWorkspace(id), Manager: h.c.Manager,
			Prefixes: Prefixes(id), Bound: true, Resolve: h.c.Resolve, Log: log.With("child", id)}
		h.wg.Go(func() { hostprocess.Supervise(cctx, childCfg) })
		log.Info("a child instance's workspace is supervised", "child", id, "workspace", ChildWorkspace(id))
	}
	for id, cancel := range h.children {
		if !live[id] {
			cancel()
			delete(h.children, id)
			log.Info("a child instance's workspace is gone; stopping its processes", "child", id)
		}
	}
	return h.sweep(ctx, keep, log)
}

func kept(name string, keep []string) bool {
	for _, p := range keep {
		if strings.HasPrefix(name, p) {
			return true
		}
	}
	return false
}

func (h *host) sweep(ctx context.Context, keep []string, log *slog.Logger) error {
	var gone []hostsvc.UnitStatus
	for _, k := range sweptKinds {
		units, err := h.c.Manager.List(ctx, k)
		if err != nil {
			return err
		}
		for _, u := range units {
			if !kept(u.Name, keep) && (u.State == hostsvc.Active || u.State == hostsvc.Activating || u.State == hostsvc.Failed) {
				gone = append(gone, u)
			}
		}
	}
	sort.Slice(gone, func(i, j int) bool { return gone[i].ActiveSince.After(gone[j].ActiveSince) })
	for _, u := range gone {
		if err := h.c.Manager.Stop(ctx, u.Name); err != nil {
			log.Warn("stopping a gone child's unit", "unit", u.Name, "err", err.Error())
			continue
		}
		h.c.Manager.ResetFailed(ctx, u.Name)
		log.Info("stopped a unit whose child workspace is gone", "unit", u.Name, "pid", u.MainPID)
	}
	return nil
}