johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package main

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"maps"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	appgateway "github.com/johnandersen777/socialweb-computer-kcp/internal/app/gateway"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketnetwork"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/assets"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/clicfg"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/clusterhome"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostcontract"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/stack/gatewaykcp"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/stack/request"
)

var requesterGatewayOptions = []string{"name", "state-dir", "ready-file", "cluster", "network-timeout", "log-level"}

var requesterGatewayAssets = []string{gatewaykcp.ExeK3s, gatewaykcp.ExeExternalSecrets, gatewaykcp.ExeESOBundle, gatewaykcp.ExeWebDownloads,
	gatewaykcp.ExeDumbpipe, gatewaykcp.ExeKine, gatewaykcp.ExeBao}

func (r *requesterRun) gatewayPreflight(vals clicfg.Values) (map[string]devinstance.Executable, error) {
	results := hostcontract.EvaluateRequesterGateway(hostcontract.GatherRequesterGateway(hostcontract.Options{StateDir: r.dir}))
	if failed := hostcontract.Failed(results); len(failed) > 0 {
		hostcontract.Write(r.a.stderr, results, false)
		return nil, fmt.Errorf("--with-gateway runs k3s and kine in this account's user manager, and the host does not provide what they need: %s (check-host -role gateway names the same fixes)",
			strings.Join(failed, ", "))
	}
	if !assets.Release() {
		return nil, errors.New("--with-gateway runs the gateway's embedded k3s and external-secrets: build with scripts/build-socialweb-computer.sh (--dev for a dev build)")
	}
	cache, err := r.a.assetCache(vals)
	if err != nil {
		return nil, err
	}
	out := map[string]devinstance.Executable{}
	for _, name := range requesterGatewayAssets {
		e, err := assets.Resolve(assets.Options{Cache: cache, Log: r.a.log}, name)
		if err != nil {
			return nil, fmt.Errorf("--with-gateway: %w", err)
		}
		out[name] = devinstance.Executable{Path: e.Path, Version: e.Version}
	}
	return out, nil
}

func (r *requesterRun) instanceRecord() (devinstance.Record, error) {
	var rec devinstance.Record
	if r.instance != nil {
		return *r.instance, nil
	}
	if r.inv.Mode != request.ModeNetwork {
		err := fsx.ReadJSON(r.standalone.Record, &rec)
		return rec, err
	}
	netLayout, _, err := r.networkLayout()
	if err != nil {
		return rec, err
	}
	return r.childRecord(netLayout, r.inv.Instance)
}

func (r *requesterRun) startGateway(ctx context.Context, assetExes map[string]devinstance.Executable, did string) (<-chan error, *request.ReadyGateway, error) {
	rec, err := r.instanceRecord()
	if err != nil {
		return nil, nil, fmt.Errorf("the requester's instance record: %w", err)
	}
	exes := map[string]devinstance.Executable{}
	maps.Copy(exes, assetExes)
	for name, e := range rec.Executables {
		if e.Path != "" {
			exes[name] = e
		}
	}
	if exes[gatewaykcp.ExeCore].Path == "" {
		exe, err := selfExe()
		if err != nil {
			return nil, nil, err
		}
		exes[gatewaykcp.ExeCore] = devinstance.Executable{Path: exe}
	}
	spec := gatewaykcp.RequesterSpec(v1alpha1.GatewaySpec{}, request.RootName, exes[gatewaykcp.ExeWebDownloads].Path, r.nodeDownloadBase())
	cfg := gatewaykcp.Config{Backend: rec.Backend, Executables: exes, Spec: spec, RootName: appgateway.GatewayRootName, Environ: os.Environ(),
		Exe: exes[gatewaykcp.ExeCore].Path, Stderr: r.a.stderr, Log: r.a.log}
	ws := cmp.Or(strings.TrimSpace(rec.BaseWorkspace), r.workspace, "root")
	applied := make(chan struct{})
	errc := make(chan error, 1)
	go func() { errc <- gatewaykcp.RunRequester(ctx, cfg, rec, did, func() { close(applied) }) }()
	select {
	case err := <-errc:
		return nil, nil, fmt.Errorf("the gateway did not start: %w", err)
	case <-applied:
	case <-ctx.Done():
		return nil, nil, ctx.Err()
	}
	r.log("the Gateway root (profile requester) is applied; its clusterprovider runs in this process", "workspace", ws, "root", appgateway.GatewayRootName)
	wctx, cancel := context.WithCancel(ctx)
	defer cancel()
	waited := make(chan error, 1)
	go func() {
		_, err := gatewaykcp.WaitRequesterGateway(wctx, rec.AdminKubeconfig, ws, appgateway.GatewayRootName, r.networkTimeout, func(st v1alpha1.ApplicationStatus) {
			r.log("gateway", "phase", string(st.Phase), "message", st.Message)
		})
		waited <- err
	}()
	select {
	case err := <-errc:
		return nil, nil, fmt.Errorf("the gateway's clusterprovider ended: %w", err)
	case err := <-waited:
		if err != nil {
			return nil, nil, err
		}
	}
	return errc, &request.ReadyGateway{Workspace: ws, Name: appgateway.GatewayRootName, Clusters: appgateway.ClusterStateDir(rec.Layout.StateDir)}, nil
}

