johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package bidder

import (
	"context"
	"fmt"
	"log/slog"
	"maps"
	"os"
	"path/filepath"
	"sync/atomic"
	"time"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap/table"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/computevm"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devcore"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/devinstance"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/install"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fsx"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/notify"
)

const (
	ExeCore           = "socialweb-computer"
	ExeKcp            = "kcp"
	ExeKine           = "kine"
	ExeBao            = "bao"
	ApplyTimeout      = time.Minute
	RootApplyDeadline = 3 * time.Minute
	RootRetryLogEvery = time.Minute
	RootRetryMax      = 15 * time.Second
)

type CoreConfig struct {
	StateDir     string
	Prefix       string
	Handle       string
	Executables  map[string]devinstance.Executable
	Spec         v1alpha1.MarketBidderSpec
	VM           devinstance.VMRuntime
	ComputeVM    computevm.Config
	FrontAddress string
	ReadyTimeout time.Duration
	Unit         string
	Log          *slog.Logger
}

func NewLayout(state, prefix, handle string) devinstance.Layout {
	return install.BidderLayout(state, prefix, handle)
}

func (c CoreConfig) app() table.App {
	a := table.Bidder(c.Handle, install.HandleID(c.Handle))
	if c.Prefix != "" {
		a.Prefix = c.Prefix
	}
	return a
}

func (c CoreConfig) record(l devinstance.Layout) (devinstance.Record, error) {
	var rec devinstance.Record
	if err := fsx.ReadJSON(l.Record, &rec); err != nil {
		rec = devinstance.Record{Created: time.Now().UTC()}
	}
	p := bootstrap.InstallParams(c.app(), c.StateDir)
	exes := map[string]devinstance.Executable{}
	maps.Copy(exes, c.Executables)
	if kcp, kine := c.Executables[ExeKcp].Path, c.Executables[ExeKine].Path; kcp != "" && kine != "" {
		if err := bootstrap.LinkAssets(p, kcp, kine, ""); err != nil {
			return rec, err
		}
	}
	for name, link := range map[string]string{ExeKcp: p.KcpPath, ExeKine: p.KinePath} {
		if _, err := os.Stat(link); err != nil {
			return rec, fmt.Errorf("bidder: %s is not linked at %s (install links it; a dev build names it with its -path option): %w", name, link, err)
		}
		e := exes[name]
		e.Path = link
		exes[name] = e
	}
	vm := c.VM
	rec.Layout, rec.Profile, rec.Backend, rec.Executables = l, bootstrap.ProfileQuick, hostsvc.BackendSystemdUser, exes
	rec.BaseWorkspace, rec.Host, rec.Parent, rec.StandIns, rec.OpenBao, rec.VM = "", false, nil, nil, nil, &vm
	return rec, nil
}

func RunCore(ctx context.Context, c CoreConfig, m hostsvc.Manager) error {
	if !filepath.IsAbs(c.StateDir) {
		return fmt.Errorf("bidder: state dir %q is not absolute", c.StateDir)
	}
	l := NewLayout(c.StateDir, c.Prefix, c.Handle)
	for _, d := range []string{l.StateDir, l.KcpDir(), l.RuntimeDir, l.Logs} {
		if err := os.MkdirAll(d, 0o700); err != nil {
			return err
		}
	}
	if n := len(filepath.Join(l.RuntimeDir, "kine-a.sock")); n > devinstance.MaxSocketPath {
		return fmt.Errorf("bidder: the runtime dir %s makes socket paths of %d bytes, over %d", l.RuntimeDir, n, devinstance.MaxSocketPath)
	}
	rec, err := c.record(l)
	if err != nil {
		return err
	}
	if err := fsx.WriteJSON(l.Record, rec, 0o600); err != nil {
		return err
	}
	app := c.app()
	log := c.Log.With("component", "run-bidder", "handle", c.Handle)
	log.Info("the bidder application instance starts: kcp store, kine, front, its slot, then its MarketBidder root object",
		"state", l.StateDir, "units", app.Names().UnitPrefix(), "profile", c.Spec.Profile)
	cvm := c.ComputeVM
	var watchdogLogged atomic.Bool
	go notify.KeepAlive(ctx, func(err error) {
		if watchdogLogged.CompareAndSwap(false, true) {
			log.Warn("telling systemd's watchdog the core is alive", "err", err.Error())
		}
	})
	return devcore.Run(ctx, devcore.Config{Layout: l, ReadyTimeout: c.ReadyTimeout, Log: c.Log, Application: &app,
		FrontAddress: c.FrontAddress, ComputeVM: &cvm, SignInGuard: CheckSignIn,
		OnStep: func(step string) { log.Info("the core brings up the stack", "step", step) },
		OnTick: func() {
			if err := notify.Watchdog(); err != nil && watchdogLogged.CompareAndSwap(false, true) {
				log.Warn("telling systemd's watchdog the core is alive", "err", err.Error())
			}
		},
		Members: func(ctx context.Context, rec devinstance.Record) error {
			k, err := OpenKcp(rec.AdminKubeconfig)
			if err != nil {
				return err
			}
			if err := c.applyRoot(ctx, k, log); err != nil {
				return err
			}
			if err := notify.Ready(); err != nil {
				return fmt.Errorf("bidder: telling systemd the core is ready: %w", err)
			}
			<-ctx.Done()
			return nil
		}}, m)
}

func attempts(n int) string {
	if n == 1 {
		return "1 attempt"
	}
	return fmt.Sprintf("%d attempts", n)
}

func (c CoreConfig) applyRoot(ctx context.Context, k *Kcp, log *slog.Logger) error {
	started := time.Now()
	last := started
	var lastErr error
	delay := Poll
	for attempt := 1; ; attempt++ {
		remaining := RootApplyDeadline - time.Since(started)
		if remaining <= 0 {
			return fmt.Errorf("bidder: the MarketBidder root object is not applied after %s and %s: %w",
				time.Since(started).Round(time.Second), attempts(attempt-1), lastErr)
		}
		actx, cancel := context.WithTimeout(ctx, remaining)
		err := k.ApplyRoot(actx, c.Spec)
		cancel()
		if err == nil {
			log.Info("the MarketBidder root object is applied; the core renders its components", "profile", c.Spec.Profile,
				"attempts", attempt, "after", time.Since(started).Round(time.Second).String())
			return nil
		}
		lastErr = err
		if attempt == 1 || time.Since(last) >= RootRetryLogEvery {
			log.Warn("applying the MarketBidder root object; still retrying",
				"attempt", attempt, "elapsed", time.Since(started).Round(time.Second).String(),
				"deadline", RootApplyDeadline.String(), "err", err.Error())
			last = time.Now()
		}
		retry := time.NewTimer(min(delay, max(RootApplyDeadline-time.Since(started), 0)))
		select {
		case <-ctx.Done():
			retry.Stop()
			return ctx.Err()
		case <-retry.C:
		}
		delay = min(2*delay, RootRetryMax)
	}
}