johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package marketnetwork

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

	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime/schema"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	ctlruntime "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/jsonx"
)

const (
	Name = "marketnetwork"

	Kind = "MarketNetwork"

	NetworkSet          = "market-network"
	NetworkProfile      = "market-network"
	BidderAccountDomain = "bob.test"
	NetworkServiceState = "state"
	NetworkPortsFile    = "ports.json"
	NetworkPortFirst    = 30000
	NetworkPortLast     = 39999
	NetworkPortBlock    = 8
	FrontRebindAfter    = 30 * time.Second

	NetworkAlicePdsName      = "alice"
	NetworkBobPdsName        = "bob"
	NetworkDispatcherTLSName = "dispatcher-tls"

	NetworkPlcHost        = "plc.default.global.svc.kcp.local"
	NetworkRelayHost      = v1alpha1.NetworkRelayHost
	NetworkIngressHost    = "dispatcher.default.relay.svc.kcp.local"
	NetworkDispatcherHost = v1alpha1.NetworkDispatcherHost
	NetworkAlicePdsHost   = "pds.default.alice.svc.kcp.local"
	NetworkBobPdsHost     = "pds.default.bob.svc.kcp.local"
	NetworkPortPlc        = "Global/plc"
	NetworkPortRelay      = "Relay/relay"
	NetworkPortIngress    = "IngressProxy/http"
	NetworkPortDispatcher = "DispatcherTLS/tls"
	NetworkPortAlicePds   = "AlicePds/http"
	NetworkPortBobPds     = "BobPds/http"

	ConditionServicesReady   = "ServicesReady"
	ConditionVerifierPassed  = "VerifierPassed"
	ConditionOneKcp          = "OneKcp"
	ConditionChildrenApplied = "ChildrenApplied"

	VerifierProbeTimeout = 5 * time.Second

	BidderIdentity    = "bidder"
	RequesterIdentity = "requester"
)

var networkPortNames = []string{NetworkPortPlc, NetworkPortRelay, NetworkPortIngress, NetworkPortDispatcher, NetworkPortAlicePds, NetworkPortBobPds}

type Controller struct {
	c *compose.Composer

	mu sync.Mutex

	pkiPEM []byte

	peers string

	value *v1alpha1.MarketNetworkValue

	valueUID string

	unservedMu sync.Mutex

	unserved map[string]string

	unplaced map[string]string

	roots sync.Map
}

func Register(m *ctlruntime.Manager, c *compose.Composer) error {
	ctl := &Controller{c: c}
	c.ProvideTrust(schema.GroupKind{Group: v1alpha1.GroupName, Kind: Kind})
	opts := append([]ctlruntime.Option{ctlruntime.Export(compose.ExportApplications), ctlruntime.Finalizer(compose.Finalizer)}, c.Watches(Kind)...)
	opts = append(opts, childRootWatch[*v1alpha1.MarketBidder](ctl), childRootWatch[*v1alpha1.MarketRequester](ctl),
		childRootWatch[*v1alpha1.SshGateway](ctl), childRootWatch[*v1alpha1.Gateway](ctl))
	if err := ctlruntime.For[*v1alpha1.MarketNetwork](m, Name, ctl, opts...); err != nil {
		return fmt.Errorf("marketnetwork: registering the MarketNetwork controller: %w", err)
	}
	return nil
}

func childRootWatch[T ctlruntime.Object](ctl *Controller) ctlruntime.Option {
	return ctlruntime.Watches(func(_ context.Context, src ctlruntime.Request, obj T) []ctlruntime.Request {
		if obj.GetName() != compose.ChildRootName || !strings.HasPrefix(src.Workspace, ctl.c.Workspace()+":") {
			return nil
		}
		var out []ctlruntime.Request
		ctl.roots.Range(func(_, v any) bool {
			out = append(out, v.(ctlruntime.Request))
			return true
		})
		return out
	})
}

func (ctl *Controller) log() *slog.Logger { return ctl.c.Log().With("root", Name) }

