johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package devcore

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

	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/app"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/gateway"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketbidder"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketnetwork"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/marketrequester"
	atprotooauth "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/oauth"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap/table"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/component"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/computevm"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostos"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostprocess"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/install"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kcpclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/logcap"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/openbaoctl"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/postgresctl"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/stack/componentcore"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/storemaint"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/vaultlayout"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/versions"
)

const (
	CheckInterval  = 5 * time.Second
	recordLockWait = 2 * time.Minute
)

var TTLSkipWorkspaces = []string{"socialweb-provider"}

func TTLScope(root string) func(workspace string) bool {
	return func(workspace string) bool {
		if workspace == root {
			return true
		}
		rest, ok := strings.CutPrefix(workspace, root+":")
		return ok && !strings.Contains(rest, ":") && !slices.Contains(TTLSkipWorkspaces, rest)
	}
}

type Config struct {
	Layout       devinstance.Layout
	ReadyTimeout time.Duration
	Log          *slog.Logger
	TLSCertFile  string
	TLSKeyFile   string
	ComputeVM    *computevm.Config
	Application  *table.App
	FrontAddress string
	OnStep       func(step string)
	OnTick       func()
	Before       func(ctx context.Context, rec devinstance.Record) error
	Members      func(ctx context.Context, rec devinstance.Record) error
	Listeners    func(owner string) []v1alpha1.ApplicationListener
	WebApp       *gateway.WebApp
	OnComponents func(owner string, listeners []v1alpha1.ApplicationListener)
	SignInGuard  marketbidder.SignInGuard
	Handlers     *component.HandlerRegistry
	RemotePDS    *pdsv1alpha1.RemotePDS

	NodeClaimAgent *kcpclient.ServiceAccountRef
}

func LogCapConfig(l devinstance.Layout, log *slog.Logger) logcap.Config {
	c := logcap.Config{Dirs: []string{l.Logs}, Log: log}
	if info, err := os.Stat(l.SupState); l.SupState != "" && err == nil && info.IsDir() {
		c.Dirs = append(c.Dirs, filepath.Join(l.SupState, "logs"))
		c.Files = append(c.Files, filepath.Join(l.SupState, "devsup.log"))
	}
	return c
}

func tryLockRecord(path string) (*os.File, bool) {
	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
	if err != nil {
		return nil, false
	}
	if err := hostos.Lock(f, hostos.LockExclusive|hostos.LockNonBlocking); err != nil {
		f.Close()
		return nil, false
	}
	return f, true
}

func lockRecord(ctx context.Context, path string) (*os.File, error) {
	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
	if err != nil {
		return nil, err
	}
	if err := hostos.LockWait(ctx, f, hostos.LockExclusive, recordLockWait); err != nil {
		f.Close()
		return nil, fmt.Errorf("devcore: locking %s: %w", path, err)
	}
	return f, nil
}

