johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package gateway

import (
	"cmp"
	"context"
	"fmt"
	"path/filepath"
	"slices"
	"strings"

	apierrors "k8s.io/apimachinery/pkg/api/errors"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketrequester"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clusterstack"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests"
)

const (
	ProfileGatewayRequester = "requester"
	GatewayDownloadsOption  = "downloads-dir"
	GatewayNodeDownloadBase = "node-download-base"
	requesterSetPhase       = "rbac"
	requesterNodeVaultMount = marketrequester.RequesterVaultClient
)

func requesterProfile(o *v1alpha1.Gateway) bool {
	return o.Spec.RequesterRef != nil || o.Spec.Profile == ProfileGatewayRequester
}

func ClusterStateDir(stateRoot string) string { return filepath.Join(stateRoot, "clusterprovider") }

func (p *pass) requesterRoot(ctx context.Context) (*v1alpha1.MarketRequester, string, bool, error) {
	o, r := p.o, p.r
	ref := o.Spec.RequesterRef
	switch {
	case o.Spec.Profile != ProfileGatewayRequester:
		return nil, fmt.Sprintf("spec.requesterRef needs profile %s, not %q", ProfileGatewayRequester, o.Spec.Profile), true, nil
	case ref == nil || strings.TrimSpace(ref.Name) == "":
		return nil, fmt.Sprintf("profile %s needs spec.requesterRef naming the MarketRequester whose Requester rents the nodes", ProfileGatewayRequester), true, nil
	case ref.Kind != "" && ref.Kind != string(v1alpha1.ApplicationKindMarketRequester):
		return nil, fmt.Sprintf("spec.requesterRef names a %s; it names a MarketRequester", ref.Kind), true, nil
	case cmp.Or(strings.TrimSpace(ref.Workspace), r.Workspace) != r.Workspace:
		return nil, fmt.Sprintf("spec.requesterRef names workspace %s; the MarketRequester runs in the Gateway's own workspace %s", ref.Workspace, r.Workspace), true, nil
	}
	mr, err := compose.Get[v1alpha1.MarketRequester](ctx, p.c, r.Workspace, compose.MarketRequesterGVR, ref.Name)
	if apierrors.IsNotFound(err) {
		return nil, "waiting for the MarketRequester " + ref.Name, false, nil
	}
	if err != nil {
		return nil, "", false, err
	}
	if mr.Status.Phase != v1alpha1.ApplicationPhaseReady || strings.TrimSpace(mr.Status.Did) == "" {
		return mr, fmt.Sprintf("waiting for the MarketRequester %s to be Ready with a did (%s: %s)", ref.Name, mr.Status.Phase, mr.Status.Message), false, nil
	}
	return mr, "", false, nil
}

func (p *pass) requesterLinks() []compose.Link {
	return []compose.Link{compose.NodeClaimVaultLink(p.r.Workspace, marketrequester.RequesterVaultServer, requesterNodeVaultMount+"/contracts/*")}
}

func (p *pass) requesterDesired(mr *v1alpha1.MarketRequester) ([]compose.Desired, error) {
	c, o, ws := p.c, p.o, p.r.Workspace
	network := mr.Spec.Network
	stateRoot := c.StateRoot()
	clusterState := ClusterStateDir(stateRoot)
	downloads := strings.TrimSpace(compose.OptionValue(o.Spec.Options, GatewayDownloadsOption))
	if downloads == "" && p.ctl.web != nil {
		downloads = p.ctl.web.DownloadsDir
	}
	storage := &v1alpha1.ComponentStorage{BlobsDir: downloads, CertsDir: filepath.Join(clusterState, clusterstack.PublicCADirName)}
	var required []string
	if downloads != "" {
		required = slices.Clone(GatewayRequiredDownloads)
	}
	var clusterProvider *v1alpha1.GatewayClusterProvider
	if o.Spec.ClusterProvider != nil {
		cp := *o.Spec.ClusterProvider
		clusterProvider = &cp
	}
	market := &v1alpha1.GatewayNodeMarket{HomeWorkspace: p.ws, ContractWorkspace: ws,
		NodePolicy: compose.OptionValue(o.Spec.Options, GatewayNodePolicy), NodePolicyArgs: compose.OptionValue(o.Spec.Options, GatewayNodePolicyArgs),
		InfraRecord: &v1alpha1.GatewayInfraRecord{RepoWorkspace: ws}, VaultMount: requesterNodeVaultMount}
	api := compose.Desired{Kind: compose.ComponentKindGatewayAPI, Name: GatewayAPIComponent, Spec: v1alpha1.GatewayAPISpec{
		ComponentSpec:     compose.NewComponentSpec(compose.ComponentCRStateDir(stateRoot, compose.ComponentKindGatewayAPI, GatewayAPIComponent), network),
		RequesterRef:      compose.ComponentRefFor(compose.ComponentKindRequester, marketrequester.RequesterComponent),
		VaultRef:          compose.ComponentRefFor(compose.KindOpenBaoServer, marketrequester.RequesterVaultServer),
		Storage:           storage,
		ClusterProvider:   clusterProvider,
		ClusterStateDir:   clusterState,
		Audience:          mr.Status.Did,
		PublicOrigin:      strings.TrimRight(strings.TrimSpace(compose.OptionValue(o.Spec.Options, GatewayNodeDownloadBase)), "/"),
		RequiredDownloads: required,
		NodeMarket:        market,
		NodesOnly:         true,
	}}
	out := append([]compose.Desired{api}, compose.HTTPServiceExposures(ws, compose.ComponentKindGatewayAPI, GatewayAPIComponent, "http", network, nil, true, false)...)
	return compose.ApplyOverrides(o.Spec.Components, out)
}

