johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package gatewaykcp

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

	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clusterprovider"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clusterstack"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/clusterhome"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	appgateway "github.com/johnandersen777/socialweb-computer-kcp/internal/app/gateway"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/providermain"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/tlsfront"
)

const RequesterGatewayDir = "gateway"

func RequesterSpec(base v1alpha1.GatewaySpec, requester, downloads, nodeDownloadBase string) v1alpha1.GatewaySpec {
	spec := base
	spec.Profile = appgateway.ProfileGatewayRequester
	spec.RequesterRef = &commonv1.ObjectRef{Kind: string(v1alpha1.ApplicationKindMarketRequester), Name: requester}
	var opts []v1alpha1.ApplicationOption
	for _, o := range base.Options {
		if o.Name != appgateway.GatewayDownloadsOption && o.Name != appgateway.GatewayNodeDownloadBase {
			opts = append(opts, o)
		}
	}
	if downloads != "" {
		opts = append(opts, v1alpha1.ApplicationOption{Name: appgateway.GatewayDownloadsOption, Value: downloads})
	}
	if nodeDownloadBase != "" {
		opts = append(opts, v1alpha1.ApplicationOption{Name: appgateway.GatewayNodeDownloadBase, Value: nodeDownloadBase})
	}
	spec.Options = opts
	return spec
}

func RunRequester(ctx context.Context, c Config, rec devinstance.Record, did string, applied func()) error {
	if rec.AdminKubeconfig == "" || rec.KcpURL == "" {
		return errors.New("gatewaykcp: the requester's instance record names no kcp")
	}
	if strings.TrimSpace(did) == "" {
		return errors.New("gatewaykcp: a requester's gateway serves the clusters of the requester's DID, and it has none yet")
	}
	if c.Spec.RequesterRef == nil || c.Spec.Profile != appgateway.ProfileGatewayRequester {
		return fmt.Errorf("gatewaykcp: a requester's gateway is profile %s with spec.requesterRef", appgateway.ProfileGatewayRequester)
	}
	c.StateDir = rec.Layout.StateDir
	if err := os.MkdirAll(rec.Layout.KcpDir(), 0o700); err != nil {
		return err
	}
	g := &gateway{c: c, l: rec.Layout, reg: &listeners{addr: map[string]string{}, kind: map[string]string{}}}
	if err := g.before(ctx, rec); err != nil {
		return err
	}
	if applied != nil {
		applied()
	}
	cert, key, err := tlsfront.EnsureSelfSigned(filepath.Join(c.StateDir, RequesterGatewayDir, "tls"))
	if err != nil {
		return err
	}
	env := newEnv(c.Environ)
	if err := g.defaults(env, rec, cert, key); err != nil {
		return err
	}
	env.set("SOCIALWEB_CLUSTERPROVIDER_ACCOUNT", strings.TrimSpace(did))
	if got, want := env.get("SOCIALWEB_CLUSTERPROVIDER_STATE_DIR"), appgateway.ClusterStateDir(c.StateDir); got != want {
		return fmt.Errorf("gatewaykcp: SOCIALWEB_CLUSTERPROVIDER_STATE_DIR is %s; the requester's GatewayAPI reads the clusters under %s", got, want)
	}
	log := c.log()
	if code := providermain.Main(ctx, g.invocation(env, []string{"migrate"}, "migrate")); code != 0 {
		log.Warn("the legacy-key migration did not complete; the clusterprovider runs the same migration itself", "exit", code)
	}
	var wg sync.WaitGroup
	defer wg.Wait()
	mctx, cancel := context.WithCancel(ctx)
	defer cancel()
	wg.Go(func() {
		followMemberToken(mctx, g.activeSlotRoot, g.kcp.SlotsFile(), filepath.Join(g.l.KcpDir(), MemberToken), log)
	})
	log.Info("the requester's gateway runs its clusterprovider in this process; its k3s and kine are HostProcesses of the requester's core",
		"workspace", g.ws, "account", did, "unitPrefix", env.get("SOCIALWEB_CLUSTERPROVIDER_UNIT_PREFIX"), "clusters", env.get("SOCIALWEB_CLUSTERPROVIDER_STATE_DIR"))
	err = exitErr(providermain.Main(mctx, g.invocation(env, nil, "serve", "provider")))
	if mctx.Err() != nil {
		return ctx.Err()
	}
	if err != nil {
		return fmt.Errorf("gatewaykcp: the requester's clusterprovider: %w", err)
	}
	return errors.New("gatewaykcp: the requester's clusterprovider returned")
}

func WaitRequesterGateway(ctx context.Context, kubeconfig, ws, name string, timeout time.Duration, progress func(v1alpha1.ApplicationStatus)) (*v1alpha1.Gateway, error) {
	restCfg, err := kcpserver.FrontRestConfig(kubeconfig)
	if err != nil {
		return nil, err
	}
	dc, err := kube.New(clusterConfig(restCfg, ws))
	if err != nil {
		return nil, err
	}
	last := ""
	o, err := kube.Await(ctx, dc.Resource(compose.GatewayGVR), name, timeout, func(o *v1alpha1.Gateway) (bool, error) {
		if o == nil {
			return false, nil
		}
		st := o.Status
		if key := string(st.Phase) + st.Message; key != last && progress != nil {
			last = key
			progress(st)
		}
		if st.ObservedGeneration != o.Generation {
			return false, nil
		}
		if st.Phase == v1alpha1.ApplicationPhaseFailed {
			return false, fmt.Errorf("gatewaykcp: the Gateway %s failed: %s", name, st.Message)
		}
		return st.Phase == v1alpha1.ApplicationPhaseReady, nil
	})
	if errors.Is(err, kube.ErrAwaitTimeout) {
		return nil, fmt.Errorf("gatewaykcp: the Gateway %s was not Ready within %s: %s", name, timeout, last)
	}
	return o, err
}

func TenantAdminKubeconfig(clusters, did, cluster string) (string, error) {
	key, err := clusterhome.Key(clusterhome.HomePath(did, cluster))
	if err != nil {
		return "", err
	}
	dir, err := clusterprovider.ClusterDataDir(clusters, key)
	if err != nil {
		return "", err
	}
	return clusterstack.AdminKubeconfig(dir), nil
}

func RequesterCluster(ctx context.Context, kubeconfig, base, did, cluster string, log *slog.Logger) (string, error) {
	if err := bootstrapCluster(ctx, kubeconfig, base, did, cluster, log); err != nil {
		return "", err
	}
	return clusterhome.HomePath(did, cluster), nil
}

func RequesterClusterPhase(ctx context.Context, kubeconfig, did, cluster string) (v1alpha1.ClusterPhase, error) {
	restCfg, err := memberRestConfig(kubeconfig)
	if err != nil {
		return "", err
	}
	home, err := kube.New(clusterConfig(restCfg, clusterhome.HomePath(did, cluster)))
	if err != nil {
		return "", err
	}
	c, err := kube.Get[v1alpha1.Cluster](ctx, home.Resource(clusterGVR), cluster)
	if err != nil {
		return "", err
	}
	return c.Status.Phase, nil
}