type pass struct {
	ctx context.Context

	ctl *Controller

	c *compose.Composer

	r compose.Root

	o *v1alpha1.MarketNetwork

	st v1alpha1.MarketNetworkStatus
}

func (p *pass) write(phase v1alpha1.ApplicationPhase, msg string) (ctlruntime.Result, error) {
	p.st.Phase, p.st.Message = phase, msg
	return ctlruntime.Result{}, p.c.WriteStatus(p.ctx, p.r, compose.MarketNetworkGVR, p.st)
}

func (p *pass) retry(phase v1alpha1.ApplicationPhase, what string, err error) (ctlruntime.Result, error) {
	msg := err.Error()
	if what != "" {
		msg = what + ": " + msg
	}
	if _, werr := p.write(phase, msg); werr != nil {
		return ctlruntime.Result{}, errors.Join(err, werr)
	}
	return ctlruntime.Result{}, err
}

func (p *pass) waiting(msg string) (ctlruntime.Result, error) {
	p.st.Phase, p.st.Message = v1alpha1.ApplicationPhaseDeleting, msg
	return ctlruntime.Result{Waiting: true}, p.c.WriteStatus(p.ctx, p.r, compose.MarketNetworkGVR, p.st)
}

func (ctl *Controller) Reconcile(ctx context.Context, req ctlruntime.Request, o *v1alpha1.MarketNetwork) (ctlruntime.Result, error) {
	c := ctl.c
	if !c.Mine(req) {
		return ctlruntime.Result{}, nil
	}
	ctl.mu.Lock()
	defer ctl.mu.Unlock()
	r := compose.RootOf(req, Kind, o, o.Spec.Suspend)
	c.Observe(r)
	ctl.roots.Store(string(r.UID), ctlruntime.Request{Cluster: req.Cluster, Name: req.Name})
	p := &pass{ctx: ctx, ctl: ctl, c: c, r: r, o: o, st: v1alpha1.MarketNetworkStatus{ObservedGeneration: o.Generation, AppliedSet: o.Status.AppliedSet, Network: o.Status.Network,
		BidderPdsURL: o.Status.BidderPdsURL, Apps: o.Status.Apps, Conditions: append([]metav1.Condition(nil), o.Status.Conditions...)}}
	if r.Deleting {
		return p.delete(ctx)
	}
	res, err := p.reconcile(ctx)
	if err == nil {
		c.Adopt(ctx, r)
		c.ApplyQuotas(ctx, r, nil)
	}
	return res, err
}

func (p *pass) delete(ctx context.Context) (ctlruntime.Result, error) {
	if err := p.ctl.teardownAll(ctx, p.r, nil); err != nil {
		res, werr := p.waiting(err.Error())
		return res, errors.Join(err, werr)
	}
	left, err := p.c.StopComponents(ctx, p.r, "")
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if left > 0 {
		return p.waiting(fmt.Sprintf("%d component(s) stopping", left))
	}
	vaults, err := p.c.DeleteVaults(ctx, p.r)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	if vaults > 0 {
		return p.waiting("the OpenBaoServer " + NetworkVaultServer + " is stopping")
	}
	if p.ctl.valueUID == string(p.r.UID) {
		p.ctl.value, p.ctl.valueUID = nil, ""
	}
	p.ctl.pkiPEM = nil
	p.ctl.roots.Delete(string(p.r.UID))
	p.c.Drop(p.r)
	p.ctl.log().Info("the market network is gone: children, components and its OpenBaoServer")
	return ctlruntime.Result{}, nil
}

func (p *pass) ensure(ctx context.Context, desired []compose.Desired) error {
	return p.c.EnsureLinked(ctx, p.r, desired, desired)
}

func (p *pass) prune(ctx context.Context, desired []compose.Desired) error {
	_, err := p.c.DeleteComponentsExcept(ctx, p.r.Workspace, p.r.Owner(), compose.DesiredKeys(desired))
	return err
}

func (p *pass) snapshot(ctx context.Context) (compose.Snapshot, error) {
	snap, err := p.c.Snapshot(ctx, p.r.Workspace, p.r.Owner())
	if err != nil {
		return snap, err
	}
	p.st.Components, p.st.Listeners = snap.Components, snap.Listeners
	return snap, nil
}

