johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package marketnetwork

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strings"
	"time"

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

	rbacv1 "k8s.io/api/rbac/v1"

	tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1"

	"github.com/johnandersen777/socialweb-computer-kcp/api/kcpgvr"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/app/compose"
	"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/kcpserver"
)

const (
	AppWorkspaceLabel = "apps.socialweb.computer/application"
	AppWorkspaceType  = "universal"
	PeerReadRole      = "market-network-peers"

	accountWorkspaceType = "account"
	foreignAccountPrefix = "system:kcp:serviceaccount:"
)

var (
	clusterRoleGVR        = kcpgvr.ClusterRoles
	clusterRoleBindingGVR = kcpgvr.ClusterRoleBindings
	serviceAccountGVR     = kcpgvr.ServiceAccounts
)

var reservedAppNames = []string{"global", "relay", "alice", "bob", "dev", compose.ComponentWorkspace}

func ForeignAccount(cluster, namespace, name string) string {
	return foreignAccountPrefix + cluster + ":" + namespace + ":" + name
}

func checkAppName(name string) error {
	if slices.Contains(reservedAppNames, name) || strings.HasSuffix(name, manifests.SharedWorkspaceSuffix) {
		return fmt.Errorf("apps[].name %q names a workspace the market network keeps for itself (%s, *%s)", name,
			strings.Join(reservedAppNames, ", "), manifests.SharedWorkspaceSuffix)
	}
	return nil
}

func (ctl *Controller) appWorkspace(name string) string { return ctl.c.Workspace() + ":" + name }

func (ctl *Controller) applyRBAC(ctx context.Context, r compose.Root, ws string, gvr schema.GroupVersionResource, obj any, kind string) error {
	m, err := compose.ToMap(obj)
	if err != nil {
		return err
	}
	u := kube.Doc(m)
	u["apiVersion"], u["kind"] = gvr.GroupVersion().String(), kind
	meta := compose.DocMeta(u)
	labels := r.Labels(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, ws, gvr, u)
}

func (ctl *Controller) ensureAppWorkspace(ctx context.Context, r compose.Root, name string) error {
	ws := ctl.c.Workspace()
	ri, err := ctl.c.Client(ws, compose.WorkspaceGVR)
	if err != nil {
		return err
	}
	u := ctl.c.Object(r, ws, "tenancy.kcp.io/v1alpha1", "Workspace", name, map[string]string{AppWorkspaceLabel: name})
	if _, err := kube.Get[tenancyv1alpha1.Workspace](ctx, ri, name); apierrors.IsNotFound(err) {
		compose.DocMeta(u)["annotations"] = map[string]string{kcpserver.WorkspaceOwnerAnnotation: kcpserver.ShardAdminOwner}
	} else if err != nil {
		return err
	}
	u["spec"] = map[string]any{"type": map[string]any{"name": AppWorkspaceType, "path": compose.DefaultWorkspace}}
	if err := ctl.c.Apply(ctx, r, ws, compose.WorkspaceGVR, u); err != nil {
		return fmt.Errorf("applying the workspace %s: %w", ctl.appWorkspace(name), err)
	}
	last, err := kube.Await(ctx, ri, name, compose.ApplyTimeout, func(cur *tenancyv1alpha1.Workspace) (bool, error) {
		if cur == nil {
			return false, nil
		}
		if cur.DeletionTimestamp != nil {
			return false, fmt.Errorf("the workspace %s is being deleted; the next reconcile places it again", ctl.appWorkspace(name))
		}
		return cur.Status.Phase == "Ready", nil
	})
	if !errors.Is(err, kube.ErrAwaitTimeout) {
		return err
	}
	phase := ""
	var getErr error
	if last == nil {
		getErr = apierrors.NewNotFound(compose.WorkspaceGVR.GroupResource(), name)
	} else {
		phase = string(last.Status.Phase)
	}
	return fmt.Errorf("the workspace %s is not Ready within %s (phase %q): %v", ctl.appWorkspace(name), compose.ApplyTimeout, phase, getErr)
}

func (ctl *Controller) waitGone(ctx context.Context, ws string, gvr schema.GroupVersionResource, name, what string) error {
	ri, err := ctl.c.Client(ws, gvr)
	if err != nil {
		return err
	}
	_, err = kube.Await(ctx, ri, name, ChildDeleteTimeout, kube.Absent[metav1.PartialObjectMetadata])
	if errors.Is(err, kube.ErrAwaitTimeout) {
		return fmt.Errorf("%s is still there %s after its delete", what, ChildDeleteTimeout)
	}
	return err
}

func (ctl *Controller) deleteAppWorkspace(ctx context.Context, name string) error {
	if name == "" {
		return nil
	}
	ri, err := ctl.c.Client(ctl.c.Workspace(), compose.WorkspaceGVR)
	if err != nil {
		return err
	}
	if err := ri.Delete(ctx, name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
		return err
	}
	return ctl.waitGone(ctx, ctl.c.Workspace(), compose.WorkspaceGVR, name, "the workspace "+ctl.appWorkspace(name))
}

func (ctl *Controller) deleteAccountWorkspaces(ctx context.Context) error {
	ri, err := ctl.c.Client(ctl.c.Workspace(), compose.WorkspaceGVR)
	if err != nil {
		return err
	}
	deleted := map[string]bool{}
	left := 0
	_, err = kube.AwaitList(ctx, ri, metav1.ListOptions{}, ChildDeleteTimeout, func(items []tenancyv1alpha1.Workspace) (bool, error) {
		left = 0
		for _, w := range items {
			if w.Spec.Type == nil || w.Spec.Type.Name != accountWorkspaceType {
				continue
			}
			left++
			if w.DeletionTimestamp == nil && !deleted[w.Name] {
				if err := ri.Delete(ctx, w.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
					return false, err
				}
				deleted[w.Name] = true
			}
		}
		return left == 0, nil
	})
	if errors.Is(err, kube.ErrAwaitTimeout) {
		return fmt.Errorf("%d account workspace(s) are still there %s after their delete", left, ChildDeleteTimeout)
	}
	return err
}

