package marketnetwork import ( "cmp" "context" "errors" "fmt" "os" "path/filepath" "regexp" "slices" "strings" "sync" "time" authenticationv1 "k8s.io/api/authentication/v1" corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/clientcmd" clientcmdapi "k8s.io/client-go/tools/clientcmd/api" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose" "github.com/johnandersen777/socialweb-computer-kcp/internal/assets" "github.com/johnandersen777/socialweb-computer-kcp/internal/bootstrap" "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/kube" "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/kit/fsx" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/jsonx" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/kcprest" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/supervise" "github.com/johnandersen777/socialweb-computer-kcp/internal/lease/local" ) const ( CoreAccount = "app-core" CoreKubeconfig = "parent.kubeconfig" CoreTokenTTL = 30 * 24 * time.Hour CoreTokenRenewAfter = 7 * 24 * time.Hour ChildDeleteTimeout = 2 * time.Minute ChildKillGrace = 5 * time.Second ChildLockTimeout = 2 * time.Minute StoppedMessage = "stopped: its core and its components are down; its workspace and state dir are kept and clearing spec.apps[].stopped starts them again from it" ) var appName = regexp.MustCompile(`^[a-z0-9]([-a-z0-9]*[a-z0-9])?$`) type child struct { app v1alpha1.NetworkApp id string layout devinstance.Layout } func (ctl *Controller) layout() devinstance.Layout { return ctl.c.Config().Layout } func (ctl *Controller) child(app v1alpha1.NetworkApp) (child, error) { if !appName.MatchString(app.Name) { return child{}, fmt.Errorf("apps[].name %q is not a DNS label", app.Name) } if err := checkAppName(app.Name); err != nil { return child{}, err } l := ctl.layout() id := l.ID + "-" + app.Name cl, err := devinstance.NewLayout(id, filepath.Join(l.StateDir, compose.AppsDir), filepath.Join(l.RuntimeDir, compose.AppsDir)) if err != nil { return child{}, err } if _, err := devinstance.ID(id, "", nil); err != nil { return child{}, fmt.Errorf("the child instance id %s: %w", id, err) } return child{app: app, id: id, layout: cl}, nil } func (ctl *Controller) remote(ch child) (*hostprocess.Remote, error) { cfg := ctl.c.Config() return hostprocess.NewRemote(hostprocess.RemoteConfig{Kubeconfig: cfg.Kubeconfig, Workspace: devinstance.ChildWorkspace(ch.id), Reader: cfg.Manager, Log: ctl.log()}) } func (ctl *Controller) coreCredential(ctx context.Context, r compose.Root, ch child) (string, error) { path := filepath.Join(ch.layout.StateDir, CoreKubeconfig) if st, err := os.Stat(path); err == nil && time.Since(st.ModTime()) < CoreTokenRenewAfter { return path, nil } ws := devinstance.ChildWorkspace(ch.id) restCfg := ctl.c.RestConfig() sa := &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: CoreAccount, Namespace: "default"}} if err := ctl.applyRBAC(ctx, r, ws, serviceAccountGVR, sa, "ServiceAccount"); err != nil { return "", fmt.Errorf("the core's ServiceAccount in %s: %w", ws, err) } role := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{Name: CoreAccount}, Rules: []rbacv1.PolicyRule{{ APIGroups: []string{v1alpha1.GroupName}, Resources: []string{"hostprocesses", "hostprocesses/status"}, Verbs: []string{"*"}}}} if err := ctl.applyRBAC(ctx, r, ws, clusterRoleGVR, role, "ClusterRole"); err != nil { return "", fmt.Errorf("the core's ClusterRole in %s: %w", ws, err) } binding := &rbacv1.ClusterRoleBinding{ObjectMeta: metav1.ObjectMeta{Name: CoreAccount}, RoleRef: rbacv1.RoleRef{APIGroup: rbacv1.GroupName, Kind: "ClusterRole", Name: CoreAccount}, Subjects: []rbacv1.Subject{{Kind: rbacv1.ServiceAccountKind, Name: CoreAccount, Namespace: "default"}}} if err := ctl.applyRBAC(ctx, r, ws, clusterRoleBindingGVR, binding, "ClusterRoleBinding"); err != nil { return "", fmt.Errorf("the core's ClusterRoleBinding in %s: %w", ws, err) } cs, err := kubernetes.NewForConfig(kcprest.ForCluster(restCfg, ws)) if err != nil { return "", err } secs := int64(CoreTokenTTL / time.Second) tr, err := cs.CoreV1().ServiceAccounts("default").CreateToken(ctx, CoreAccount, &authenticationv1.TokenRequest{Spec: authenticationv1.TokenRequestSpec{ExpirationSeconds: &secs}}, metav1.CreateOptions{}) if err != nil { return "", fmt.Errorf("minting the core's token in %s: %w", ws, err) } kc := clientcmdapi.NewConfig() kc.Clusters["parent"] = &clientcmdapi.Cluster{Server: restCfg.Host, CertificateAuthorityData: restCfg.CAData, TLSServerName: "127.0.0.1"} kc.AuthInfos[CoreAccount] = &clientcmdapi.AuthInfo{Token: tr.Status.Token} kc.Contexts["parent"] = &clientcmdapi.Context{Cluster: "parent", AuthInfo: CoreAccount} kc.CurrentContext = "parent" if err := os.MkdirAll(ch.layout.StateDir, 0o700); err != nil { return "", err } if err := clientcmd.WriteToFile(*kc, path+".tmp"); err != nil { return "", err } os.Chmod(path+".tmp", 0o600) return path, os.Rename(path+".tmp", path) } func lockChild(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, ChildLockTimeout); err != nil { f.Close() return nil, fmt.Errorf("%s stayed locked: %w", path, err) } return f, nil } func (ctl *Controller) coreLive(ctx context.Context, ch child) bool { var core devinstance.Core if fsx.ReadJSON(ch.layout.CoreRecord, &core) != nil || core.AdoptedAt.IsZero() { return false } st, err := ctl.c.Config().Manager.Inspect(ctx, ch.layout.CoreUnit()) return err == nil && st.State == hostsvc.Active && st.InvocationID == core.InvocationID } func (ctl *Controller) placeChild(ctx context.Context, r compose.Root, ch child, gw *devinstance.GatewayChild) (devinstance.Record, error) { var rec devinstance.Record if fsx.ReadJSON(ch.layout.Record, &rec) == nil && rec.Ready && rec.SharedKcp != "" && ctl.coreLive(ctx, ch) && jsonx.Equal(rec.Gateway, gw) { return rec, nil } cfg := ctl.c.Config() if cfg.Record == nil { return rec, errors.New("marketnetwork: the core has no instance record to place children from") } parent, err := cfg.Record() if err != nil { return rec, err } for _, d := range []string{ch.layout.StateDir, ch.layout.RuntimeDir, ch.layout.Logs} { if err := os.MkdirAll(d, 0o700); err != nil { return rec, err } } if err := ctl.ensureAppWorkspace(ctx, r, ch.app.Name); err != nil { return rec, err } if err := devinstance.EnsureChildWorkspace(ctx, cfg.Kubeconfig, ch.id, devinstance.ChildWorkspaceReady, ctl.log().With("child", ch.id)); err != nil { return rec, err } kubeconfig, err := ctl.coreCredential(ctx, r, ch) if err != nil { return rec, err } lk, err := lockChild(ctx, ch.layout.Lock) if err != nil { return rec, err } defer lk.Close() if fsx.ReadJSON(ch.layout.Record, &rec) != nil { rec = devinstance.Record{Created: time.Now().UTC()} } rec.Layout, rec.Profile, rec.Backend, rec.Devsup = ch.layout, bootstrap.ProfileQuick, hostsvc.BackendProc, parent.Devsup rec.Executables, rec.SourceRoot, rec.BaseWorkspace = parent.Executables, parent.SourceRoot, ctl.appWorkspace(ch.app.Name) rec.SharedKcp, rec.AdminKubeconfig, rec.KcpURL, rec.KineURL, rec.Units = parent.Layout.StateDir, parent.AdminKubeconfig, parent.KcpURL, "", nil rec.OpenBao, rec.Host, rec.Worktree = nil, false, "" if ch.app.Kind == v1alpha1.ApplicationKindMarketRequester { rec.Role = devinstance.RoleRequester } if rec.VM, err = ChildVM(ch.app, ch.layout, parent); err != nil { return rec, err } rec.Gateway = gw rec.Parent = &devinstance.Parent{ID: parent.Layout.ID, Socket: parent.Layout.Socket, Kubeconfig: kubeconfig, Workspace: devinstance.ChildWorkspace(ch.id)} remote, err := ctl.remote(ch) if err != nil { return rec, err } exe := parent.Executables["socialweb-computer"].Path bcfg := bootstrap.Config{Layout: ch.layout, Exe: exe, ReadyTimeout: cfg.ReadyTimeout, Log: ctl.log().With("child", ch.id)} started := time.Now() if err := bootstrap.Ensure(ctx, remote, bcfg, &rec); err != nil { return rec, fmt.Errorf("the bootstrap layer of %s: %w", ch.id, err) } if err := fsx.WriteJSON(ch.layout.Record, rec, 0o600); err != nil { return rec, err } spec, err := bcfg.CoreSpec(&rec) if err != nil { return rec, err } lk.Close() st, err := remote.Start(ctx, spec) if err != nil { return rec, fmt.Errorf("placing the core of %s: %w", ch.id, err) } if err := waitChildCore(ctx, ch.layout, st.InvocationID, cfg.ReadyTimeout); err != nil { return rec, err } fsx.ReadJSON(ch.layout.Record, &rec) ctl.log().Info("a child application runs in its workspace of the network's kcp", "child", ch.id, "kind", ch.app.Kind, "kcp", rec.KcpURL, "workspace", rec.BaseWorkspace, "seconds", time.Since(started).Seconds()) return rec, nil } func waitChildCore(ctx context.Context, l devinstance.Layout, invocation string, timeout time.Duration) error { wctx, cancel := context.WithCancel(ctx) defer cancel() changed, err := local.Watch(wctx, l.CoreRecord) if err != nil { return fmt.Errorf("the core of %s: following its record %s: %w", l.ID, l.CoreRecord, err) } timer := time.NewTimer(timeout) defer timer.Stop() for { var core devinstance.Core if err := fsx.ReadJSON(l.CoreRecord, &core); err == nil && core.InvocationID == invocation && !core.AdoptedAt.IsZero() { return nil } select { case _, ok := <-changed: if !ok { if ctx.Err() != nil { return ctx.Err() } return fmt.Errorf("the core of %s: the watch of its record %s closed", l.ID, l.CoreRecord) } case <-timer.C: b, _ := os.ReadFile(filepath.Join(l.Logs, "core.log")) return fmt.Errorf("the core of %s (invocation %s) did not record its adoption within %s\n%s", l.ID, invocation, timeout, tail(b, 2048)) case <-ctx.Done(): return ctx.Err() } } } func tail(b []byte, n int) []byte { if len(b) > n { return b[len(b)-n:] } return b } func childWorkspace(rec devinstance.Record) string { return cmp.Or(rec.BaseWorkspace, compose.DefaultWorkspace) } func notServed(err error) bool { return apierrors.IsNotFound(err) || apimeta.IsNoMatchError(err) } func (ctl *Controller) markUnserved(app, msg string) { ctl.unservedMu.Lock() defer ctl.unservedMu.Unlock() if ctl.unserved == nil { ctl.unserved = map[string]string{} } ctl.unserved[app] = msg } func (ctl *Controller) takeUnserved() error { ctl.unservedMu.Lock() defer ctl.unservedMu.Unlock() if len(ctl.unserved) == 0 { return nil } names := make([]string, 0, len(ctl.unserved)) for name := range ctl.unserved { names = append(names, name) } slices.Sort(names) parts := make([]string, 0, len(names)) for _, name := range names { parts = append(parts, name+": "+ctl.unserved[name]) } ctl.unserved = nil return fmt.Errorf("the workspace of %s does not serve its root API yet; retried with backoff (%s)", strings.Join(names, ", "), strings.Join(parts, "; ")) } func (ctl *Controller) markUnplaced(app, msg string) { ctl.unservedMu.Lock() defer ctl.unservedMu.Unlock() if ctl.unplaced == nil { ctl.unplaced = map[string]string{} } ctl.unplaced[app] = msg } func (ctl *Controller) takeUnplaced() error { ctl.unservedMu.Lock() defer ctl.unservedMu.Unlock() if len(ctl.unplaced) == 0 { return nil } names := make([]string, 0, len(ctl.unplaced)) for name := range ctl.unplaced { names = append(names, name) } slices.Sort(names) parts := make([]string, 0, len(names)) for _, name := range names { parts = append(parts, name+": "+ctl.unplaced[name]) } ctl.unplaced = nil return fmt.Errorf("%s not placed yet; retried with backoff (%s)", strings.Join(names, ", "), strings.Join(parts, "; ")) } func (ctl *Controller) applyRoot(ctx context.Context, r compose.Root, rec devinstance.Record, kind v1alpha1.ApplicationKind, obj any) error { gvr, err := compose.GVRFor(kind) if err != nil { return err } m, err := compose.ToMap(obj) if err != nil { return err } u := kube.Doc(m) u["apiVersion"], u["kind"] = v1alpha1.GroupVersion.String(), string(kind) meta := compose.DocMeta(u) labels := r.Labels(string(kind)) if own, ok := meta["labels"].(map[string]any); ok { for k, v := range own { if s, ok := v.(string); ok { labels[k] = s } } } meta["labels"] = labels return ctl.c.Apply(ctx, r, childWorkspace(rec), gvr, u) } func (ctl *Controller) readRoot(ctx context.Context, rec devinstance.Record, kind v1alpha1.ApplicationKind) (v1alpha1.ApplicationStatus, bool, error) { gvr, err := compose.GVRFor(kind) if err != nil { return v1alpha1.ApplicationStatus{}, false, err } u, err := compose.Get[compose.Generic](ctx, ctl.c, childWorkspace(rec), gvr, compose.ChildRootName) if apierrors.IsNotFound(err) { return v1alpha1.ApplicationStatus{}, false, nil } if err != nil { return v1alpha1.ApplicationStatus{}, false, err } st, err := compose.ApplicationStatusOf(u) return st, true, err } func (ctl *Controller) liveUnits(ctx context.Context, ch child) ([]hostsvc.UnitStatus, error) { units, err := devinstance.ChildUnits(ctx, ctl.c.Config().Manager, ch.id) if err != nil { return nil, err } var live []hostsvc.UnitStatus for _, u := range units { if unitLive(u) { live = append(live, u) } } return live, nil } func unitLive(u hostsvc.UnitStatus) bool { return u.State == hostsvc.Active || u.State == hostsvc.Activating || u.State == hostsvc.Deactivating } func (ctl *Controller) awaitUnitsDown(ctx context.Context, ch child, timeout time.Duration) ([]hostsvc.UnitStatus, bool, error) { m := ctl.c.Config().Manager watches := supervise.New(ctx) defer func() { _ = watches.Stop() }() wctx := watches.Context() events := make(chan hostsvc.UnitEvent) closed := make(chan struct{}, len(devinstance.Prefixes(ch.id))) for _, prefix := range devinstance.Prefixes(ch.id) { src, err := m.Watch(wctx, prefix) if err != nil { return nil, false, fmt.Errorf("following the units of %s: %w", ch.id, err) } watches.Go("marketnetwork: units "+prefix, func(ctx context.Context) error { for ev := range src { select { case events <- ev: case <-ctx.Done(): return nil } } closed <- struct{}{} return nil }) } units, err := devinstance.ChildUnits(wctx, m, ch.id) if err != nil { return nil, false, err } state := map[string]hostsvc.UnitStatus{} for _, u := range units { state[u.Name] = u } timer := time.NewTimer(timeout) defer timer.Stop() for { var live []hostsvc.UnitStatus for _, u := range state { if unitLive(u) { live = append(live, u) } } if len(live) == 0 { return nil, false, nil } slices.SortFunc(live, func(a, b hostsvc.UnitStatus) int { return strings.Compare(a.Name, b.Name) }) select { case ev := <-events: if ev.Status.State == hostsvc.NotFound { delete(state, ev.Name) continue } st := ev.Status st.Name = cmp.Or(st.Name, ev.Name) state[ev.Name] = st case <-closed: if ctx.Err() != nil { return live, false, ctx.Err() } return live, false, fmt.Errorf("the watch of the units of %s closed with %d unit(s) still running", ch.id, len(live)) case <-timer.C: return live, true, nil case <-ctx.Done(): return live, false, ctx.Err() } } } func (ctl *Controller) stopChild(ctx context.Context, ch child) error { live, err := ctl.liveUnits(ctx, ch) if err != nil { return err } if len(live) == 0 { KillUnder(ch.layout.StateDir, ChildKillGrace) return nil } log := ctl.log().With("child", ch.id) if err := ctl.stopAppComponents(ctx, ch.app.Name); err != nil { return err } if err := devinstance.DeleteChildWorkspace(ctx, ctl.c.Config().Kubeconfig, ch.id); err != nil { return err } live, timedOut, err := ctl.awaitUnitsDown(ctx, ch, ChildDeleteTimeout) if err != nil { return err } if timedOut { for _, u := range live { if err := ctl.c.Config().Manager.Stop(ctx, u.Name); err != nil { log.Warn("stopping a unit the supervisor left running", "unit", u.Name, "err", err.Error()) } } } if pids := KillUnder(ch.layout.StateDir, ChildKillGrace); len(pids) > 0 { log.Warn("stopped processes of the child its units left behind", "pids", fmt.Sprint(pids)) } if err := os.Remove(filepath.Join(ch.layout.StateDir, CoreKubeconfig)); err != nil && !os.IsNotExist(err) { return err } log.Info("a child application is stopped; its state dir is kept", "state", ch.layout.StateDir) return nil } func (ctl *Controller) teardownChild(ctx context.Context, ch child) error { log := ctl.log().With("child", ch.id) var rec devinstance.Record recErr := fsx.ReadJSON(ch.layout.Record, &rec) if rec.Gateway != nil { if err := ctl.deleteAccountWorkspaces(ctx); err != nil { log.Warn("the gateway's account workspaces did not go; its workspace is deleted anyway", "err", err.Error()) } } if recErr == nil && rec.AdminKubeconfig != "" && ctl.coreLive(ctx, ch) { for _, kind := range []v1alpha1.ApplicationKind{v1alpha1.ApplicationKindMarketBidder, v1alpha1.ApplicationKindMarketRequester, v1alpha1.ApplicationKindGateway, v1alpha1.ApplicationKindSshGateway} { gvr, err := compose.GVRFor(kind) if err != nil { continue } ri, err := ctl.c.Client(childWorkspace(rec), gvr) if err != nil { continue } if err := ri.Delete(ctx, compose.ChildRootName, metav1.DeleteOptions{}); err != nil { continue } if err := ctl.waitGone(ctx, childWorkspace(rec), gvr, compose.ChildRootName, "the child's "+string(kind)); err != nil { log.Warn("the child's root object did not go within the timeout; its workspace is deleted anyway", "kind", kind, "err", err.Error()) } } } if err := ctl.deleteAppWorkspace(ctx, ch.app.Name); err != nil { log.Warn("the child's workspace did not go; its core is stopped anyway", "workspace", ctl.appWorkspace(ch.app.Name), "err", err.Error()) } if err := devinstance.DeleteChildWorkspace(ctx, ctl.c.Config().Kubeconfig, ch.id); err != nil { return err } live, timedOut, err := ctl.awaitUnitsDown(ctx, ch, ChildDeleteTimeout) if err != nil { return err } if timedOut { return fmt.Errorf("%d unit(s) of %s still run after its workspace was deleted", len(live), ch.id) } if pids := KillUnder(ch.layout.StateDir, ChildKillGrace); len(pids) > 0 { log.Warn("stopped processes of the child its units left behind", "pids", fmt.Sprint(pids)) } err = errors.Join(assets.RemoveTree(ch.layout.StateDir), assets.RemoveTree(ch.layout.RuntimeDir)) log.Info("a child application is gone: its root object, its workspace, its core and its state", "err", fmt.Sprint(err)) return err } func (ctl *Controller) knownChildren() []string { entries, err := os.ReadDir(devinstance.DevRoot(filepath.Join(ctl.layout().StateDir, compose.AppsDir))) if err != nil { return nil } var out []string for _, e := range entries { if e.IsDir() { out = append(out, e.Name()) } } return out } func (ctl *Controller) childByID(id string) child { l := ctl.layout() name := strings.TrimPrefix(id, l.ID+"-") ch, err := ctl.child(v1alpha1.NetworkApp{Name: name}) if err != nil { cl, _ := devinstance.NewLayout(id, filepath.Join(l.StateDir, compose.AppsDir), filepath.Join(l.RuntimeDir, compose.AppsDir)) return child{id: id, layout: cl} } return ch } func (ctl *Controller) teardownAll(ctx context.Context, _ compose.Root, keep []string) error { var wg sync.WaitGroup var mu sync.Mutex var errs []string for _, id := range ctl.knownChildren() { if slices.Contains(keep, id) { continue } ch := ctl.childByID(id) wg.Go(func() { if err := ctl.teardownChild(ctx, ch); err != nil { mu.Lock() errs = append(errs, ch.id+": "+err.Error()) mu.Unlock() } }) } wg.Wait() if len(errs) > 0 { return fmt.Errorf("tearing children down: %s", strings.Join(errs, "; ")) } return nil } func (ctl *Controller) ensureApp(ctx context.Context, r compose.Root, app v1alpha1.NetworkApp, value v1alpha1.MarketNetworkValue, comms *v1alpha1.MarketComms, pdsURL, aliceURL string, bidderDids []string, suspend, placedOnly bool) v1alpha1.NetworkAppStatus { out := v1alpha1.NetworkAppStatus{Name: app.Name, Kind: app.Kind, Phase: v1alpha1.ApplicationPhaseProvisioning} fail := func(err error) v1alpha1.NetworkAppStatus { out.Message = err.Error() if notServed(err) { ctl.markUnserved(app.Name, err.Error()) } ctl.log().Warn("a child application is not ready yet", "app", app.Name, "err", err.Error()) return out } ch, err := ctl.child(app) if err != nil { out.Phase = v1alpha1.ApplicationPhaseFailed return fail(err) } out.Instance = ch.id if app.Stopped { if err := ctl.stopChild(ctx, ch); err != nil { return fail(err) } out.Phase, out.Message = v1alpha1.ApplicationPhasePending, StoppedMessage return out } if _, err := compose.GVRFor(app.Kind); err != nil { out.Phase = v1alpha1.ApplicationPhaseFailed return fail(err) } if placedOnly { var rec devinstance.Record if fsx.ReadJSON(ch.layout.Record, &rec) != nil || rec.AdminKubeconfig == "" || !ctl.coreLive(ctx, ch) { out.Phase, out.Message = v1alpha1.ApplicationPhaseSuspended, compose.SuspendedMessage return out } } var gw *devinstance.GatewayChild if app.Kind == v1alpha1.ApplicationKindGateway { if gw, err = GatewayChildOf(ctx, ctl.c, r, ch.id, ch.layout.StateDir, app, value, aliceURL, bidderDids, suspend); err != nil { return fail(err) } } rec, err := ctl.placeChild(ctx, r, ch, gw) if err != nil { if ctx.Err() == nil { ctl.markUnplaced(app.Name, err.Error()) } return fail(err) } out.KcpURL, out.Workspace = rec.KcpURL, rec.BaseWorkspace meta := metav1.ObjectMeta{Name: compose.ChildRootName} var root any switch app.Kind { case v1alpha1.ApplicationKindMarketBidder: var fc *v1alpha1.FirecrackerProvider if app.Firecracker != nil { f := *app.Firecracker fc = &f } var kcp *v1alpha1.KcpProvider if app.Kcp != nil { k := *app.Kcp k.Handle = KcpHandle(app) kcp = &k fc = nil } root = &v1alpha1.MarketBidder{ObjectMeta: meta, Spec: v1alpha1.MarketBidderSpec{Profile: app.Profile, Suspend: suspend, Network: &value, Identity: v1alpha1.ApplicationIdentity{Handle: app.Name + "." + BidderAccountDomain, PdsURL: pdsURL}, Policy: app.Policy, GuestNetwork: app.GuestNetwork, Firecracker: fc, Kcp: kcp, Options: app.Options, Quota: app.Quota, Comms: comms, Components: app.Components, BidderDids: append([]string(nil), bidderDids...), ComputeKinds: append([]string(nil), app.ComputeKinds...)}} case v1alpha1.ApplicationKindMarketRequester: root = &v1alpha1.MarketRequester{ObjectMeta: meta, Spec: v1alpha1.MarketRequesterSpec{Profile: app.Profile, Suspend: suspend, Network: &value, Identity: compose.NetworkRequesterIdentity(app.Options, pdsURL), BidderDids: bidderDids, Options: app.Options, Quota: app.Quota, Comms: comms, Components: app.Components, SecretsFallback: app.SecretsFallback, RequesterIngress: app.RequesterIngress}} case v1alpha1.ApplicationKindSshGateway: root = &v1alpha1.SshGateway{ObjectMeta: meta, Spec: v1alpha1.SshGatewaySpec{Profile: app.Profile, Suspend: suspend, Network: &value, BidderDids: bidderDids, Options: app.Options, Quota: app.Quota, OwnerSessionsFile: app.OwnerSessionsFile, Comms: comms, Components: app.Components, RequesterIngress: app.RequesterIngress}} } if root != nil { if err := ctl.applyRoot(ctx, r, rec, app.Kind, root); err != nil { return fail(fmt.Errorf("applying the root object into %s's workspace %s: %w", ch.id, rec.BaseWorkspace, err)) } } rs, found, err := ctl.readRoot(ctx, rec, app.Kind) if err != nil { return fail(err) } if !found { ctl.markUnserved(app.Name, fmt.Sprintf("the root object of %s is not readable yet", ch.id)) return fail(fmt.Errorf("the root object of %s is not readable yet", ch.id)) } out.Phase, out.Message, out.Did, out.Components, out.Listeners = rs.Phase, rs.Message, rs.Did, rs.Components, rs.Listeners out.Phase = cmp.Or(out.Phase, v1alpha1.ApplicationPhasePending) return out }