func (p *pass) fill(desired []compose.Desired, snap compose.Snapshot) {
	compose.FillComponentStatus(&p.st.Conditions, &p.st.ComponentSummary, &p.st.ComponentOverrides, p.o.Generation, p.r.Suspend,
		p.o.Spec.Components, desired, snap, nil)
}

func (p *pass) reconcile(ctx context.Context) (ctlruntime.Result, error) {
	c, r, o, ctl := p.c, p.r, p.o, p.ctl
	if o.Spec.Profile != NetworkProfile {
		return p.write(v1alpha1.ApplicationPhaseFailed, fmt.Sprintf("profile %q: the built MarketNetwork profile is %s", o.Spec.Profile, NetworkProfile))
	}
	ws := r.Workspace
	stateRoot := c.StateRoot()
	if err := c.ApplyVault(ctx, r, NetworkVaultServer); err != nil {
		return ctlruntime.Result{}, err
	}
	intact := true
	if !r.Suspend {
		var err error
		if intact, err = c.CheckWorkloads(ctx, r)(c.PhaseWorkloads(NetworkSet, &manifests.MarketNetwork{}, "vault")); err != nil {
			return p.retry(v1alpha1.ApplicationPhaseProvisioning, "its set workloads", err)
		}
	}
	if !c.Applied(r) {
		digest, digestErr := compose.PhasesDigest(NetworkSet, &manifests.MarketNetwork{}, nil, "vault")
		if intact && digestErr == nil && c.AppliedBefore(r, o.Status.AppliedSet, digest) {
			c.MarkApplied(r)
		} else {
			if _, err := p.write(v1alpha1.ApplicationPhaseProvisioning, "applying the network's OpenBaoServer"); err != nil {
				return ctlruntime.Result{}, err
			}
			if err := c.ApplySetPhases(ctx, r, NetworkSet, &manifests.MarketNetwork{}, "vault"); err != nil {
				return p.retry(v1alpha1.ApplicationPhaseProvisioning, "the network's OpenBaoServer", err)
			}
			c.MarkApplied(r)
			if digestErr == nil {
				p.st.AppliedSet = &v1alpha1.AppliedSet{Generation: o.Generation, Digest: digest}
			}
			ctl.log().Info("applied the market network's OpenBaoServer", "set", NetworkSet)
		}
	}
	if err := ctl.vaultReady(ctx, ws); err != nil {
		return p.write(v1alpha1.ApplicationPhaseProvisioning, "waiting for the network's OpenBaoServer: "+err.Error())
	}
	if err := c.ApplyLinks(ctx, r, "root", vaultLinks(r)); err != nil {
		return ctlruntime.Result{}, err
	}
	if err := EnsureGuestTLS(filepath.Join(NetworkStateDir(stateRoot), GuestTLSDir)); err != nil {
		return p.write(v1alpha1.ApplicationPhaseFailed, err.Error())
	}
	leases, err := ctl.portLeases()
	if err != nil {
		return p.write(v1alpha1.ApplicationPhaseFailed, err.Error())
	}
	bundle, err := ctl.caBundle(ctx, r)
	if err != nil {
		return p.retry(v1alpha1.ApplicationPhaseProvisioning, "the network CA bundle", err)
	}
	c.PublishTrust(r.Ref(), []byte(bundle))
	value := networkValueFromLeases(leases, bundle)
	if ctl.value != nil && ctl.valueUID == string(r.UID) {
		value = *ctl.value
	}
	netIngress := compose.NetworkIngressConfig(o.Spec.Ingress, ws)
	value.Ingress = netIngress
	if err := compose.NetworkIngressSelection(netIngress); err != nil {
		compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, NetworkAlicePdsName, true, err)
		compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, NetworkBobPdsName, true, err)
		return p.write(v1alpha1.ApplicationPhaseFailed, "the network Pds ingress: "+err.Error())
	}
	for _, name := range []string{NetworkAlicePdsName, NetworkBobPdsName} {
		if err := compose.EnsurePDSKeyFiles(compose.ComponentCRStateDir(NetworkStateDir(stateRoot), compose.ComponentKindPds, name)); err != nil {
			return p.write(v1alpha1.ApplicationPhaseFailed, "the "+name+" PDS keys: "+err.Error())
		}
	}
	alicePublicURL, aliceIngressErr := c.PdsPublicURL(ctx, ws, NetworkAlicePdsName, true)
	bobPublicURL, bobIngressErr := c.PdsPublicURL(ctx, ws, NetworkBobPdsName, true)
	for _, item := range []struct {
		name string
		err  error
	}{{NetworkAlicePdsName, aliceIngressErr}, {NetworkBobPdsName, bobIngressErr}} {
		if item.err == nil || compose.PdsIngressReason(item.err) == compose.PdsIngressReasonNotReady {
			continue
		}
		compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, item.name, true, item.err)
		return p.write(v1alpha1.ApplicationPhaseFailed, "the "+item.name+" PDS ingress: "+item.err.Error())
	}
	compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, NetworkAlicePdsName, true, aliceIngressErr)
	compose.SetPdsIngressCondition(&p.st.Conditions, o.Generation, NetworkBobPdsName, true, bobIngressErr)
	desired, err := networkDesired(o, ws, stateRoot, leases, value, alicePublicURL, bobPublicURL)
	if err != nil {
		compose.SetOverrideErrorIfAny(&p.st.Conditions, &p.st.ComponentSummary, o.Generation, err)
		return p.write(v1alpha1.ApplicationPhaseFailed, err.Error())
	}
	if err := c.ApplyRelayStatus(ctx, r, compose.NetworkRelayName); err != nil {
		return ctlruntime.Result{}, err
	}
	applyDesired := desired
	if alicePublicURL == "" || bobPublicURL == "" {
		appliedAlice, err := c.PdsPublicURLApplied(ctx, ws, NetworkAlicePdsName)
		if err != nil {
			return ctlruntime.Result{}, err
		}
		appliedBob, err := c.PdsPublicURLApplied(ctx, ws, NetworkBobPdsName)
		if err != nil {
			return ctlruntime.Result{}, err
		}
		if appliedAlice && appliedBob {
			applyDesired = compose.WithoutPdsNamed(desired, NetworkAlicePdsName, NetworkBobPdsName)
		}
	}
	if r.Suspend {
		return p.suspended(ctx, desired, applyDesired)
	}
	if err := p.c.EnsureLinked(ctx, p.r, desired, applyDesired); err != nil {
		return p.retry(v1alpha1.ApplicationPhaseProvisioning, "", err)
	}
	if err := p.prune(ctx, desired); err != nil {
		return ctlruntime.Result{}, err
	}
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	p.fill(desired, snap)
	if !snap.Satisfied(desired) {
		var waiting []string
		for _, d := range desired {
			if !compose.IsExposureKind(d.Kind) && !snap.Ready(d.Kind, d.Name) {
				waiting = append(waiting, compose.ComponentKey(d.Kind, d.Name))
			}
		}
		conds.Set(&p.st.Conditions, o.Generation, ConditionServicesReady, false, "Waiting", strings.Join(waiting, "; "))
		return p.write(v1alpha1.ApplicationPhaseProvisioning, "waiting for the network's components to report ready")
	}
	if alicePublicURL == "" || bobPublicURL == "" {
		return p.write(v1alpha1.ApplicationPhaseProvisioning, "waiting for the network's PDS ingress URLs")
	}
	published, err := networkValueFromExposures(snap, bundle)
	if err != nil {
		if ctl.value == nil || ctl.valueUID != string(r.UID) {
			conds.Set(&p.st.Conditions, o.Generation, ConditionServicesReady, false, "Endpoints", err.Error())
			return p.write(v1alpha1.ApplicationPhaseProvisioning, "the network's Exposure endpoints: "+err.Error())
		}
		published = *ctl.value
	} else {
		published.Ingress = netIngress
		if ctl.value == nil || ctl.valueUID != string(r.UID) || !jsonx.Equal(*ctl.value, published) {
			ctl.value, ctl.valueUID = &published, string(r.UID)
			if desired, err = networkDesired(o, ws, stateRoot, leases, published, alicePublicURL, bobPublicURL); err != nil {
				compose.SetOverrideErrorIfAny(&p.st.Conditions, &p.st.ComponentSummary, o.Generation, err)
				return p.write(v1alpha1.ApplicationPhaseFailed, err.Error())
			}
			if err := p.ensure(ctx, desired); err != nil {
				return p.retry(v1alpha1.ApplicationPhaseProvisioning, "", err)
			}
			if err := p.prune(ctx, desired); err != nil {
				return ctlruntime.Result{}, err
			}
		}
	}
	published.Ingress = netIngress
	p.st.Network = &published
	if o.Status.Network == nil || !jsonx.Equal(*o.Status.Network, published) {
		p.st.Phase, p.st.Message = v1alpha1.ApplicationPhaseProvisioning, "publishing the network endpoints before the child applications start"
		if err := c.WriteStatus(ctx, r, compose.MarketNetworkGVR, p.st); err != nil {
			return ctlruntime.Result{}, err
		}
	}
	pdsURL, _ := snap.ExposureURL(compose.ServiceNameFor(compose.ComponentKindPds, NetworkBobPdsName) + compose.ExposureLocalSuffix)
	aliceURL, _ := snap.ExposureURL(compose.ServiceNameFor(compose.ComponentKindPds, NetworkAlicePdsName) + compose.ExposureLocalSuffix)
	if pdsURL == "" {
		pdsURL = networkValueHostURL(published, NetworkBobPdsHost)
	}
	if aliceURL == "" {
		aliceURL = networkValueHostURL(published, NetworkAlicePdsHost)
	}
	if pdsURL == "" || aliceURL == "" {
		return p.write(v1alpha1.ApplicationPhaseProvisioning, "waiting for the network's PDS Exposure URLs")
	}
	p.st.BidderPdsURL = pdsURL
	var keep []string
	for _, app := range o.Spec.Apps {
		if ch, err := ctl.child(app); err == nil {
			keep = append(keep, ch.id)
		}
	}
	if err := ctl.teardownAll(ctx, r, keep); err != nil {
		ctl.log().Warn("removing children the spec no longer names", "err", err.Error())
	}
	apps := ctl.ensureApps(ctx, r, o, published, pdsURL, aliceURL, false)
	unserved := ctl.takeUnserved()
	unplaced := ctl.takeUnplaced()
	ctl.declareOperatorVouches(ctx, r, o.Spec.Apps)
	p.st.Apps = apps
	names := make([]string, 0, len(o.Spec.Apps))
	for _, a := range o.Spec.Apps {
		names = append(names, a.Name)
	}
	if err := ctl.grantPeers(ctx, r, names); err != nil {
		ctl.log().Warn("granting the child applications reads of each other's MarketIdentity objects", "err", err.Error())
	}
	conds.Set(&p.st.Conditions, o.Generation, ConditionServicesReady, true, "Ready",
		fmt.Sprintf("%d component(s) are ready and report their endpoints", len(snap.Views)))
	if wrong := ctl.oneKcp(ws, apps); len(wrong) > 0 {
		conds.Set(&p.st.Conditions, o.Generation, ConditionOneKcp, false, "Apart", strings.Join(wrong, "; "))
	} else {
		conds.Set(&p.st.Conditions, o.Generation, ConditionOneKcp, true, "Workspaces", "every child application runs in its own workspace of the network's kcp")
	}
	if unserved != nil {
		conds.Set(&p.st.Conditions, o.Generation, ConditionChildrenApplied, false, "RootNotServed", unserved.Error())
	} else {
		conds.Set(&p.st.Conditions, o.Generation, ConditionChildrenApplied, true, "Applied", "every child application's root object is applied in its workspace")
	}
	verr := ctl.verify(ctx, snap, apps)
	if verr != nil {
		conds.Set(&p.st.Conditions, o.Generation, ConditionVerifierPassed, false, "Probe", verr.Error())
	}
	if unserved != nil {
		return p.retry(v1alpha1.ApplicationPhaseProvisioning, "the child applications", unserved)
	}
	if unplaced != nil {
		return p.retry(v1alpha1.ApplicationPhaseProvisioning, "the child applications", unplaced)
	}
	if verr != nil {
		return p.write(v1alpha1.ApplicationPhaseProvisioning, "the network verifier: "+verr.Error())
	}
	conds.Set(&p.st.Conditions, o.Generation, ConditionVerifierPassed, true, "Verified",
		"the PLC, relay, PDS, ingress and child application endpoints answer")
	for _, t := range []string{ConditionServicesReady, ConditionOneKcp, ConditionChildrenApplied, ConditionVerifierPassed} {
		if cond := meta.FindStatusCondition(p.st.Conditions, t); cond == nil || cond.Status != metav1.ConditionTrue {
			return p.write(v1alpha1.ApplicationPhaseProvisioning, "the child applications are ready; waiting for condition "+t)
		}
	}
	return p.write(v1alpha1.ApplicationPhaseReady, fmt.Sprintf("%d component(s) and %d child application(s), each in its own workspace of one kcp, are ready", len(snap.Views), len(apps)))
}

