package computevm import ( "cmp" "context" "errors" "fmt" "log/slog" "strings" "time" "k8s.io/client-go/rest" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kcpapply" "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube" "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests" "github.com/johnandersen777/socialweb-computer-kcp/internal/hostprocess" "github.com/johnandersen777/socialweb-computer-kcp/internal/hostsvc" "github.com/johnandersen777/socialweb-computer-kcp/internal/kcpserver" "github.com/johnandersen777/socialweb-computer-kcp/internal/openbaoctl" "github.com/johnandersen777/socialweb-computer-kcp/internal/statesync" "github.com/johnandersen777/socialweb-computer-kcp/internal/vault/openbao" "github.com/johnandersen777/socialweb-computer-kcp/internal/vmruntime" ) const ( ChildPrefix = "vm-" LabelComputeVM = "socialweb.computer/computevm" LabelSlot = "socialweb.computer/vm-slot" BidderSet = "bidder" RetryAfter = 3 * time.Second OrphanGrace = 2 * time.Minute MessageCap = 4000 ApplyTimeout = time.Minute ) var GVR = v1alpha1.ComputeVMGVR type Config struct { Kubeconfig string Workspace string Handle string Runtime vmruntime.Runtime Manager hostsvc.Manager OrphanGrace time.Duration VaultServer string VaultClient string VaultMount string Logins *openbao.Logins Log *slog.Logger } type Host struct { c Config Rest *rest.Config VMs kube.Resource Backends kube.Resource HostProcesses kube.Resource servers kube.Resource grants kube.Resource syncs kube.Resource } func (c Config) log() *slog.Logger { if c.Log == nil { return slog.New(slog.DiscardHandler) } return c.Log.With("component", "computevm", "handle", c.Handle) } func (c Config) Grace() time.Duration { return cmp.Or(c.OrphanGrace, OrphanGrace) } func ChildName(name string) string { return ChildPrefix + name } func Open(ctx context.Context, c Config) (*Host, error) { if c.Handle == "" { return nil, errors.New("computevm: the controller names no handle; VM units and work dirs are namespaced by it") } if c.Runtime == nil || c.Manager == nil { return nil, errors.New("computevm: no VM runtime") } if c.Logins == nil { return nil, errors.New("computevm: no OpenBao logins") } c.Workspace = cmp.Or(c.Workspace, hostprocess.DefaultWorkspace) restCfg, err := kcpserver.FrontRestConfig(c.Kubeconfig) if err != nil { return nil, err } if err := applySets(ctx, c, restCfg); err != nil { return nil, err } cfg := rest.CopyConfig(restCfg) cfg.Host = strings.TrimSuffix(cfg.Host, "/") + "/clusters/" + c.Workspace dc, err := kube.New(cfg) if err != nil { return nil, err } h := &Host{c: c, Rest: restCfg, VMs: dc.Resource(GVR), Backends: dc.Resource(v1alpha1.FirecrackerVMGVR), HostProcesses: dc.Resource(hostprocess.GVR), servers: dc.Resource(openbaoctl.GVR), grants: dc.Resource(openbaoctl.GrantGVR), syncs: dc.Resource(statesync.StateSyncGVR)} return h, nil } func (h *Host) Config() Config { return h.c } func ApplyBidderSets(ctx context.Context, c Config, kubeconfig, workspace string) error { restCfg, err := kcpserver.FrontRestConfig(kubeconfig) if err != nil { return err } c.Workspace = cmp.Or(workspace, hostprocess.DefaultWorkspace) return applySets(ctx, c, restCfg) } func (h *Host) Log() *slog.Logger { return h.c.log() } func applySets(ctx context.Context, c Config, restCfg *rest.Config) error { ap, err := kcpapply.New(restCfg, kcpapply.Options{WaitTimeout: ApplyTimeout, Base: c.Workspace, Log: c.log()}) if err != nil { return err } ix, err := manifests.Load() if err != nil { return err } for _, name := range []string{hostprocess.ProviderSet, hostprocess.ManifestSet, BidderSet} { set, ok := ix.Set(name) if !ok { return fmt.Errorf("computevm: deploy/sets.yaml names no %s set", name) } values, err := manifests.DecodeValues(set.Values, nil) if err != nil { return err } if _, err := ap.Apply(ctx, set, values); err != nil { return fmt.Errorf("computevm: applying the %s set: %w", name, err) } } return nil } func CapMessage(s string) string { if len(s) <= MessageCap { return s } return s[:MessageCap/2] + "\n...\n" + s[len(s)-MessageCap/2:] } func Supervisor(c Config) hostprocess.Config { return hostprocess.Config{Kubeconfig: c.Kubeconfig, Workspace: cmp.Or(c.Workspace, hostprocess.DefaultWorkspace), Manager: c.Manager, Prefix: vmruntime.UnitPrefix(c.Handle), Bound: true, Selector: LabelComputeVM, KeepOrphans: true, Name: "computevm", Log: c.Log} }