From bbd95d730c05d0f6610f75bf3365dee6f5ba9e7d Mon Sep 17 00:00:00 2001 From: orian Date: Mon, 7 Sep 2026 12:33:05 +0200 Subject: [PATCH] perf(web): load schemas in parallel with progress --- README.md | 6 ++ cmd/hclexp/drift.go | 23 +++++-- cmd/hclexp/parallel_load.go | 114 +++++++++++++++++++++++++++++++ cmd/hclexp/parallel_load_test.go | 80 ++++++++++++++++++++++ cmd/hclexp/web_manifest.go | 57 ++++++++++++---- cmd/hclexp/web_manifest_test.go | 93 ++++++++++++++++++++++++- docs/README.hcl.md | 5 ++ 7 files changed, 358 insertions(+), 20 deletions(-) create mode 100644 cmd/hclexp/parallel_load.go create mode 100644 cmd/hclexp/parallel_load_test.go diff --git a/README.md b/README.md index 5f05714..d264f06 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/cmd/hclexp/drift.go b/cmd/hclexp/drift.go index f010aa4..4e94b82 100644 --- a/cmd/hclexp/drift.go +++ b/cmd/hclexp/drift.go @@ -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{"*"} @@ -193,7 +201,7 @@ 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 @@ -201,15 +209,17 @@ func loadDriftNodes(dir, glob string) ([]driftNode, error) { 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 { @@ -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 diff --git a/cmd/hclexp/parallel_load.go b/cmd/hclexp/parallel_load.go new file mode 100644 index 0000000..2a3f8b3 --- /dev/null +++ b/cmd/hclexp/parallel_load.go @@ -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 := ¶llelLoadReporter{ + 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) +} diff --git a/cmd/hclexp/parallel_load_test.go b/cmd/hclexp/parallel_load_test.go new file mode 100644 index 0000000..99c40db --- /dev/null +++ b/cmd/hclexp/parallel_load_test.go @@ -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)") +} diff --git a/cmd/hclexp/web_manifest.go b/cmd/hclexp/web_manifest.go index 43366ec..e0e1a75 100644 --- a/cmd/hclexp/web_manifest.go +++ b/cmd/hclexp/web_manifest.go @@ -3,6 +3,7 @@ package main import ( "fmt" "html/template" + "io" "io/fs" "log/slog" "net/http" @@ -181,15 +182,24 @@ 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) @@ -197,18 +207,28 @@ func buildMultiServer(comps []composition, layerRoot string, reloadInterval time 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 { @@ -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 } @@ -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) @@ -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) diff --git a/cmd/hclexp/web_manifest_test.go b/cmd/hclexp/web_manifest_test.go index 6dfa88d..db06439 100644 --- a/cmd/hclexp/web_manifest_test.go +++ b/cmd/hclexp/web_manifest_test.go @@ -1,11 +1,16 @@ package main import ( + "bufio" + "bytes" + "context" "crypto/tls" + "fmt" "net/http" "net/http/httptest" "net/url" "os" + "os/exec" "path/filepath" "strings" "testing" @@ -115,8 +120,11 @@ func TestWebManifest_BrowseSchemas(t *testing.T) { root := manifestFixture(t) comps, err := manifestCompositions(filepath.Join(root, "manifest.hcl"), "") require.NoError(t, err) - ms, err := buildMultiServer(comps, root, 0) + var status bytes.Buffer + ms, err := buildMultiServerWithProgress(comps, root, 0, &status) require.NoError(t, err) + assert.Contains(t, status.String(), "web: loading 3 manifest models in parallel (3 workers)") + assert.Contains(t, status.String(), "web: loaded 3 manifest models") // Top-level list shows every env/role and links to each base path. code, body := getMulti(t, ms, "/") @@ -153,6 +161,89 @@ func TestWebManifest_BrowseSchemas(t *testing.T) { assert.Equal(t, http.StatusNotFound, code) } +func TestWebDump_LoadsThirtyNodeModelsWithProgress(t *testing.T) { + root := writeWebDumpNodes(t, 30) + + var status bytes.Buffer + ms, err := buildDumpMultiServerWithOptionsAndProgress(root, "*", 0, hclload.DiffOptions{}, &status) + require.NoError(t, err) + require.Len(t, ms.servers, 30) + for i := range 30 { + assert.Contains(t, ms.servers, nodeBasePath(fmt.Sprintf("node-%02d", i))) + } + assert.Contains(t, status.String(), "web: loading 30 dump models in parallel (30 workers)") + assert.Contains(t, status.String(), "web: loaded 30 dump models") +} + +func TestWebParallelLoadingCLIProcess(t *testing.T) { + if os.Getenv("HCLEXP_WEB_PARALLEL_HELPER") != "1" { + return + } + for i, arg := range os.Args { + if arg == "--" { + runWeb(os.Args[i+1:]) + return + } + } + t.Fatal("missing web CLI arguments") +} + +func TestWebDumpParallelLoadingStatusEndToEnd(t *testing.T) { + root := writeWebDumpNodes(t, 30) + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, os.Args[0], + "-test.run=^TestWebParallelLoadingCLIProcess$", "--", + "-dump", root, "-glob", "*", "-addr", "127.0.0.1:0", "-reload-interval", "0") + cmd.Env = append(os.Environ(), "HCLEXP_WEB_PARALLEL_HELPER=1") + var stdout bytes.Buffer + cmd.Stdout = &stdout + stderr, err := cmd.StderrPipe() + require.NoError(t, err) + require.NoError(t, cmd.Start()) + waited := false + defer func() { + if !waited { + _ = cmd.Process.Kill() + _ = cmd.Wait() + } + }() + + var lines []string + scanner := bufio.NewScanner(stderr) + for scanner.Scan() { + line := scanner.Text() + lines = append(lines, line) + if strings.Contains(line, "serving dump browser") { + break + } + } + require.NoError(t, scanner.Err()) + require.NoError(t, cmd.Process.Kill()) + _ = cmd.Wait() + waited = true + + output := strings.Join(lines, "\n") + assert.Contains(t, output, "web: loading 30 dump models in parallel (30 workers)") + assert.Contains(t, output, "web: loading dump models: 1/30 complete") + assert.Contains(t, output, "web: loaded 30 dump models") + assert.Contains(t, output, "serving dump browser") + assert.Empty(t, stdout.String(), "web startup status belongs on stderr") +} + +func writeWebDumpNodes(t *testing.T, count int) string { + t.Helper() + root := t.TempDir() + for i := range count { + node := fmt.Sprintf("node-%02d", i) + writeFileT(t, filepath.Join(root, node+".hcl"), `node "`+node+`" { + macros = { cluster = "cluster-a", hostClusterRole = "data" } +} +`+tableLayer("events")) + } + return root +} + func TestWebManifest_LookupAcrossAndWithinSchemas(t *testing.T) { root := manifestFixture(t) comps, err := manifestCompositions(filepath.Join(root, "manifest.hcl"), "") diff --git a/docs/README.hcl.md b/docs/README.hcl.md index 83827a7..af26a28 100644 --- a/docs/README.hcl.md +++ b/docs/README.hcl.md @@ -1350,6 +1350,11 @@ a schema list at `/`, each schema under `/s///`. `-env` filters to a single environment; `-layer-root` prefixes the manifest's layer paths. Each composed schema auto-reloads from its own layers like the single-schema mode. +Manifest compositions and per-node dump models are loaded concurrently through +a bounded pool of up to 32 workers. Startup progress is written to stderr, and +the final browser ordering remains deterministic regardless of completion +order. + ## `view` A `view` block declares a ClickHouse **plain** (non-materialized) view — a