func (ctl *Controller) stopAppComponents(ctx context.Context, name string) error {
	ws := ctl.appWorkspace(name) + ":" + compose.ComponentWorkspace
	types := append(append([]compose.ComponentType{}, compose.ComponentTypes...), compose.PolicyTypes...)
	deadline := time.Now().Add(ChildDeleteTimeout)
	var pending []kube.Resource
	seen := map[string]bool{}
	for _, ct := range types {
		if seen[ct.Kind] {
			continue
		}
		seen[ct.Kind] = true
		nri, err := ctl.c.Client(ws, ct.GVR)
		if err != nil {
			return err
		}
		ri := nri.In("default")
		list, err := kube.List[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{})
		if apierrors.IsNotFound(err) {
			continue
		}
		if err != nil {
			return err
		}
		if len(list) == 0 {
			continue
		}
		if err := deleteListed(ctx, ri, list, nil); err != nil {
			return err
		}
		pending = append(pending, ri)
	}
	left := 0
	for _, ri := range pending {
		remaining := time.Until(deadline)
		if remaining <= 0 {
			list, err := kube.List[metav1.PartialObjectMetadata](ctx, ri, metav1.ListOptions{})
			if err != nil && !apierrors.IsNotFound(err) {
				return err
			}
			if err == nil {
				left += len(list)
			}
			continue
		}
		deleted := map[string]bool{}
		items, err := kube.AwaitList(ctx, ri, metav1.ListOptions{}, remaining, func(items []metav1.PartialObjectMetadata) (bool, error) {
			return len(items) == 0, deleteListed(ctx, ri, items, deleted)
		})
		if errors.Is(err, kube.ErrAwaitTimeout) {
			left += len(items)
			continue
		}
		if err != nil {
			return err
		}
	}
	if left > 0 {
		return fmt.Errorf("%d component(s) of %s still run %s after their delete", left, ws, ChildDeleteTimeout)
	}
	return nil
}

func deleteListed(ctx context.Context, ri kube.Resource, items []metav1.PartialObjectMetadata, deleted map[string]bool) error {
	for _, p := range items {
		if p.DeletionTimestamp != nil || deleted[p.Name] {
			continue
		}
		if err := ri.Delete(ctx, p.Name, metav1.DeleteOptions{}); err != nil && !apierrors.IsNotFound(err) {
			return err
		}
		if deleted != nil {
			deleted[p.Name] = true
		}
	}
	return nil
}

func (ctl *Controller) clusterOf(ctx context.Context, parent, name string) string {
	w, err := compose.Get[tenancyv1alpha1.Workspace](ctx, ctl.c, parent, compose.WorkspaceGVR, name)
	if err != nil {
		return ""
	}
	return w.Spec.Cluster
}

func (ctl *Controller) grantPeers(ctx context.Context, r compose.Root, names []string) error {
	ids := map[string]map[string]string{}
	for _, name := range names {
		ws := ctl.appWorkspace(name)
		own := map[string]string{}
		if id := ctl.clusterOf(ctx, ctl.c.Workspace(), name); id != "" {
			own[ws] = id
		}
		if id := ctl.clusterOf(ctx, ws, compose.ComponentWorkspace); id != "" {
			own[ws+":"+compose.ComponentWorkspace] = id
		}
		if len(own) > 0 {
			ids[name] = own
		}
	}
	key := fmt.Sprint(ids)
	if ctl.peers == key {
		return nil
	}
	role := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{Name: PeerReadRole}, Rules: []rbacv1.PolicyRule{
		{NonResourceURLs: []string{"/"}, Verbs: []string{"access"}},
		{APIGroups: []string{v1alpha1.GroupName}, Resources: []string{"marketidentities"}, Verbs: []string{"get", "list"}}}}
	for name, own := range ids {
		var readers []rbacv1.Subject
		for other, theirs := range ids {
			if other == name {
				continue
			}
			for _, id := range theirs {
				readers = append(readers, rbacv1.Subject{Kind: rbacv1.UserKind, APIGroup: rbacv1.GroupName, Name: ForeignAccount(id, "default", compose.PdsComponent)})
			}
		}
		slices.SortFunc(readers, func(a, b rbacv1.Subject) int { return strings.Compare(a.Name, b.Name) })
		for ws := range own {
			if err := ctl.applyRBAC(ctx, r, ws, clusterRoleGVR, role, "ClusterRole"); err != nil {
				return fmt.Errorf("the peer read role in %s: %w", ws, err)
			}
			b := &rbacv1.ClusterRoleBinding{ObjectMeta: metav1.ObjectMeta{Name: PeerReadRole}, Subjects: readers,
				RoleRef: rbacv1.RoleRef{APIGroup: rbacv1.GroupName, Kind: "ClusterRole", Name: PeerReadRole}}
			if err := ctl.applyRBAC(ctx, r, ws, clusterRoleBindingGVR, b, "ClusterRoleBinding"); err != nil {
				return fmt.Errorf("the peer read binding in %s: %w", ws, err)
			}
		}
	}
	ctl.peers = key
	ctl.log().Info("the pds account of every child application may read the MarketIdentity objects of the others", "clusters", key)
	return nil
}