func Run(ctx context.Context, cfg Config, m hostsvc.Manager) error {
	l := cfg.Layout
	log := cfg.Log.With("component", "dev-core", "instance", l.ID)
	handlers := cfg.Handlers
	if handlers == nil {
		handlers = component.NewHandlerRegistry(ctx)
	}
	cfg.Handlers = handlers
	lf, err := os.OpenFile(l.CoreLock, os.O_CREATE|os.O_RDWR, 0o600)
	if err != nil {
		return err
	}
	defer lf.Close()
	if err := hostos.Lock(lf, hostos.LockExclusive|hostos.LockNonBlocking); err != nil {
		return fmt.Errorf("devcore: another core holds %s: %w", l.CoreLock, err)
	}
	var rec devinstance.Record
	if err := fsx.ReadJSON(l.Record, &rec); err != nil {
		return fmt.Errorf("devcore: reading %s: %w", l.Record, err)
	}
	var prev devinstance.Core
	fsx.ReadJSON(l.CoreRecord, &prev)
	core := devinstance.Core{Generation: prev.Generation + 1, PID: os.Getpid(), InvocationID: os.Getenv("INVOCATION_ID"),
		StartedAt: time.Now().UTC()}
	bcfg := bootstrap.Config{Layout: l, KcpPath: rec.Executables["kcp"].Path, KinePath: rec.Executables["kine"].Path, PostgresRoot: rec.Executables["postgres"].Path,
		Exe: rec.Executables["socialweb-computer"].Path, ReadyTimeout: cfg.ReadyTimeout, Application: cfg.Application, FrontAddress: cfg.FrontAddress,
		OnStep: cfg.OnStep, OnTick: cfg.OnTick, Log: log}
	lk, err := lockRecord(ctx, l.Lock)
	if err != nil {
		return err
	}
	if err := fsx.ReadJSON(l.Record, &rec); err != nil {
		lk.Close()
		return fmt.Errorf("devcore: reading %s: %w", l.Record, err)
	}
	err = ensure(ctx, m, bcfg, l, &rec, &core, log)
	if err == nil && cfg.Application != nil && rec.Backend == hostsvc.BackendSystemdUser && rec.SharedKcp == "" {
		err = rollDrifted(ctx, m, bcfg, cfg, l, &rec, &core, log)
	}
	lk.Close()
	if err != nil {
		return err
	}
	if cfg.Before != nil {
		if err := cfg.Before(ctx, rec); err != nil {
			return err
		}
	}
	g := supervise.New(ctx)
	defer g.Stop()
	ctx = g.Context()
	logCfg := LogCapConfig(l, log)
	g.Go("logcap", func(ctx context.Context) error {
		logcap.Keep(ctx, logCfg)
		return nil
	})
	if len(rec.StandIns) == 0 && rec.AdminKubeconfig != "" {
		ac := AppConfig(bcfg, rec, m, log)
		ac.Bus, ac.RemotePDS = cfg.Handlers, cfg.RemotePDS
		ac.Listeners, ac.OnComponents = cfg.Listeners, cfg.OnComponents
		var ttlReport atomic.Pointer[func() *v1alpha1.TTLStatus]
		ac.TTLReport = func() *v1alpha1.TTLStatus {
			if report := ttlReport.Load(); report != nil {
				return (*report)()
			}
			return nil
		}
		coreCfg := ComponentCoreConfig(ctx, cfg, rec, log)
		coreCfg.Postgres = postgresctl.New(PostgresConfig(bcfg, rec))
		if servers, err := openbaoctl.New(OpenBaoServerConfig(bcfg, rec, log)); err != nil {
			log.Warn("the OpenBaoServer controller does not start", "err", err.Error())
		} else {
			coreCfg.OpenBao = servers
		}
		coreCfg.TTLScope = TTLScope(cmp.Or(ac.Workspace, hostprocess.DefaultWorkspace))
		coreCfg.OnTTL = func(report func() *v1alpha1.TTLStatus) { ttlReport.Store(&report) }
		if rec.SharedKcp == "" {
			if sc, err := StoreMaintConfig(bcfg, rec, log); err != nil {
				log.Warn("the store maintenance does not start", "err", err.Error())
			} else {
				mon := storemaint.New(sc)
				ac.StoreUsage = mon.Usage
				g.Go("storemaint", func(ctx context.Context) error {
					mon.Run(ctx)
					return nil
				})
			}
		}
		coreCfg.Apps = &app.Config{Compose: ac, WebApp: cfg.WebApp, SignInGuard: cfg.SignInGuard}
		g.Go("componentcore", func(ctx context.Context) error {
			componentcore.Supervise(ctx, coreCfg)
			return nil
		})
		hpCfg := HostProcessConfig(bcfg, rec, m, log)
		g.Go("hostprocess", func(ctx context.Context) error {
			hostprocess.Supervise(ctx, hpCfg)
			return nil
		})
	}
	members := make(chan error, 1)
	if cfg.Members != nil {
		launched := rec
		g.Go("members", func(ctx context.Context) error {
			members <- cfg.Members(ctx, launched)
			return nil
		})
	}
	if rec.Host && rec.AdminKubeconfig != "" {
		hostCfg := devinstance.HostConfig{Kubeconfig: rec.AdminKubeconfig, Manager: m, Own: devinstance.Prefixes(l.ID), Log: log}
		g.Go("devinstance-host", func(ctx context.Context) error {
			devinstance.RunHost(ctx, hostCfg)
			return nil
		})
	}
	var events <-chan hostsvc.UnitEvent
	watched := ""
	unwatch := context.CancelFunc(func() {})
	defer func() { unwatch() }()
	retry := time.NewTimer(0)
	defer retry.Stop()
	denied := handlers.DeniedChanged()
	for {
		select {
		case <-ctx.Done():
			log.Info("core stopping", "generation", core.Generation)
			return g.Stop()
		case <-denied:
			core.BusDenied, core.BusDeniedBy = handlers.Denied(), handlers.DeniedByCapability()
			if err := fsx.WriteJSON(l.CoreRecord, core, 0o600); err != nil {
				log.Warn("recording the bus deny count", "err", err.Error())
			}
			continue
		case err := <-members:
			if ctx.Err() != nil {
				return nil
			}
			if err == nil {
				err = errors.New("devcore: an in-process member returned")
			}
			log.Error("an in-process member ended; the core exits", "err", err.Error())
			return err
		case _, open := <-events:
			if !open {
				events, watched = nil, ""
			}
			for drained := events == nil; !drained; {
				select {
				case _, open := <-events:
					if !open {
						events, watched, drained = nil, "", true
					}
				default:
					drained = true
				}
			}
		case <-retry.C:
		}
		if ctx.Err() != nil {
			continue
		}
		cur, settled := check(ctx, m, bcfg, l, &core, cfg.OnTick, log)
		if prefix := unitPrefix(cur.Units); prefix != "" && prefix != watched {
			unwatch()
			events, watched = nil, ""
			wctx, cancel := context.WithCancel(ctx)
			if ch, err := m.Watch(wctx, prefix); err != nil {
				cancel()
				log.Error("watching the bootstrap units", "prefix", prefix, "err", err.Error())
				settled = false
			} else {
				events, watched, unwatch = ch, prefix, cancel
				retry.Reset(0)
				continue
			}
		}
		if !settled {
			retry.Reset(CheckInterval)
		}
	}
}

