Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -886,6 +886,12 @@ are rejected rather than silently combined. `-dump` is mutually exclusive with
the single-schema and manifest inputs, and `-glob` accepts the same
comma-separated filename patterns as `drift`.

At startup, manifest compositions and dump node models load concurrently in a
bounded pool of up to 32 workers. Progress is printed to stderr while loading,
so a large fleet shows that work is advancing without interfering with other
command output. The browser retains deterministic manifest/file ordering even
when models finish loading in a different order.

The dump-node list links to **Review object differences**, a fleet-wide
inventory of every qualified object found in the loaded dumps. It uses the same
semantic diff as the CLI, summarizes uniform and differing schemas, groups
Expand Down
23 changes: 16 additions & 7 deletions cmd/hclexp/drift.go
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,14 @@ func renderDriftText(w io.Writer, doc hclload.DriftJSON, details bool) {
// from the file's node{} block when present and otherwise from the filename.
// An empty glob matches every .hcl file.
func loadDriftNodes(dir, glob string) ([]driftNode, error) {
return loadDriftNodesWithOptions(dir, glob, 1, parallelLoadProgress{})
}

func loadDriftNodesWithProgress(dir, glob string, progress parallelLoadProgress) ([]driftNode, error) {
return loadDriftNodesWithOptions(dir, glob, parallelLoadLimit, progress)
}

func loadDriftNodesWithOptions(dir, glob string, parallelism int, progress parallelLoadProgress) ([]driftNode, error) {
globs := splitList(glob)
if len(globs) == 0 {
globs = []string{"*"}
Expand All @@ -193,23 +201,25 @@ func loadDriftNodes(dir, glob string) ([]driftNode, error) {
if err != nil {
return nil, fmt.Errorf("read dir %q: %w", dir, err)
}
var nodes []driftNode
var paths []string
for _, e := range entries {
if e.IsDir() || filepath.Ext(e.Name()) != ".hcl" {
continue
}
if !matchesAny(globs, e.Name()) {
continue
}
path := filepath.Join(dir, e.Name())
paths = append(paths, filepath.Join(dir, e.Name()))
}
return loadInParallel(paths, parallelism, progress, func(path string) (driftNode, error) {
schema, err := loadDumpSchema(path)
if err != nil {
return nil, fmt.Errorf("load %s: %w", path, err)
return driftNode{}, fmt.Errorf("load %s: %w", path, err)
}

n := driftNode{
File: path,
Name: strings.TrimSuffix(e.Name(), ".hcl"),
Name: strings.TrimSuffix(filepath.Base(path), ".hcl"),
Schema: schema,
}
if len(schema.Nodes) > 0 {
Expand All @@ -219,9 +229,8 @@ func loadDriftNodes(dir, glob string) ([]driftNode, error) {
n.Macros = schema.Nodes[0].Macros
}
n.Shard, n.Replica, n.Role = parseNodeIdentity(n.Name)
nodes = append(nodes, n)
}
return nodes, nil
return n, nil
})
}

// loadDumpSchema accepts snapshots captured by users that can use a
Expand Down
114 changes: 114 additions & 0 deletions cmd/hclexp/parallel_load.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
package main

import (
"fmt"
"io"
"sync"
)

const parallelLoadLimit = 32

// parallelLoadProgress describes optional human-readable progress. Writer is
// normally os.Stderr at a CLI boundary and nil for library-style callers and
// tests that do not want status output.
type parallelLoadProgress struct {
Writer io.Writer
Prefix string
Label string
}

// loadInParallel applies loader through a bounded worker pool. Results and
// errors retain input order regardless of completion order; when several loads
// fail, the caller receives the first error in deterministic input order.
func loadInParallel[T, R any](items []T, parallelism int, progress parallelLoadProgress, loader func(T) (R, error)) ([]R, error) {
if len(items) == 0 {
return []R{}, nil
}
if parallelism < 1 {
parallelism = 1
}
if parallelism > len(items) {
parallelism = len(items)
}

reporter := newParallelLoadReporter(progress, len(items), parallelism)
results := make([]R, len(items))
errs := make([]error, len(items))
jobs := make(chan int, len(items))
for i := range items {
jobs <- i
}
close(jobs)

var workers sync.WaitGroup
workers.Add(parallelism)
for range parallelism {
go func() {
defer workers.Done()
for i := range jobs {
results[i], errs[i] = loader(items[i])
reporter.complete(errs[i])
}
}()
}
workers.Wait()
for _, err := range errs {
if err != nil {
return nil, err
}
}
return results, nil
}

type parallelLoadReporter struct {
progress parallelLoadProgress
total int
every int
next int
completed int
failed int
mu sync.Mutex
}

func newParallelLoadReporter(progress parallelLoadProgress, total, workers int) *parallelLoadReporter {
r := &parallelLoadReporter{
progress: progress,
total: total,
every: max(1, (total+9)/10),
}
r.next = r.every
if progress.Writer != nil && total > 1 {
fmt.Fprintf(progress.Writer, "%s: loading %d %s in parallel (%d workers)\n",
progress.Prefix, total, progress.Label, workers)
}
return r
}

func (r *parallelLoadReporter) complete(err error) {
if r.progress.Writer == nil || r.total <= 1 {
return
}
r.mu.Lock()
defer r.mu.Unlock()
r.completed++
if err != nil {
r.failed++
}
if r.completed == r.total {
if r.failed == 0 {
fmt.Fprintf(r.progress.Writer, "%s: loaded %d %s\n", r.progress.Prefix, r.total, r.progress.Label)
} else {
fmt.Fprintf(r.progress.Writer, "%s: loading %s finished (%d/%d failed)\n",
r.progress.Prefix, r.progress.Label, r.failed, r.total)
}
return
}
if r.completed != 1 && r.completed < r.next {
return
}
for r.next <= r.completed {
r.next += r.every
}
fmt.Fprintf(r.progress.Writer, "%s: loading %s: %d/%d complete\n",
r.progress.Prefix, r.progress.Label, r.completed, r.total)
}
80 changes: 80 additions & 0 deletions cmd/hclexp/parallel_load_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
package main

import (
"bytes"
"errors"
"fmt"
"testing"
"time"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

func TestLoadInParallelStartsThirtyWorkersAndPreservesOrder(t *testing.T) {
items := make([]int, 30)
for i := range items {
items[i] = i
}
started := make(chan struct{}, len(items))
release := make(chan struct{})
released := false
defer func() {
if !released {
close(release)
}
}()

type result struct {
values []string
err error
}
var status bytes.Buffer
done := make(chan result, 1)
go func() {
values, err := loadInParallel(items, parallelLoadLimit, parallelLoadProgress{
Writer: &status,
Prefix: "web",
Label: "dump models",
}, func(item int) (string, error) {
started <- struct{}{}
<-release
return fmt.Sprintf("node-%02d", item), nil
})
done <- result{values: values, err: err}
}()

for range items {
select {
case <-started:
case <-time.After(time.Second):
require.FailNow(t, "all 30 loaders did not start concurrently")
}
}
close(release)
released = true
got := <-done
require.NoError(t, got.err)
for i, value := range got.values {
assert.Equal(t, fmt.Sprintf("node-%02d", i), value)
}
assert.Contains(t, status.String(), "web: loading 30 dump models in parallel (30 workers)")
assert.Contains(t, status.String(), "web: loading dump models: 1/30 complete")
assert.Contains(t, status.String(), "web: loaded 30 dump models")
}

func TestLoadInParallelReturnsFirstInputError(t *testing.T) {
var status bytes.Buffer
_, err := loadInParallel([]int{0, 1, 2, 3}, parallelLoadLimit, parallelLoadProgress{
Writer: &status,
Prefix: "web",
Label: "models",
}, func(item int) (int, error) {
if item == 1 || item == 3 {
return 0, fmt.Errorf("item %d: %w", item, errors.New("broken"))
}
return item, nil
})
require.EqualError(t, err, "item 1: broken")
assert.Contains(t, status.String(), "web: loading models finished (2/4 failed)")
}
57 changes: 45 additions & 12 deletions cmd/hclexp/web_manifest.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package main
import (
"fmt"
"html/template"
"io"
"io/fs"
"log/slog"
"net/http"
Expand Down Expand Up @@ -181,34 +182,53 @@ func newMultiServer(title, heading, empty string) (*multiServer, error) {
// each, plus the top-level list. layerRoot prefixes the manifest's layer paths;
// reloadInterval (when > 0) arms per-schema auto-reload.
func buildMultiServer(comps []composition, layerRoot string, reloadInterval time.Duration) (*multiServer, error) {
return buildMultiServerWithProgress(comps, layerRoot, reloadInterval, nil)
}

func buildMultiServerWithProgress(comps []composition, layerRoot string, reloadInterval time.Duration, status io.Writer) (*multiServer, error) {
ms, err := newMultiServer("Schemas", "Schemas", "No schemas in this manifest.")
if err != nil {
return nil, err
}

groups := map[string]*envGroup{}
var envOrder []string

for _, c := range comps {
type loadedComposition struct {
composition composition
server *webServer
}
loaded, err := loadInParallel(comps, parallelLoadLimit, parallelLoadProgress{
Writer: status,
Prefix: "web",
Label: "manifest models",
}, func(c composition) (loadedComposition, error) {
stack := make([]string, len(c.Layers))
for i, l := range c.Layers {
stack[i] = filepath.Join(layerRoot, l)
}
layers := strings.Join(stack, ",")
schema, err := loadSide(layers)
if err != nil {
return nil, fmt.Errorf("compose %s/%s: %w", c.Env, c.Role, err)
return loadedComposition{}, fmt.Errorf("compose %s/%s: %w", c.Env, c.Role, err)
}
srv, err := newWebServer(schema)
if err != nil {
return nil, fmt.Errorf("build server %s/%s: %w", c.Env, c.Role, err)
return loadedComposition{}, fmt.Errorf("build server %s/%s: %w", c.Env, c.Role, err)
}
base := schemaBasePath(c.Env, c.Role)
srv.basePath = base
srv.label = c.Env + " / " + c.Role
if reloadInterval > 0 {
srv.enableReload("", layers, reloadInterval)
}
return loadedComposition{composition: c, server: srv}, nil
})
if err != nil {
return nil, err
}

groups := map[string]*envGroup{}
var envOrder []string
for _, item := range loaded {
c, srv := item.composition, item.server
base := schemaBasePath(c.Env, c.Role)
srv.basePath = base
srv.label = c.Env + " / " + c.Role
ms.servers[base] = srv

if _, ok := groups[c.Env]; !ok {
Expand Down Expand Up @@ -239,7 +259,20 @@ func buildDumpMultiServerWithOptions(
reloadInterval time.Duration,
options hclload.DiffOptions,
) (*multiServer, error) {
nodes, err := loadDriftNodes(dir, glob)
return buildDumpMultiServerWithOptionsAndProgress(dir, glob, reloadInterval, options, nil)
}

func buildDumpMultiServerWithOptionsAndProgress(
dir, glob string,
reloadInterval time.Duration,
options hclload.DiffOptions,
status io.Writer,
) (*multiServer, error) {
nodes, err := loadDriftNodesWithProgress(dir, glob, parallelLoadProgress{
Writer: status,
Prefix: "web",
Label: "dump models",
})
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -517,7 +550,7 @@ func runWebManifest(manifestPath, env, layerRoot, addr string, reloadInterval ti
slog.Error("failed to read manifest", "file", manifestPath, "err", err)
os.Exit(1)
}
ms, err := buildMultiServer(comps, layerRoot, reloadInterval)
ms, err := buildMultiServerWithProgress(comps, layerRoot, reloadInterval, os.Stderr)
if err != nil {
slog.Error("failed to build schema browser", "err", err)
os.Exit(1)
Expand All @@ -531,7 +564,7 @@ func runWebManifest(manifestPath, env, layerRoot, addr string, reloadInterval ti

// runWebDump serves one independently browsable schema per selected node dump.
func runWebDump(dir, glob, addr string, reloadInterval time.Duration, options hclload.DiffOptions) {
ms, err := buildDumpMultiServerWithOptions(dir, glob, reloadInterval, options)
ms, err := buildDumpMultiServerWithOptionsAndProgress(dir, glob, reloadInterval, options, os.Stderr)
if err != nil {
slog.Error("failed to build dump browser", "dir", dir, "err", err)
os.Exit(1)
Expand Down
Loading