package kcpapply import ( "cmp" "context" "encoding/json" "errors" "fmt" "io" "log/slog" "maps" "net/http" "net/url" "reflect" "sort" "strings" "sync" "time" tenancyv1alpha1 "github.com/kcp-dev/sdk/apis/tenancy/v1alpha1" authenticationv1 "k8s.io/api/authentication/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/dynamic" "k8s.io/client-go/rest" "k8s.io/client-go/tools/cache" watchtools "k8s.io/client-go/tools/watch" "github.com/johnandersen777/socialweb-computer-kcp/api/kcpgvr" "github.com/johnandersen777/socialweb-computer-kcp/internal/controllers/runtime/manifests" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt" ) const ( FieldManager = "socialweb-computer" DefaultWaitTimeout = time.Minute DefaultPoll = time.Second FirstPoll = 50 * time.Millisecond defaultNamespace = "default" schemaKind = "APIResourceSchema" ) var ( ErrSchemaChanged = errors.New("kcpapply: an APIResourceSchema is immutable and its live spec differs; publish a new revision name") ErrKubectlManaged = errors.New("kcpapply: the object was applied by kubectl and is never adopted; apply into a fresh instance") ErrWaitTimedOut = errors.New("kcpapply: a wait did not hold before its deadline") ErrWaitFailed = errors.New("kcpapply: a waited-on object reports a failure, so the wait cannot hold") ) type Action string const ( Created Action = "created" Configured Action = "configured" Unchanged Action = "unchanged" ) type Options struct { DryRun bool WaitTimeout time.Duration Poll time.Duration FieldManager string Base string Labels map[string]string Log *slog.Logger } func (a *Applier) fieldManager() string { if a.opts.FieldManager != "" { return a.opts.FieldManager } return FieldManager } type FieldChange struct { Field string `json:"field"` From []string `json:"from"` To []string `json:"to"` } type Change struct { Kind string `json:"kind"` Namespace string `json:"namespace,omitempty"` Name string `json:"name"` Action Action `json:"action"` Managers []FieldChange `json:"managers,omitempty"` } type WaitReport struct { When string `json:"when"` Kind string `json:"kind"` Name string `json:"name,omitempty"` Workspace string `json:"workspace"` For string `json:"for"` Met bool `json:"met"` Captured string `json:"captured,omitempty"` Seconds float64 `json:"seconds"` } type PhaseReport struct { Set string `json:"set"` Phase string `json:"phase"` Workspace string `json:"workspace"` Changes []Change `json:"changes"` Waits []WaitReport `json:"waits,omitempty"` } type Report struct { DryRun bool `json:"dryRun"` Phases []PhaseReport `json:"phases"` } func (r Report) Changed() int { n := 0 for _, p := range r.Phases { for _, c := range p.Changes { if c.Action != Unchanged { n++ } } } return n } type Applier struct { base *rest.Config opts Options ownerMu sync.Mutex owner string } type workspaceOwner struct { Username string `json:"username"` Groups []string `json:"groups,omitempty"` } func New(cfg *rest.Config, o Options) (*Applier, error) { if cfg == nil { return nil, errors.New("kcpapply: no rest.Config") } u, err := url.Parse(cfg.Host) if err != nil || u.Host == "" { return nil, fmt.Errorf("kcpapply: the rest.Config host %q is no URL", cfg.Host) } if i := strings.Index(u.Path, "/clusters/"); i >= 0 { u.Path = u.Path[:i] } u.Path = strings.TrimSuffix(u.Path, "/") base := rest.CopyConfig(cfg) base.Host = u.String() if base.QPS == 0 { base.QPS, base.Burst = 50, 100 } o.WaitTimeout = opt.Positive(o.WaitTimeout, DefaultWaitTimeout) o.Poll = opt.Positive(o.Poll, DefaultPoll) if o.Log == nil { o.Log = slog.New(slog.DiscardHandler) } return &Applier{base: base, opts: o}, nil } func (a *Applier) ForWorkspace(path string) *rest.Config { c := rest.CopyConfig(a.base) c.Host = a.base.Host + "/clusters/" + path return c } func (a *Applier) Apply(ctx context.Context, set manifests.Set, values any) (Report, error) { set.Base = cmp.Or(set.Base, a.opts.Base) rep := Report{DryRun: a.opts.DryRun} if set.BaseCluster == "" && set.Scope != manifests.ScopeKcp && set.Base != "" && set.Base != "root" { id, err := a.ClusterOf(ctx, set.Base) if err != nil { return rep, fmt.Errorf("kcpapply: set %s: the logical cluster of %s: %w", set.Name, set.Base, err) } set.BaseCluster = id } for _, p := range set.Phases { pr, err := a.applyPhase(ctx, set, p, values) rep.Phases = append(rep.Phases, pr) if err != nil { return rep, fmt.Errorf("kcpapply: set %s phase %s: %w", set.Name, p.Name, err) } } return rep, nil } func (a *Applier) applyPhase(ctx context.Context, set manifests.Set, p manifests.Phase, values any) (PhaseReport, error) { pr := PhaseReport{Set: set.Name, Phase: p.Name, Changes: []Change{}} ws, err := set.Workspace(p, nil, values) if err != nil { return pr, err } pr.Workspace = ws for _, w := range p.Before { wr, err := a.wait(ctx, set, p, w, values, nil, "before") pr.Waits = append(pr.Waits, wr) if err != nil { return pr, err } } r, err := set.RenderPhase(p, values) if err != nil { return pr, err } if len(r.Objects) > 0 { changes, err := a.ApplyObjects(ctx, ws, r.Objects) pr.Changes = append(pr.Changes, changes...) if err != nil { return pr, err } a.opts.Log.Info("phase applied", "set", set.Name, "phase", p.Name, "workspace", ws, "objects", len(changes)) } for _, w := range p.After { wr, err := a.wait(ctx, set, p, w, values, r.Objects, "after") pr.Waits = append(pr.Waits, wr) if err != nil { return pr, err } } return pr, nil } func (a *Applier) ClusterOf(ctx context.Context, path string) (string, error) { i := strings.LastIndex(path, ":") if i < 0 { return path, nil } dyn, err := dynamic.NewForConfig(a.ForWorkspace(path[:i])) if err != nil { return "", err } u, err := dyn.Resource(kcpgvr.Workspaces).Get(ctx, path[i+1:], metav1.GetOptions{}) if err != nil { return "", err } id, _, _ := unstructured.NestedString(u.Object, "spec", "cluster") if id == "" { return "", fmt.Errorf("the workspace %s names no logical cluster yet", path) } return id, nil } func (a *Applier) ApplyObjects(ctx context.Context, workspace string, objs []manifests.Object) ([]Change, error) { cfg := a.ForWorkspace(workspace) served, err := a.servedFor(ctx, cfg, workspace, objs) if err != nil { return nil, err } dyn, err := dynamic.NewForConfig(cfg) if err != nil { return nil, err } var out []Change for _, o := range objs { u := o.Unstructured gvk := u.GroupVersionKind() m, ok := served[gvk] if !ok { return out, fmt.Errorf("%s: %s %s is not served in %s", o.Path, gvk, u.GetName(), workspace) } var ri dynamic.ResourceInterface = dyn.Resource(m.resource) ns := "" if m.namespaced { ns = u.GetNamespace() ns = cmp.Or(ns, defaultNamespace) ri = dyn.Resource(m.resource).Namespace(ns) } c, err := a.applyOne(ctx, ri, u) c.Namespace = ns out = append(out, c) if err != nil { return out, fmt.Errorf("%s: %s %s in %s: %w", o.Path, gvk.Kind, u.GetName(), workspace, err) } } return out, nil } type servedResource struct { resource schema.GroupVersionResource namespaced bool } func (a *Applier) servedFor(ctx context.Context, cfg *rest.Config, workspace string, objs []manifests.Object) (map[schema.GroupVersionKind]servedResource, error) { deadline := time.Now().Add(a.opts.WaitTimeout) for attempt := 1; ; attempt++ { served, err := discover(ctx, cfg, objs) if err != nil { return nil, fmt.Errorf("discovery for %s: %w", workspace, err) } var missing *manifests.Object for i := range objs { if _, ok := served[objs[i].Unstructured.GroupVersionKind()]; !ok { missing = &objs[i] break } } if missing == nil { return served, nil } if !time.Now().Before(deadline) { return nil, fmt.Errorf("%s: %s %s is not served in %s after %s", missing.Path, missing.Unstructured.GroupVersionKind(), missing.Unstructured.GetName(), workspace, a.opts.WaitTimeout) } a.opts.Log.Info("waiting for every object to be served", "workspace", workspace, "attempt", attempt, "path", missing.Path, "kind", missing.Unstructured.GroupVersionKind().String(), "deadline", a.opts.WaitTimeout.String()) select { case <-ctx.Done(): return nil, ctx.Err() case <-time.After(min(FirstPoll< 0 { u = u.DeepCopy() labels := u.GetLabels() if labels == nil { labels = map[string]string{} } maps.Copy(labels, a.opts.Labels) u.SetLabels(labels) } if u.GroupVersionKind().Group == tenancyv1alpha1.SchemeGroupVersion.Group && u.GetKind() == "Workspace" { stamped, err := a.stampOwner(ctx, u, live) if err != nil { return c, err } u = stamped } body, err := json.Marshal(u.Object) if err != nil { return c, err } force := true got, err := ri.Patch(ctx, u.GetName(), types.ApplyPatchType, body, metav1.PatchOptions{FieldManager: a.fieldManager(), Force: &force, DryRun: a.dryRun()}) if err != nil { return c, err } c.Managers = managerChanges(live, got) switch { case live == nil: c.Action = Created case len(c.Managers) == 0 && sameContent(live, got): c.Action = Unchanged default: c.Action = Configured } return c, nil } func (a *Applier) stampOwner(ctx context.Context, u, live *unstructured.Unstructured) (*unstructured.Unstructured, error) { if _, set := u.GetAnnotations()[tenancyv1alpha1.ExperimentalWorkspaceOwnerAnnotationKey]; set { return u, nil } var owner string if live != nil { var held bool if owner, held = live.GetAnnotations()[tenancyv1alpha1.ExperimentalWorkspaceOwnerAnnotationKey]; !held { return u, nil } } else { var err error if owner, err = a.workspaceOwner(ctx); err != nil { return nil, err } } out := u.DeepCopy() ann := out.GetAnnotations() if ann == nil { ann = map[string]string{} } ann[tenancyv1alpha1.ExperimentalWorkspaceOwnerAnnotationKey] = owner out.SetAnnotations(ann) return out, nil } func (a *Applier) workspaceOwner(ctx context.Context) (string, error) { a.ownerMu.Lock() defer a.ownerMu.Unlock() if a.owner == "" { dyn, err := dynamic.NewForConfig(a.ForWorkspace("root")) if err != nil { return "", err } review := &unstructured.Unstructured{Object: map[string]any{"apiVersion": "authentication.k8s.io/v1", "kind": "SelfSubjectReview"}} got, err := dyn.Resource(kcpgvr.SelfSubjectReviews).Create(ctx, review, metav1.CreateOptions{}) if err != nil { return "", fmt.Errorf("reading the applying identity for the workspace owner: %w", err) } var r authenticationv1.SelfSubjectReview if err := runtime.DefaultUnstructuredConverter.FromUnstructured(got.Object, &r); err != nil { return "", err } b, err := json.Marshal(workspaceOwner{Username: r.Status.UserInfo.Username, Groups: r.Status.UserInfo.Groups}) if err != nil { return "", err } a.owner = string(b) } return a.owner, nil } func (a *Applier) applySchema(ctx context.Context, ri dynamic.ResourceInterface, u, live *unstructured.Unstructured, c Change) (Change, error) { if live == nil { got, err := ri.Create(ctx, u, metav1.CreateOptions{FieldManager: a.fieldManager(), DryRun: a.dryRun()}) switch { case err == nil: c.Action, c.Managers = Created, managerChanges(nil, got) return c, nil case apierrors.IsAlreadyExists(err): live, err = ri.Get(ctx, u.GetName(), metav1.GetOptions{}) if err != nil { return c, err } default: return c, err } } want, _, _ := unstructured.NestedMap(u.Object, "spec") have, _, _ := unstructured.NestedMap(live.Object, "spec") if at, same := derivative(want, have, "spec"); !same { return c, fmt.Errorf("%w: %s differs at %s", ErrSchemaChanged, u.GetName(), at) } c.Action = Unchanged return c, nil } func sameContent(a, b *unstructured.Unstructured) bool { strip := func(u *unstructured.Unstructured) map[string]any { c := u.DeepCopy() unstructured.RemoveNestedField(c.Object, "metadata", "resourceVersion") unstructured.RemoveNestedField(c.Object, "metadata", "managedFields") unstructured.RemoveNestedField(c.Object, "metadata", "generation") return c.Object } _, same := derivative(strip(a), strip(b), "") _, back := derivative(strip(b), strip(a), "") return same && back } func derivative(want, have any, at string) (string, bool) { switch w := want.(type) { case map[string]any: h, ok := have.(map[string]any) if !ok { return at, false } keys := make([]string, 0, len(w)) for k := range w { keys = append(keys, k) } sort.Strings(keys) for _, k := range keys { if p, same := derivative(w[k], h[k], at+"."+k); !same { return p, false } } return at, true case []any: h, ok := have.([]any) if !ok || len(h) != len(w) { return at, false } for i := range w { if p, same := derivative(w[i], h[i], fmt.Sprintf("%s[%d]", at, i)); !same { return p, false } } return at, true } if wn, ok := number(want); ok { hn, ok := number(have) return at, ok && wn == hn } return at, reflect.DeepEqual(want, have) } func number(v any) (float64, bool) { switch n := v.(type) { case int64: return float64(n), true case int: return float64(n), true case float64: return n, true case json.Number: f, err := n.Float64() return f, err == nil } return 0, false } func refuseKubectl(u *unstructured.Unstructured) error { for _, m := range u.GetManagedFields() { if strings.HasPrefix(m.Manager, "kubectl") && !annotationsOnly(m) { return fmt.Errorf("%w: %s %s is managed by %s", ErrKubectlManaged, u.GetKind(), u.GetName(), m.Manager) } } return nil } func annotationsOnly(m metav1.ManagedFieldsEntry) bool { if m.FieldsV1 == nil || len(m.FieldsV1.Raw) == 0 { return false } var fields map[string]any if err := json.Unmarshal(m.FieldsV1.Raw, &fields); err != nil { return false } allowed := map[string]bool{"f:metadata": true, "f:apiVersion": true, "f:kind": true} for k := range fields { if !allowed[k] { return false } } meta, ok := fields["f:metadata"].(map[string]any) if !ok { return false } for k := range meta { if k != "f:annotations" && k != "f:labels" { return false } } return true } func owners(u *unstructured.Unstructured) map[string][]string { out := map[string][]string{} if u == nil { return out } for _, m := range u.GetManagedFields() { if m.FieldsV1 == nil { continue } var tree map[string]any if json.Unmarshal(m.FieldsV1.GetRawBytes(), &tree) != nil { continue } who := fmt.Sprintf("%s (%s)", m.Manager, m.Operation) walkFields(tree, "", func(path string) { out[path] = append(out[path], who) }) } for k := range out { sort.Strings(out[k]) } return out } func walkFields(tree map[string]any, at string, leaf func(string)) { for k, v := range tree { if k == "." { continue } path := k if at != "" { path = at + "." + k } child, _ := v.(map[string]any) if len(child) == 0 || (len(child) == 1 && child["."] != nil) { leaf(path) continue } walkFields(child, path, leaf) } } func managerChanges(before, after *unstructured.Unstructured) []FieldChange { b, a := owners(before), owners(after) paths := map[string]bool{} for p := range b { paths[p] = true } for p := range a { paths[p] = true } var out []FieldChange for p := range paths { if !reflect.DeepEqual(b[p], a[p]) { out = append(out, FieldChange{Field: p, From: b[p], To: a[p]}) } } sort.Slice(out, func(i, j int) bool { return out[i].Field < out[j].Field }) return out } func (a *Applier) wait(ctx context.Context, set manifests.Set, p manifests.Phase, w manifests.Wait, values any, applied []manifests.Object, when string) (WaitReport, error) { wr := WaitReport{When: when, Kind: w.Kind, Name: w.Name, For: w.For} ws, err := set.Workspace(p, &w, values) if err != nil { return wr, err } wr.Workspace = ws kind := manifests.WaitKinds[w.Kind] dyn, err := dynamic.NewForConfig(a.ForWorkspace(ws)) if err != nil { return wr, err } names := []string{w.Name} if w.Name == "" && w.Kind != "APIBinding" { names = nil for _, o := range applied { if o.Unstructured.GetKind() == w.Kind { names = append(names, o.Unstructured.GetName()) } } if len(names) == 0 { return wr, fmt.Errorf("a %s %s wait names no object and the phase applied none", w.Kind, w.For) } } start := time.Now() var met bool var captured string var last error if a.opts.DryRun { met, captured, last = a.check(ctx, dyn, kind.Resource, kind.Namespaced, w, names) } else { met, captured, last = a.watchFor(ctx, dyn, kind.Resource, kind.Namespaced, w, names) } wr.Seconds = time.Since(start).Seconds() if errors.Is(last, ErrWaitFailed) && !w.BestEffort { return wr, fmt.Errorf("%s %v in %s for %s: %w", w.Kind, names, ws, w.For, last) } if met { wr.Met = true if w.Capture != nil { if err := manifests.SetCaptured(values, *w.Capture, captured); err != nil { return wr, err } wr.Captured = captured } return wr, nil } if err := ctx.Err(); err != nil { return wr, err } if a.opts.DryRun || w.BestEffort { a.opts.Log.Warn("wait not met; continuing", "kind", w.Kind, "name", w.Name, "workspace", ws, "for", w.For, "dryRun", a.opts.DryRun, "err", last) return wr, nil } if w.Kind == "Workspace" { if why := a.workspaceStall(ctx, ws, names); why != "" { last = fmt.Errorf("%v; %s", last, why) } } return wr, fmt.Errorf("%w: %s %v in %s for %s after %s (last error: %v)", ErrWaitTimedOut, w.Kind, names, ws, w.For, a.opts.WaitTimeout, last) } func (a *Applier) workspaceStall(ctx context.Context, ws string, names []string) string { var out []string for _, name := range names { dyn, err := dynamic.NewForConfig(a.ForWorkspace(ws)) if err != nil { continue } u, err := dyn.Resource(kcpgvr.Workspaces).Get(ctx, name, metav1.GetOptions{}) if err != nil { continue } phase, _, _ := unstructured.NestedString(u.Object, "status", "phase") initializers, _, _ := unstructured.NestedStringSlice(u.Object, "status", "initializers") why := fmt.Sprintf("Workspace %s phase %s initializers %v", name, phase, initializers) if child, err := dynamic.NewForConfig(a.ForWorkspace(ws + ":" + name)); err == nil { if lc, err := child.Resource(kcpgvr.LogicalClusters).Get(ctx, "cluster", metav1.GetOptions{}); err == nil { conditions, _, _ := unstructured.NestedSlice(lc.Object, "status", "conditions") for _, raw := range conditions { c, _ := raw.(map[string]any) if c["status"] == "False" { why += fmt.Sprintf("; %v=False %v: %v", c["type"], c["reason"], c["message"]) } } } } out = append(out, why) } return strings.Join(out, "; ") } func (a *Applier) watchFor(ctx context.Context, dyn dynamic.Interface, gvr schema.GroupVersionResource, namespaced bool, w manifests.Wait, names []string) (bool, string, error) { var ri dynamic.ResourceInterface = dyn.Resource(gvr) if namespaced { ri = dyn.Resource(gvr).Namespace(defaultNamespace) } wctx, cancel := context.WithTimeout(ctx, a.opts.WaitTimeout) defer cancel() lw := &cache.ListWatch{ ListWithContextFunc: func(ctx context.Context, o metav1.ListOptions) (runtime.Object, error) { return ri.List(ctx, o) }, WatchFuncWithContext: func(ctx context.Context, o metav1.ListOptions) (watch.Interface, error) { return ri.Watch(ctx, o) }, } seen := map[string]*unstructured.Unstructured{} var met bool var captured string var last error = fmt.Errorf("no %s observed yet", w.Kind) eval := func() (bool, error) { met, captured, last = a.evaluate(w, names, seen) if errors.Is(last, ErrWaitFailed) && !w.BestEffort { return false, last } return met, nil } _, err := watchtools.UntilWithSync(wctx, lw, &unstructured.Unstructured{}, func(store cache.Store) (bool, error) { for _, item := range store.List() { if u, ok := item.(*unstructured.Unstructured); ok { seen[u.GetName()] = u } } return eval() }, func(ev watch.Event) (bool, error) { u, ok := ev.Object.(*unstructured.Unstructured) if !ok { return false, nil } switch ev.Type { case watch.Deleted: delete(seen, u.GetName()) case watch.Added, watch.Modified: seen[u.GetName()] = u default: return false, nil } return eval() }) if met { return true, captured, nil } if errors.Is(last, ErrWaitFailed) { return false, "", last } if err != nil && !wait.Interrupted(err) && !errors.Is(err, watchtools.ErrWatchClosed) { last = err } return false, "", last } func (a *Applier) evaluate(w manifests.Wait, names []string, seen map[string]*unstructured.Unstructured) (bool, string, error) { if w.Kind == "APIBinding" { if len(seen) == 0 { return false, "", errors.New("no APIBinding yet") } for name, u := range seen { if met, _ := w.Met(u); !met { return false, "", fmt.Errorf("APIBinding %s not Ready", name) } } return true, "", nil } captured := "" for _, n := range names { u, ok := seen[n] if !ok { return false, "", fmt.Errorf("%s %s not found", w.Kind, n) } met, value := w.Met(u) if !met { if why := failedObject(u); why != "" { return false, "", fmt.Errorf("%w: %s %s %s", ErrWaitFailed, w.Kind, n, why) } return false, "", fmt.Errorf("%s %s not %s", w.Kind, n, w.For) } captured = value } return true, captured, nil } func (a *Applier) check(ctx context.Context, dyn dynamic.Interface, gvr schema.GroupVersionResource, namespaced bool, w manifests.Wait, names []string) (bool, string, error) { var ri dynamic.ResourceInterface = dyn.Resource(gvr) if namespaced { ri = dyn.Resource(gvr).Namespace(defaultNamespace) } if w.Kind == "APIBinding" { list, err := ri.List(ctx, metav1.ListOptions{}) if err != nil { return false, "", err } if len(list.Items) == 0 { return false, "", errors.New("no APIBinding yet") } for i := range list.Items { if met, _ := w.Met(&list.Items[i]); !met { return false, "", fmt.Errorf("APIBinding %s not Ready", list.Items[i].GetName()) } } return true, "", nil } captured := "" for _, n := range names { u, err := ri.Get(ctx, n, metav1.GetOptions{}) if err != nil { return false, "", err } met, value := w.Met(u) if !met { if why := failedObject(u); why != "" { return false, "", fmt.Errorf("%w: %s %s %s", ErrWaitFailed, w.Kind, n, why) } return false, "", fmt.Errorf("%s %s not %s", w.Kind, n, w.For) } captured = value } return true, captured, nil } func failedObject(u *unstructured.Unstructured) string { phase, _, _ := unstructured.NestedString(u.Object, "status", "phase") msg, _, _ := unstructured.NestedString(u.Object, "status", "message") observed, found, _ := unstructured.NestedInt64(u.Object, "status", "observedGeneration") current := !found || observed == u.GetGeneration() if phase == "Failed" && current { return "is Failed: " + msg } return "" }