func check(ctx context.Context, m hostsvc.Manager, bcfg bootstrap.Config, l devinstance.Layout, core *devinstance.Core,
	onTick func(), log *slog.Logger) (devinstance.Record, bool) {
	if onTick != nil {
		onTick()
	}
	cur := devinstance.Record{}
	lk, ok := tryLockRecord(l.Lock)
	if !ok {
		log.Info("a dev verb holds the instance; this check waits for it")
		return cur, false
	}
	defer lk.Close()
	if err := fsx.ReadJSON(l.Record, &cur); err != nil {
		log.Error("re-reading the instance record", "err", err.Error())
		return cur, false
	}
	lost := false
	core.Verdicts = nil
	for _, u := range cur.Units {
		obs, restarted, err := bootstrap.Observe(ctx, m, u.Identity)
		if onTick != nil {
			onTick()
		}
		if err != nil {
			log.Error("check", "unit", u.Unit, "err", err.Error())
			continue
		}
		if restarted {
			lost = true
			log.Warn("unit restarted in place by devsup; re-identifying it", "unit", u.Unit, "detail", obs.Detail)
			core.Verdicts = append(core.Verdicts, fmt.Sprintf("%s=Restarted", u.Unit))
			continue
		}
		core.Verdicts = append(core.Verdicts, fmt.Sprintf("%s=%s %s", u.Unit, obs.Verdict, obs.Reason))
		if obs.Verdict == hostsvc.VerdictLost {
			lost = true
			log.Warn("unit lost", "unit", u.Unit, "reason", string(obs.Reason), "detail", obs.Detail)
		}
	}
	core.LastCheck = time.Now().UTC()
	if lost {
		if err := ensure(ctx, m, bcfg, l, &cur, core, log); err != nil {
			log.Error("re-placing the bootstrap units", "err", err.Error())
			return cur, false
		}
		return cur, true
	}
	fsx.WriteJSON(l.CoreRecord, *core, 0o600)
	return cur, true
}