func (p *pass) suspended(ctx context.Context, desired, applyDesired []compose.Desired) (ctlruntime.Result, error) {
	c, r, o, ctl := p.c, p.r, p.o, p.ctl
	snap, err := p.snapshot(ctx)
	if err != nil {
		return ctlruntime.Result{}, err
	}
	held := compose.Existing(snap, applyDesired)
	if err := p.ensure(ctx, held); err != nil {
		return p.retry(v1alpha1.ApplicationPhaseSuspended, "", err)
	}
	p.fill(compose.Existing(snap, desired), snap)
	value := o.Status.Network
	if ctl.value != nil && ctl.valueUID == string(r.UID) {
		value = ctl.value
	}
	if value != nil {
		pdsURL, aliceURL := o.Status.BidderPdsURL, networkValueHostURL(*value, NetworkAlicePdsHost)
		if u, ok := snap.ExposureURL(compose.ServiceNameFor(compose.ComponentKindPds, NetworkAlicePdsName) + compose.ExposureLocalSuffix); ok {
			aliceURL = u
		}
		p.st.Apps = ctl.ensureApps(ctx, r, o, *value, pdsURL, aliceURL, true)
	}
	conds.Set(&p.st.Conditions, o.Generation, ConditionVerifierPassed, false, compose.ReasonSuspended, "the network is suspended")
	return ctlruntime.Result{}, c.WriteStatus(ctx, r, compose.MarketNetworkGVR, p.suspendStatus())
}