func runRequesterKubeconfig(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	ready, _, _, code := readRequesterReady(a, vals, rest, "kubeconfig")
	if code != exitOK {
		return code
	}
	cluster := strings.TrimSpace(vals.String("cluster"))
	if cluster == "" {
		return a.fail(exitRefused, "run requester kubeconfig: --cluster names the tenant cluster")
	}
	if ready.Gateway == nil {
		return a.fail(exitRefused, "run requester kubeconfig: the requester %s runs no gateway (start it with --with-gateway)", ready.Name)
	}
	path, err := gatewaykcp.TenantAdminKubeconfig(ready.Gateway.Clusters, ready.Did, cluster)
	if err != nil {
		return a.fail(exitRefused, "run requester kubeconfig: %v", err)
	}
	if !fsx.Exists(path) {
		return a.fail(exitFailed, "run requester kubeconfig: the cluster %s has no admin kubeconfig yet (%s); kubectl get clusters.socialweb.computer in its home workspace shows its phase", cluster, path)
	}
	timeout, err := time.ParseDuration(vals.String("network-timeout"))
	if err != nil {
		return a.fail(exitRefused, "run requester kubeconfig: network-timeout: %v", err)
	}
	cctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	phase, err := gatewaykcp.RequesterClusterPhase(cctx, ready.Kubeconfig, ready.Did, cluster)
	if err != nil {
		return a.fail(exitFailed, "run requester kubeconfig: the cluster %s: %v", cluster, err)
	}
	if phase != v1alpha1.ClusterReady {
		return a.fail(exitFailed, "run requester kubeconfig: the cluster %s is %s, not Ready yet (its home workspace %s)", cluster, cmp.Or(string(phase), "without a phase"), clusterhome.HomePath(ready.Did, cluster))
	}
	fmt.Fprintln(a.stdout, path)
	return exitOK
}

func runRequesterCluster(ctx context.Context, a *app, vals clicfg.Values, rest []string) int {
	ready, _, _, code := readRequesterReady(a, vals, rest, "cluster")
	if code != exitOK {
		return code
	}
	cluster := strings.TrimSpace(vals.String("cluster"))
	if cluster == "" {
		return a.fail(exitRefused, "run requester cluster: --cluster names the tenant cluster")
	}
	if ready.Gateway == nil {
		return a.fail(exitRefused, "run requester cluster: the requester %s runs no gateway (start it with --with-gateway)", ready.Name)
	}
	if ready.Did == "" {
		return a.fail(exitFailed, "run requester cluster: the ready file names no did yet")
	}
	timeout, err := time.ParseDuration(vals.String("network-timeout"))
	if err != nil {
		return a.fail(exitRefused, "run requester cluster: network-timeout: %v", err)
	}
	cctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()
	home, err := gatewaykcp.RequesterCluster(cctx, ready.Kubeconfig, ready.Gateway.Workspace, ready.Did, cluster, a.log)
	if err != nil {
		return a.fail(exitFailed, "run requester cluster: %v", err)
	}
	a.printJSON(map[string]string{"cluster": cluster, "account": ready.Did, "home": home,
		"kubeconfig": "run requester kubeconfig " + ready.Name + " --cluster " + cluster})
	return exitOK
}

func requesterGatewayMarker(l devinstance.Layout) string {
	return filepath.Join(l.StateDir, gatewaykcp.RequesterGatewayDir)
}

func (a *app) requesterGatewayOutbound(ctx context.Context, rec devinstance.Record) (func(), error) {
	if rec.Gateway != nil || rec.SharedKcp == "" || rec.AdminKubeconfig == "" || !fsx.Exists(requesterGatewayMarker(rec.Layout)) {
		return func() {}, nil
	}
	if err := os.MkdirAll(rec.Layout.KcpDir(), 0o700); err != nil {
		return nil, err
	}
	n, err := marketnetwork.ReadNetwork(ctx, rec.AdminKubeconfig)
	if err != nil {
		return nil, err
	}
	_, stop, err := gatewaykcp.NetworkOutbound(ctx, n.Status.Network, filepath.Join(requesterGatewayMarker(rec.Layout), "tls"),
		gatewaykcp.RequesterEgressRecord(rec.Layout), a.log)
	return stop, err
}

func (r *requesterRun) markGateway(ctx context.Context) error {
	if r.inv.Mode != request.ModeNetwork {
		return nil
	}
	netLayout, netRec, err := r.networkLayout()
	if err != nil {
		return err
	}
	n, err := marketnetwork.ReadNetwork(ctx, netRec.AdminKubeconfig)
	if err != nil {
		return err
	}
	for _, app := range n.Spec.Apps {
		if app.Kind == v1alpha1.ApplicationKindGateway {
			return fmt.Errorf("--with-gateway: the market network %s has the gateway child %s, whose clusterprovider serves every Cluster of the network's kcp; bring the network up without --with-gateway",
				r.inv.Network, app.Name)
		}
	}
	cl, err := devinstance.NewLayout(netLayout.ID+"-"+r.inv.Name, filepath.Join(netLayout.StateDir, compose.AppsDir), filepath.Join(netLayout.RuntimeDir, compose.AppsDir))
	if err != nil {
		return err
	}
	return os.MkdirAll(requesterGatewayMarker(cl), 0o700)
}

func (r *requesterRun) nodeDownloadBase() string {
	base := strings.TrimRight(strings.TrimSpace(r.vals.String("node-download-base")), "/")
	if r.inv.Mode == request.ModeNetwork && r.vals.Source("node-download-base") == clicfg.SourceDefault {
		return ""
	}
	return base
}