func unitPrefix(units []devinstance.Unit) string {
	prefix := ""
	for i, u := range units {
		if i == 0 {
			prefix = strings.TrimSuffix(u.Unit, hostsvc.ServiceSuffix)
			continue
		}
		n := 0
		for n < len(prefix) && n < len(u.Unit) && prefix[n] == u.Unit[n] {
			n++
		}
		prefix = prefix[:n]
	}
	return prefix
}

func ensure(ctx context.Context, m hostsvc.Manager, bcfg bootstrap.Config, l devinstance.Layout, rec *devinstance.Record,
	core *devinstance.Core, log *slog.Logger) error {
	if err := bootstrap.Ensure(ctx, m, bcfg, rec); err != nil {
		return err
	}
	core.Adoptions = nil
	for _, u := range rec.Units {
		core.Adoptions = append(core.Adoptions, devinstance.Adoption{Role: u.Role, Unit: u.Unit,
			InvocationID: u.Identity.InvocationID, PID: u.Identity.PID, Adopted: u.Adopted})
	}
	core.AdoptedAt = time.Now().UTC()
	core.LastCheck = core.AdoptedAt
	if err := fsx.WriteJSON(l.Record, *rec, 0o600); err != nil {
		return err
	}
	if err := fsx.WriteJSON(l.CoreRecord, *core, 0o600); err != nil {
		return err
	}
	log.Info("core holds the instance", "generation", core.Generation, "adoptions", core.Adoptions)
	return nil
}

func rollDrifted(ctx context.Context, m hostsvc.Manager, bcfg bootstrap.Config, cfg Config, l devinstance.Layout, rec *devinstance.Record,
	core *devinstance.Core, log *slog.Logger) error {
	compat, err := versions.KcpCompatRecords()
	if err != nil {
		return err
	}
	d, rep, err := bootstrap.RollDrifted(ctx, m, bootstrap.InstallParams(*cfg.Application, l.StateDir),
		bootstrap.InstalledRoll{Compat: compat, ReadyTimeout: cfg.ReadyTimeout, Log: log})
	if err != nil {
		log.Error("the running kcp slot differs from what this binary renders and was not rolled; it keeps its old argv",
			"unit", d.Unit, "argvChanged", d.ArgvChanged, "exeChanged", d.ExeChanged, "refused", d.Refused, "err", err.Error())
		return nil
	}
	if rep == nil {
		return nil
	}
	log.Info("the kcp slot rolled to this binary's argv", "from", rep.Roll.From, "to", rep.Roll.To, "unit", d.Unit)
	return ensure(ctx, m, bcfg, l, rec, core, log)
}