func (p *pass) suspendStatus() v1alpha1.MarketNetworkStatus {
	st := p.st
	st.Phase, st.Message = v1alpha1.ApplicationPhaseSuspended, compose.SuspendedMessage
	return st
}

func (ctl *Controller) ensureApps(ctx context.Context, r compose.Root, o *v1alpha1.MarketNetwork, value v1alpha1.MarketNetworkValue,
	pdsURL, aliceURL string, placedOnly bool) []v1alpha1.NetworkAppStatus {
	var bidderDids []string
	known := map[string]string{}
	for _, a := range o.Status.Apps {
		known[a.Name] = a.Did
		if a.Kind == v1alpha1.ApplicationKindMarketBidder && a.Did != "" {
			bidderDids = append(bidderDids, a.Did)
		}
	}
	ctl.takeUnplaced()
	apps := make([]v1alpha1.NetworkAppStatus, len(o.Spec.Apps))
	var wg sync.WaitGroup
	for i, app := range o.Spec.Apps {
		wg.Go(func() {
			apps[i] = ctl.ensureApp(ctx, r, app, value, o.Spec.Comms, pdsURL, aliceURL, bidderDids, o.Spec.Suspend || app.Suspend, placedOnly)
			apps[i].Did = cmp.Or(apps[i].Did, known[app.Name])
		})
	}
	wg.Wait()
	return apps
}