func (p *pass) requester(ctx context.Context) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	mr, msg, failed, err := p.requesterRoot(ctx)
	switch {
	case err != nil:
		return p.retry("the MarketRequester", err)
	case failed:
		return p.write(v1alpha1.ApplicationPhaseFailed, msg)
	case msg != "" && mr == nil:
		return p.wait(msg)
	}
	desired, err := p.requesterDesired(mr)
	if err != nil {
		return p.fail(err)
	}
	if r.Suspend {
		return p.requesterSuspended(ctx, mr, desired)
	}
	if msg != "" {
		return p.wait(msg)
	}
	links := p.requesterLinks()
	if err := c.ApplyLinks(ctx, r, "root", links); err != nil {
		return ctlruntime.Result{}, err
	}
	if !c.Applied(r) {
		values := &manifests.Gateway{}
		digest, digestErr := compose.PhasesDigest(GatewaySet, values, nil, requesterSetPhase)
		if digestErr == nil && c.AppliedBefore(r, o.Status.AppliedSet, digest) {
			c.MarkApplied(r)
		} else {
			if _, err := p.write(v1alpha1.ApplicationPhaseProvisioning, "applying the node claim agent of its workspace"); err != nil {
				return ctlruntime.Result{}, err
			}
			if err := c.ApplySetPhases(ctx, r, GatewaySet, values, requesterSetPhase); err != nil {
				return p.retry("its workspace", err)
			}
			c.MarkApplied(r)
			if digestErr == nil {
				p.st.AppliedSet = &v1alpha1.AppliedSet{Generation: o.Generation, Digest: digest}
			}
		}
	}
	if ok, msg := c.LinksAccepted(ctx, r, links); !ok {
		return p.wait(msg)
	}
	if err := c.EnsureLinked(ctx, r, desired, desired); err != nil {
		return p.retry("", err)
	}
	if _, err := c.DeleteComponentsExcept(ctx, r.Workspace, r.Owner(), compose.DesiredKeys(desired)); err != nil {
		return ctlruntime.Result{}, err
	}
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	p.st.Did = mr.Status.Did
	compose.FillApplicationStatus(&p.st, o.Generation, false, o.Spec.Components, desired, snap, nil)
	if snap.Satisfied(desired) {
		p.st = compose.Ready(p.st, "the node market of the MarketRequester "+mr.Name+" is bound: its Requester rents the tenant clusters' nodes")
		return ctlruntime.Result{}, c.WriteStatus(ctx, r, compose.GatewayGVR, p.st)
	}
	return p.wait("waiting for the nodes-only GatewayAPI to be ready")
}

func (p *pass) requesterSuspended(ctx context.Context, mr *v1alpha1.MarketRequester, desired []compose.Desired) (ctlruntime.Result, error) {
	c, r, o := p.c, p.r, p.o
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	held := compose.Existing(snap, desired)
	if err := p.ensure(ctx, held); err != nil {
		return p.retry("", err)
	}
	if snap, err = p.snapshot(ctx); err != nil {
		return ctlruntime.Result{}, err
	}
	if mr != nil {
		p.st.Did = mr.Status.Did
	}
	compose.FillApplicationStatus(&p.st, o.Generation, true, o.Spec.Components, held, snap, nil)
	p.st = compose.SuspendedOnceApplied(p.st, snap, held)
	return ctlruntime.Result{}, c.WriteStatus(ctx, r, compose.GatewayGVR, p.st)
}

func (p *pass) deleteRequester(ctx context.Context) (ctlruntime.Result, error) {
	c, r := p.c, p.r
	if did := strings.TrimSpace(p.o.Status.Did); did != "" {
		clusters, err := c.DeleteClustersOf(ctx, did)
		if err != nil {
			return ctlruntime.Result{}, err
		}
		if clusters > 0 {
			return p.deleting(fmt.Sprintf("%d Cluster(s) of %s tearing down", clusters, did))
		}
	}
	left, err := c.StopComponents(ctx, r, "")
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if left == 0 {
		if err := c.DeleteLinks(ctx, r); err != nil {
			return ctlruntime.Result{}, err
		}
	}
	if left > 0 {
		return p.deleting(fmt.Sprintf("%d component(s) stopping", left))
	}
	c.Drop(r)
	return ctlruntime.Result{}, nil
}