func ComponentCoreConfig(ctx context.Context, cfg Config, rec devinstance.Record, log *slog.Logger) componentcore.Config {
	out := componentcore.Config{
		Kubeconfig:              rec.AdminKubeconfig,
		Workspaces:              componentcore.AppWorkspaces(rec.BaseWorkspace),
		StateDir:                rec.Layout.StateDir,
		Log:                     log,
		TLSCertFile:             cfg.TLSCertFile,
		TLSKeyFile:              cfg.TLSKeyFile,
		Handlers:                cfg.Handlers,
		ExposureIngressIdentity: componentcore.NewExposureIngressIdentity(rec.Layout.StateDir),
		SlotsFile:               rec.Layout.SlotsFile(),
		FrontControl:            rec.Layout.FrontControl(),
		InstallStateRoot:        install.StateRoot,
		NodeClaimAgent:          cfg.NodeClaimAgent,
	}
	if cfg.ComputeVM != nil {
		c := *cfg.ComputeVM
		c.Kubeconfig, c.Workspace = rec.AdminKubeconfig, cmp.Or(rec.BaseWorkspace, hostprocess.DefaultWorkspace)
		out.Compute = &c
	}
	if rec.AdminKubeconfig != "" {
		if n, err := marketnetwork.ReadNetwork(ctx, rec.AdminKubeconfig); err == nil && n.Status.Network != nil {
			out.PLCBaseURL = n.Status.Network.PLCDirectoryURL
			out.Relays = n.Status.Network.Relays()
			network := *n.Status.Network
			out.Network = &network
		} else if joined, err := marketnetwork.ReadJoinedNetwork(ctx, rec.AdminKubeconfig); err == nil && joined != nil {
			out.PLCBaseURL = joined.PLCDirectoryURL
			out.Relays = joined.Relays()
			out.Network = joined
		}
	}
	switch {
	case cfg.Application != nil && cfg.Application.Role == table.RoleBidder:
		out.CredentialClient = marketbidder.BidderVaultClient
		out.CredentialServer = marketbidder.BidderVaultServer
		out.OAuthScopes = atprotooauth.BidderOAuthScope
	case cfg.Application != nil && cfg.Application.Role == table.RoleGateway:
		out.CredentialClient = marketrequester.RequesterVaultClient
		out.CredentialServer = vaultlayout.RootServerName
		out.OAuthScopes = atprotooauth.RequesterOAuthScope
	case cfg.Application != nil && cfg.Application.Role == table.RoleRequester, rec.Role == devinstance.RoleRequester:
		out.CredentialClient = marketrequester.RequesterVaultClient
		out.CredentialServer = marketrequester.RequesterVaultServer
		out.OAuthScopes = atprotooauth.RequesterOAuthScope
	case strings.Contains(rec.BaseWorkspace, ":gateway"), strings.Contains(rec.BaseWorkspace, ":ssh"):
		out.CredentialClient = marketrequester.RequesterVaultClient
		out.CredentialServer = vaultlayout.RootServerName
		out.OAuthScopes = atprotooauth.RequesterOAuthScope
	case strings.Contains(rec.BaseWorkspace, ":bidder"), strings.HasPrefix(rec.BaseWorkspace, "b-"), strings.HasPrefix(rec.Layout.ID, "b-"):
		out.CredentialClient = marketbidder.BidderVaultClient
		out.CredentialServer = marketbidder.BidderVaultServer
		out.OAuthScopes = atprotooauth.BidderOAuthScope
	case strings.HasPrefix(rec.BaseWorkspace, "r-"), strings.HasPrefix(rec.Layout.ID, "r-"):
		out.CredentialClient = marketrequester.RequesterVaultClient
		out.CredentialServer = vaultlayout.RootServerName
		out.OAuthScopes = atprotooauth.RequesterOAuthScope
	}
	if rec.Profile == bootstrap.ProfileQuick {
		out.InstallAccount = install.AccountBidder
	} else {
		out.InstallAccount = install.AccountGateway
	}
	if rec.Profile == bootstrap.ProfileMarketNetwork {
		out.Workspaces = componentcore.NetworkWorkspaces(rec.BaseWorkspace)
	}
	if rec.Profile == bootstrap.ProfileMarketNetwork && rec.AdminKubeconfig != "" {
		out.WorkspacesFunc = func(ctx context.Context) []string {
			n, err := marketnetwork.ReadNetwork(ctx, rec.AdminKubeconfig)
			if err != nil {
				return out.Workspaces
			}
			seen := map[string]bool{}
			var workspaces []string
			add := func(workspace string) {
				if workspace == "" || seen[workspace] {
					return
				}
				seen[workspace] = true
				workspaces = append(workspaces, workspace)
			}
			for _, workspace := range out.Workspaces {
				add(workspace)
			}
			for _, c := range n.Status.Components {
				add(c.Workspace)
			}
			return workspaces
		}
	}
	return out
}

var ErrNotReady = errors.New("devcore: the core did not record this generation in time")