diff --git a/ROADMAP.md b/ROADMAP.md index 4267d5d..6cdc476 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -320,17 +320,22 @@ these are the corners where drift would start if it starts anywhere. declaration). Think "typed status for a controller," not "Terraform outputs." -- [ ] **K7 — Expose the scheduling DAG / plan-preview.** Today only a *rooted* - composition graph is exposed (`GetChildrenGraph` → UI DagView: a resource's - `owns`/`ref` edges outward). The **operation-scheduling DAG** the dispatcher - builds from `$ref` edges — the actual apply *order* and *why* — is internal - and never surfaced. And `DryRunApply` previews one resource's diff, not a - whole submission's ordering. Add a batch **plan-preview**: given a set of - manifests (or a repo dir), return the dependency DAG + the topological - apply order (+ per-resource dry-run diff), so an operator can see "what - will happen, in what order, and what waits on what" before applying — - including for a GitOps sync. Pairs with I1 (plan-on-PR: the same preview, - rendered on a PR). +- [~] **K7 — Expose the scheduling DAG / plan-preview.** New **`openctl plan + -f `** (`internal/plan` + `internal/cli/plan.go`): loads a set + of manifests (`.cue`/`.yaml`/`.yml`, dirs walked recursively), derives the + cross-resource `$ref` dependency graph (via the same `refs.Collect` + primitive the dispatcher uses, keyed on the full apiVersion|kind|name + triple), topologically sorts it (Kahn) into **waves** — resources in a + wave have no inter-dependency and can apply concurrently — prints each + resource's direct deps (`← Kind/Name`), flags `$ref` targets **not** in the + set (must already exist), and errors on a dependency cycle (naming the + stuck resources). Offline: no controller connection. Validated by planning + the shipped `examples/homelab/` set (11 resources → 2 waves). Unit-tested + (chain/diamond/external/cycle/duplicate/full-triple-key + the batch + loader + output). **Remaining:** per-resource dry-run **diff** (`--diff`) + is server-only — it needs the controller's `DryRunApply` (drift vs stored + last-applied + live provider Get) — and a batch RPC to expose the DAG to + the UI. Pairs with I1 (plan-on-PR: render this preview on a PR). The UI/controller feature build-out is essentially complete (all UI phases + arch Phase 8 shipped). The next round is driven by the diff --git a/internal/cli/plan.go b/internal/cli/plan.go new file mode 100644 index 0000000..ff108dc --- /dev/null +++ b/internal/cli/plan.go @@ -0,0 +1,146 @@ +package cli + +import ( + "fmt" + "io" + "io/fs" + "os" + "path/filepath" + "sort" + "strings" + + "github.com/spf13/cobra" + + "github.com/openctl/openctl/internal/manifest" + "github.com/openctl/openctl/internal/plan" + "github.com/openctl/openctl/pkg/protocol" +) + +func newPlanCommand() *cobra.Command { + var files []string + + cmd := &cobra.Command{ + Use: "plan -f [-f ...]", + Short: "Preview the apply order and $ref dependency graph for a set of manifests", + Long: "Load a set of manifests (files or directories, .cue/.yaml/.yml) and print the\n" + + "cross-resource dependency graph derived from their $ref markers: the\n" + + "topological apply order as waves (resources in the same wave have no\n" + + "inter-dependency and can apply concurrently), what each resource waits on, and\n" + + "any $ref targets that are NOT in the set (they must already exist).\n\n" + + "This is offline — it needs no controller connection. (Per-resource dry-run\n" + + "diffs need a running controller and are not included here.)", + RunE: func(cmd *cobra.Command, args []string) error { + paths := append([]string{}, files...) + paths = append(paths, args...) // also accept bare path args + if len(paths) == 0 { + return fmt.Errorf("no manifests: pass -f (or bare path args)") + } + resources, err := loadManifestBatch(paths) + if err != nil { + return err + } + if len(resources) == 0 { + return fmt.Errorf("no manifests found in %s", strings.Join(paths, ", ")) + } + p, err := plan.Build(resources) + if err != nil { + return fmt.Errorf("plan: %w", err) + } + printPlan(cmd.OutOrStdout(), p) + return nil + }, + } + cmd.Flags().StringArrayVarP(&files, "file", "f", nil, "Manifest file or directory (repeatable)") + return cmd +} + +// loadManifestBatch loads every manifest under the given paths (files or +// directories), dispatching by extension. Directories are walked recursively +// for .cue/.yaml/.yml files, in sorted order for deterministic output. +func loadManifestBatch(paths []string) ([]*protocol.Resource, error) { + var files []string + for _, p := range paths { + info, err := os.Stat(p) + if err != nil { + return nil, fmt.Errorf("read %s: %w", p, err) + } + if !info.IsDir() { + files = append(files, p) + continue + } + err = filepath.WalkDir(p, func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if !d.IsDir() && isManifestFile(path) { + files = append(files, path) + } + return nil + }) + if err != nil { + return nil, fmt.Errorf("walk %s: %w", p, err) + } + } + sort.Strings(files) + + var out []*protocol.Resource + for _, f := range files { + rs, err := loadManifestFile(f) + if err != nil { + return nil, fmt.Errorf("%s: %w", f, err) + } + out = append(out, rs...) + } + return out, nil +} + +func isManifestFile(path string) bool { + switch filepath.Ext(path) { + case ".cue", ".yaml", ".yml": + return true + } + return false +} + +func loadManifestFile(path string) ([]*protocol.Resource, error) { + switch filepath.Ext(path) { + case ".cue": + return manifest.LoadCUE(path) + case ".yaml", ".yml": + return manifest.LoadMultiple(path) + default: + return nil, fmt.Errorf("unsupported manifest extension (want .cue/.yaml/.yml)") + } +} + +func printPlan(w io.Writer, p *plan.Plan) { + fmt.Fprintf(w, "Plan: %d resource(s), %d wave(s)\n\n", p.Count(), len(p.Waves)) + fmt.Fprintln(w, "Apply order:") + + external := map[string][]string{} // node display -> external refs + for i, wave := range p.Waves { + fmt.Fprintf(w, " wave %d:\n", i+1) + for _, n := range wave { + line := " " + n.Display() + if len(n.Deps) > 0 { + line += " ← " + strings.Join(n.Deps, ", ") + } + fmt.Fprintln(w, line) + if len(n.External) > 0 { + external[n.Display()] = n.External + } + } + } + + if len(external) > 0 { + fmt.Fprintln(w, "\nExternal references (must already exist — not in this set):") + names := make([]string, 0, len(external)) + for k := range external { + names = append(names, k) + } + sort.Strings(names) + for _, k := range names { + fmt.Fprintf(w, " %s → %s\n", k, strings.Join(external[k], ", ")) + } + } +} diff --git a/internal/cli/plan_test.go b/internal/cli/plan_test.go new file mode 100644 index 0000000..93a3e10 --- /dev/null +++ b/internal/cli/plan_test.go @@ -0,0 +1,88 @@ +package cli + +import ( + "bytes" + "os" + "path/filepath" + "testing" + + "github.com/openctl/openctl/internal/plan" +) + +func writeFile(t *testing.T, path, content string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatal(err) + } +} + +// TestLoadManifestBatch_DirAndMultiDoc loads a directory recursively, splitting +// multi-doc YAML and dispatching by extension. +func TestLoadManifestBatch_DirAndMultiDoc(t *testing.T) { + dir := t.TempDir() + writeFile(t, filepath.Join(dir, "a.yaml"), + "apiVersion: v1\nkind: A\nmetadata:\n name: a\n---\napiVersion: v1\nkind: B\nmetadata:\n name: b\n") + writeFile(t, filepath.Join(dir, "sub", "c.yml"), + "apiVersion: v1\nkind: C\nmetadata:\n name: c\n") + // A non-manifest file must be ignored by the dir walk. + writeFile(t, filepath.Join(dir, "notes.txt"), "ignore me") + + rs, err := loadManifestBatch([]string{dir}) + if err != nil { + t.Fatalf("loadManifestBatch: %v", err) + } + if len(rs) != 3 { + t.Fatalf("loaded %d resources, want 3 (A,B,C)", len(rs)) + } + // The loaded set feeds Build without error. + if _, err := plan.Build(rs); err != nil { + t.Fatalf("Build: %v", err) + } +} + +func TestLoadManifestBatch_MissingPath(t *testing.T) { + if _, err := loadManifestBatch([]string{filepath.Join(t.TempDir(), "nope.yaml")}); err == nil { + t.Fatal("want an error for a missing path") + } +} + +func TestLoadManifestFile_UnsupportedExt(t *testing.T) { + f := filepath.Join(t.TempDir(), "x.json") + writeFile(t, f, "{}") + if _, err := loadManifestFile(f); err == nil { + t.Fatal("want an error for an unsupported extension") + } +} + +// TestPrintPlan_Output checks the rendered plan lists waves, dependencies, and +// external references. +func TestPrintPlan_Output(t *testing.T) { + writeDir := t.TempDir() + // B $refs A (in set) and an external Cluster/home (not in set). + writeFile(t, filepath.Join(writeDir, "m.yaml"), + "apiVersion: v1\nkind: A\nmetadata:\n name: a\n"+ + "---\n"+ + "apiVersion: v1\nkind: B\nmetadata:\n name: b\nspec:\n"+ + " dep:\n $ref:\n apiVersion: v1\n kind: A\n name: a\n field: status.x\n"+ + " ext:\n $ref:\n apiVersion: k3s.openctl.io/v1\n kind: Cluster\n name: home\n field: status.x\n") + + rs, err := loadManifestBatch([]string{writeDir}) + if err != nil { + t.Fatal(err) + } + p, err := plan.Build(rs) + if err != nil { + t.Fatal(err) + } + var buf bytes.Buffer + printPlan(&buf, p) + out := buf.String() + for _, want := range []string{"wave 1:", "A/a", "wave 2:", "B/b ← A/a", "External references", "Cluster/home"} { + if !bytes.Contains(buf.Bytes(), []byte(want)) { + t.Errorf("plan output missing %q:\n%s", want, out) + } + } +} diff --git a/internal/cli/root.go b/internal/cli/root.go index b0a29ec..8181e57 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -104,6 +104,7 @@ func init() { rootCmd.AddCommand(newConfigCommand()) rootCmd.AddCommand(newApplyCommand()) rootCmd.AddCommand(newValidateCommand()) + rootCmd.AddCommand(newPlanCommand()) rootCmd.AddCommand(newExplainCommand()) rootCmd.AddCommand(newVersionCommand()) rootCmd.AddCommand(newStateCommand()) diff --git a/internal/plan/plan.go b/internal/plan/plan.go new file mode 100644 index 0000000..49347c4 --- /dev/null +++ b/internal/plan/plan.go @@ -0,0 +1,168 @@ +// Package plan computes a batch apply plan over a set of manifests: the +// cross-resource $ref dependency graph, the topological apply order (as waves +// of independently-appliable resources), and references to resources outside +// the set (which must already exist). It is pure and offline — no controller +// connection, no provider calls — so `openctl plan` can preview ordering and +// "what waits on what" before any apply. +package plan + +import ( + "fmt" + "sort" + "strings" + + "github.com/openctl/openctl/internal/controller/refs" + "github.com/openctl/openctl/pkg/protocol" +) + +// key is the stable identity of a resource in a plan: the full +// (apiVersion, kind, name) triple, so two providers sharing a Kind+Name don't +// collide. +func key(apiVersion, kind, name string) string { + return apiVersion + "|" + kind + "|" + name +} + +// Node is one resource in the plan plus its resolved dependencies. +type Node struct { + Resource *protocol.Resource + // Deps are the display labels ("Kind/Name") of in-set resources this one + // $refs — they must apply first. Sorted, de-duplicated. + Deps []string + // External are display labels for $ref targets NOT in the set — they must + // already exist when this resource applies. Sorted, de-duplicated. + External []string +} + +// Display is the human label for a node ("Kind/Name"). +func (n *Node) Display() string { + return n.Resource.Kind + "/" + n.Resource.Metadata.Name +} + +// Plan is the computed batch plan. +type Plan struct { + // Waves are the topological levels: every resource in wave i depends only + // on resources in earlier waves, so a wave's resources can apply + // concurrently. Within a wave, nodes are sorted by display label. + Waves [][]*Node +} + +// Count returns the total number of planned resources. +func (p *Plan) Count() int { + n := 0 + for _, w := range p.Waves { + n += len(w) + } + return n +} + +// Build computes the plan from a set of manifests. Returns an error on a +// duplicate resource identity or a dependency cycle (naming the resources +// stuck in the cycle). +func Build(resources []*protocol.Resource) (*Plan, error) { + nodes := make(map[string]*Node, len(resources)) + deps := make(map[string][]string, len(resources)) // key -> dependency keys (in-set) + order := make([]string, 0, len(resources)) // input order for stable iteration + for _, r := range resources { + if r == nil { + continue + } + k := key(r.APIVersion, r.Kind, r.Metadata.Name) + if _, dup := nodes[k]; dup { + return nil, fmt.Errorf("duplicate resource %s/%s (%s)", r.Kind, r.Metadata.Name, r.APIVersion) + } + nodes[k] = &Node{Resource: r} + order = append(order, k) + } + + // Derive edges from the $ref markers in each spec. + for _, k := range order { + n := nodes[k] + seenDep := map[string]bool{} + seenExt := map[string]bool{} + for _, ref := range refs.Collect(n.Resource.Spec) { + depKey := key(ref.APIVersion, ref.Kind, ref.Name) + if depKey == k { + continue // a resource referencing itself is not a dependency + } + if dep, ok := nodes[depKey]; ok { + if !seenDep[depKey] { + seenDep[depKey] = true + deps[k] = append(deps[k], depKey) + n.Deps = append(n.Deps, dep.Display()) + } + } else { + disp := ref.Kind + "/" + ref.Name + if !seenExt[disp] { + seenExt[disp] = true + n.External = append(n.External, disp) + } + } + } + sort.Strings(n.Deps) + sort.Strings(n.External) + } + + waves, err := topoWaves(nodes, deps) + if err != nil { + return nil, err + } + return &Plan{Waves: waves}, nil +} + +// topoWaves runs Kahn's algorithm level-by-level: each wave is the set of nodes +// whose dependencies are all satisfied by earlier waves. Within a wave, nodes +// are sorted by display label for deterministic output. A remaining node with +// nonzero in-degree means a cycle. +func topoWaves(nodes map[string]*Node, deps map[string][]string) ([][]*Node, error) { + indegree := make(map[string]int, len(nodes)) + dependents := make(map[string][]string) + for k := range nodes { + indegree[k] = len(deps[k]) + for _, dep := range deps[k] { + dependents[dep] = append(dependents[dep], k) + } + } + + var frontier []string + for k := range nodes { + if indegree[k] == 0 { + frontier = append(frontier, k) + } + } + + var waves [][]*Node + remaining := len(nodes) + for len(frontier) > 0 { + wave := make([]*Node, 0, len(frontier)) + for _, k := range frontier { + wave = append(wave, nodes[k]) + } + sort.Slice(wave, func(i, j int) bool { return wave[i].Display() < wave[j].Display() }) + remaining -= len(wave) + + var next []string + sort.Strings(frontier) // deterministic `next` ordering + for _, k := range frontier { + for _, d := range dependents[k] { + indegree[d]-- + if indegree[d] == 0 { + next = append(next, d) + } + } + } + waves = append(waves, wave) + frontier = next + } + + if remaining > 0 { + stuck := make([]string, 0, remaining) + for k, d := range indegree { + if d > 0 { + stuck = append(stuck, nodes[k].Display()) + } + } + sort.Strings(stuck) + return nil, fmt.Errorf("dependency cycle among: %s", strings.Join(stuck, ", ")) + } + return waves, nil +} diff --git a/internal/plan/plan_test.go b/internal/plan/plan_test.go new file mode 100644 index 0000000..8edf61a --- /dev/null +++ b/internal/plan/plan_test.go @@ -0,0 +1,171 @@ +package plan + +import ( + "strings" + "testing" + + "github.com/openctl/openctl/pkg/protocol" +) + +// res builds a resource whose spec $refs each of the given targets ("Kind/Name" +// with a fixed apiVersion "v1" unless the target carries its own "av|Kind|Name"). +func res(apiVersion, kind, name string, refTargets ...refTarget) *protocol.Resource { + spec := map[string]any{"field": "value"} + for i, rt := range refTargets { + spec["dep"+string(rune('a'+i))] = map[string]any{ + "$ref": map[string]any{ + "apiVersion": rt.apiVersion, + "kind": rt.kind, + "name": rt.name, + "field": "status.x", + }, + } + } + return &protocol.Resource{ + APIVersion: apiVersion, + Kind: kind, + Metadata: protocol.ResourceMetadata{Name: name}, + Spec: spec, + } +} + +type refTarget struct{ apiVersion, kind, name string } + +func ref(apiVersion, kind, name string) refTarget { + return refTarget{apiVersion, kind, name} +} + +// waveDisplays flattens a plan's waves into a slice of display-label slices. +func waveDisplays(p *Plan) [][]string { + out := make([][]string, len(p.Waves)) + for i, w := range p.Waves { + for _, n := range w { + out[i] = append(out[i], n.Display()) + } + } + return out +} + +func TestBuild_Chain(t *testing.T) { + // C → B → A: waves must be [[A],[B],[C]]. + a := res("v1", "A", "a") + b := res("v1", "B", "b", ref("v1", "A", "a")) + c := res("v1", "C", "c", ref("v1", "B", "b")) + + p, err := Build([]*protocol.Resource{c, b, a}) // input order shouldn't matter + if err != nil { + t.Fatalf("Build: %v", err) + } + got := waveDisplays(p) + want := [][]string{{"A/a"}, {"B/b"}, {"C/c"}} + if !equalWaves(got, want) { + t.Fatalf("waves = %v, want %v", got, want) + } + // B records its dependency label. + if len(p.Waves[1]) != 1 || len(p.Waves[1][0].Deps) != 1 || p.Waves[1][0].Deps[0] != "A/a" { + t.Errorf("B deps = %v, want [A/a]", p.Waves[1][0].Deps) + } +} + +func TestBuild_Diamond(t *testing.T) { + // D depends on B and C; both depend on A → [[A],[B,C],[D]]. + a := res("v1", "A", "a") + b := res("v1", "B", "b", ref("v1", "A", "a")) + c := res("v1", "C", "c", ref("v1", "A", "a")) + d := res("v1", "D", "d", ref("v1", "B", "b"), ref("v1", "C", "c")) + + p, err := Build([]*protocol.Resource{a, b, c, d}) + if err != nil { + t.Fatalf("Build: %v", err) + } + got := waveDisplays(p) + want := [][]string{{"A/a"}, {"B/b", "C/c"}, {"D/d"}} // wave 2 sorted by label + if !equalWaves(got, want) { + t.Fatalf("waves = %v, want %v", got, want) + } + if d := p.Waves[2][0].Deps; len(d) != 2 || d[0] != "B/b" || d[1] != "C/c" { + t.Errorf("D deps = %v, want [B/b C/c]", d) + } +} + +func TestBuild_ExternalRef(t *testing.T) { + // A $refs a resource not in the set → recorded as external, no wave dep. + a := res("v1", "A", "a", ref("k3s.openctl.io/v1", "Cluster", "home")) + p, err := Build([]*protocol.Resource{a}) + if err != nil { + t.Fatalf("Build: %v", err) + } + if len(p.Waves) != 1 || len(p.Waves[0]) != 1 { + t.Fatalf("want a single node in one wave, got %v", waveDisplays(p)) + } + n := p.Waves[0][0] + if len(n.Deps) != 0 { + t.Errorf("external ref should not be an in-set dep, got %v", n.Deps) + } + if len(n.External) != 1 || n.External[0] != "Cluster/home" { + t.Errorf("external = %v, want [Cluster/home]", n.External) + } +} + +func TestBuild_Cycle(t *testing.T) { + a := res("v1", "A", "a", ref("v1", "B", "b")) + b := res("v1", "B", "b", ref("v1", "A", "a")) + _, err := Build([]*protocol.Resource{a, b}) + if err == nil || !strings.Contains(err.Error(), "cycle") { + t.Fatalf("want a cycle error, got %v", err) + } + if !strings.Contains(err.Error(), "A/a") || !strings.Contains(err.Error(), "B/b") { + t.Errorf("cycle error should name both nodes, got %v", err) + } +} + +func TestBuild_DuplicateIdentity(t *testing.T) { + a1 := res("v1", "A", "a") + a2 := res("v1", "A", "a") + if _, err := Build([]*protocol.Resource{a1, a2}); err == nil { + t.Fatal("want a duplicate-identity error") + } +} + +func TestBuild_SameKindNameDifferentAPIVersion(t *testing.T) { + // Two resources sharing Kind+Name but distinct apiVersion are NOT duplicates + // and don't collide (full-triple key). + a := res("prov-a/v1", "Thing", "x") + b := res("prov-b/v1", "Thing", "x") + p, err := Build([]*protocol.Resource{a, b}) + if err != nil { + t.Fatalf("Build: %v", err) + } + if p.Count() != 2 { + t.Errorf("count = %d, want 2", p.Count()) + } +} + +func TestBuild_NoRefsAllOneWave(t *testing.T) { + p, err := Build([]*protocol.Resource{ + res("v1", "A", "a"), res("v1", "B", "b"), res("v1", "C", "c"), + }) + if err != nil { + t.Fatalf("Build: %v", err) + } + if len(p.Waves) != 1 || len(p.Waves[0]) != 3 { + t.Fatalf("independent resources should share one wave, got %v", waveDisplays(p)) + } +} + +func equalWaves(a, b [][]string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if len(a[i]) != len(b[i]) { + return false + } + for j := range a[i] { + if a[i][j] != b[i][j] { + return false + } + } + } + return true +}