From 4ad99300d62d651075a5bf42a73b051c435fbe27 Mon Sep 17 00:00:00 2001 From: Pushpalanka Jayawardhana Date: Wed, 24 Jun 2026 23:45:32 +0200 Subject: [PATCH] feat(builder): add nested_data_files option to reduce OPA agent memory pressure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a system source has multiple HTTP/S3 datasources, OPA's bundle.Write collapses all data into a single root /data.json in the bundle tar. OPA agents must parse and hold the entire merged document in memory at once, causing memory pressure on systems with large datasources. Add a per-bundle flag options.nested_data_files (default: false). When enabled, the builder writes each datasource's data as a separate data.json at its namespaced path within the tar (e.g. /teams/data.json, /admins/data.json), matching the Styra DAS bundle layout. OPA's bundle reader merges nested files into the same data document at load time, so runtime semantics are unchanged. The flag is per-bundle because OCP builds bundles for many systems with different data characteristics — systems with many small datasources may not benefit and would carry extra tar-entry overhead. --- config/schema.json | 3 + internal/config/config_test.go | 64 ++++++++++++++ pkg/builder/builder.go | 120 ++++++++++++++++++++++++++- pkg/builder/builder_test.go | 147 +++++++++++++++++++++++++++++++++ pkg/config/config.go | 1 + pkg/service/worker.go | 3 + schema.json | 3 + 7 files changed, 340 insertions(+), 1 deletion(-) diff --git a/config/schema.json b/config/schema.json index c65de2da..16eb1253 100644 --- a/config/schema.json +++ b/config/schema.json @@ -356,6 +356,9 @@ "ConfigOptions": { "additionalProperties": false, "properties": { + "nested_data_files": { + "type": "boolean" + }, "no_default_stack_mount": { "type": "boolean" }, diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 7d31f432..9b06a43c 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -739,6 +739,70 @@ func TestBundleOptionsOptimizationParsing(t *testing.T) { } } +func TestBundleOptionsNestedDataFilesParsing(t *testing.T) { + tests := []struct { + name string + config string + expected bool + }{ + { + name: "default (not specified) is false", + config: `{ + bundles: { + test: { + options: {} + } + } + }`, + expected: false, + }, + { + name: "explicitly false", + config: `{ + bundles: { + test: { + options: { + nested_data_files: false + } + } + } + }`, + expected: false, + }, + { + name: "explicitly true", + config: `{ + bundles: { + test: { + options: { + nested_data_files: true + } + } + } + }`, + expected: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg, err := config.Parse([]byte(tt.config)) + if err != nil { + t.Fatalf("Parse() error = %v", err) + } + + bundle := cfg.Bundles["test"] + if bundle == nil { + t.Fatal("Expected bundle 'test' to exist") + } + + if bundle.Options.NestedDataFiles != tt.expected { + t.Errorf("Expected nested_data_files %v, got %v", tt.expected, bundle.Options.NestedDataFiles) + } + }) + } +} + // TODO(sr): replace with new(i) when 1.26 is in go.mod func intPtr(i int) *int { return &i diff --git a/pkg/builder/builder.go b/pkg/builder/builder.go index 60c3d542..8247add3 100644 --- a/pkg/builder/builder.go +++ b/pkg/builder/builder.go @@ -1,8 +1,10 @@ package builder import ( + "archive/tar" "bytes" "cmp" + "compress/gzip" "context" "encoding/json" "errors" @@ -170,6 +172,7 @@ type Builder struct { excluded []string target string optimizationLevel int + nestedDataFiles bool revision string revisionFunc func(fs.FS) (string, error) } @@ -203,6 +206,11 @@ func (b *Builder) WithOptimizationLevel(level int) *Builder { return b } +func (b *Builder) WithNestedDataFiles(enabled bool) *Builder { + b.nestedDataFiles = enabled + return b +} + func (b *Builder) WithRevision(revision string) *Builder { b.revision = revision return b @@ -456,7 +464,8 @@ func (b *Builder) Build(ctx context.Context) error { } } - fsBuild := mountfs.New(buildSources.fs()) + bsFSMap := buildSources.fs() + fsBuild := mountfs.New(bsFSMap) paths := slices.Collect(maps.Keys(fsBuild)) if b.revisionFunc != nil { @@ -499,9 +508,118 @@ func (b *Builder) Build(ctx context.Context) error { result.Manifest.SetRegoVersion(effectiveRegoVersion) result.Manifest.Revision = b.revision + if b.nestedDataFiles { + return writeBundle(b.output, *result, bsFSMap) + } return bundle.Write(b.output, *result) } +// writeBundle serializes b as a .tar.gz bundle where data files are written as +// individual nested data.json entries (one per source subdirectory) rather than +// a single merged root /data.json. This matches the Styra DAS bundle layout and +// avoids loading all data into memory as one monolithic document in OPA agents. +// +// sourceFSes is the map[escapedSourceName]fs.FS produced by buildSources.fs() +// before the mountfs is assembled. Each sub-FS is walked for data files; rego +// modules are taken from the compiled bundle result so their paths already +// include mount/prefix transformations. +func writeBundle(w io.Writer, b bundle.Bundle, sourceFSes map[string]fs.FS) error { + gw := gzip.NewWriter(w) + tw := tar.NewWriter(gw) + + writeTar := func(path string, data []byte) error { + if !strings.HasPrefix(path, "/") { + path = "/" + path + } + hdr := &tar.Header{ + Name: path, + Mode: 0600, + Typeflag: tar.TypeReg, + Size: int64(len(data)), + } + if err := tw.WriteHeader(hdr); err != nil { + return err + } + _, err := tw.Write(data) + return err + } + + // 1. Manifest + manifestBytes, err := json.Marshal(b.Manifest) + if err != nil { + return fmt.Errorf("bundle manifest: %w", err) + } + if err := writeTar("/.manifest", manifestBytes); err != nil { + return fmt.Errorf("bundle manifest write: %w", err) + } + + // 2. Data files — walk each source sub-FS and write data.json/yaml/yml at + // their natural nested paths (relative to the sub-FS root, not the top-level + // mount). This preserves e.g. /teams/data.json, /admins/data.json as separate + // entries rather than merging everything into a single /data.json. + writtenData := map[string]bool{} + names := slices.Sorted(maps.Keys(sourceFSes)) + for _, name := range names { + subFS := sourceFSes[name] + if err := fs.WalkDir(subFS, ".", func(path string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + return nil + } + base := filepath.Base(path) + if base != "data.json" && base != "data.yaml" && base != "data.yml" { + return nil + } + // path is relative to the sub-FS, e.g. "teams/data.json" + tarPath := filepath.ToSlash(path) + if writtenData[tarPath] { + return nil + } + writtenData[tarPath] = true + bs, err := fs.ReadFile(subFS, path) + if err != nil { + return fmt.Errorf("read data file %s: %w", path, err) + } + return writeTar(tarPath, bs) + }); err != nil { + return fmt.Errorf("source %s data walk: %w", name, err) + } + } + + // 3. Rego modules (compiled; paths already include source name prefix) + for _, mf := range b.Modules { + if err := writeTar(mf.Path, mf.Raw); err != nil { + return fmt.Errorf("bundle module %s: %w", mf.Path, err) + } + } + + // 4. Plan modules (target=plan/ir) + for _, pm := range b.PlanModules { + if err := writeTar(pm.Path, pm.Raw); err != nil { + return fmt.Errorf("bundle plan module %s: %w", pm.Path, err) + } + } + + // 5. Wasm modules (target=wasm) + for _, wm := range b.WasmModules { + if err := writeTar(wm.Path, wm.Raw); err != nil { + return fmt.Errorf("bundle wasm module %s: %w", wm.Path, err) + } + } + if len(b.Wasm) > 0 { + if err := writeTar("/policy.wasm", b.Wasm); err != nil { + return fmt.Errorf("bundle wasm: %w", err) + } + } + + if err := tw.Close(); err != nil { + return err + } + return gw.Close() +} + type refSet struct { refs []ast.Ref } diff --git a/pkg/builder/builder_test.go b/pkg/builder/builder_test.go index 70a4eac0..bdda3b35 100644 --- a/pkg/builder/builder_test.go +++ b/pkg/builder/builder_test.go @@ -1,7 +1,9 @@ package builder_test import ( + "archive/tar" "bytes" + "compress/gzip" "encoding/json" "errors" "fmt" @@ -900,6 +902,151 @@ func TestBuilder(t *testing.T) { } +// TestBuilderNestedDataFiles verifies WithNestedDataFiles(true) produces exactly +// one data.json per datasource path — regardless of nesting depth — and that +// the default (flag absent) still produces a single root /data.json. +func TestBuilderNestedDataFiles(t *testing.T) { + // Simulate two datasources at different path depths within a source + // directory. In production these are written by httpsync into + // /datasources//data.json. + // + // datasource 1: path="teams" → teams/data.json (depth 1) + // datasource 2: path="org/admins" → org/admins/data.json (depth 2) + // datasource 3: path="us/east/servers" → us/east/servers/data.json (depth 3) + // + // The key property: exactly one data.json per datasource, at whatever depth. + // The bundle must NOT contain intermediate or root data.json files. + files := map[string]string{ + "src0/teams/data.json": `{"alice":"admin","bob":"viewer"}`, + "src0/org/admins/data.json": `{"alice":true}`, + "src0/us/east/servers/data.json": `["web01","web02"]`, + "src0/main.rego": `package main +import rego.v1 +allow if data.teams[input.user]`, + } + + tempfs.WithTempFS(t, files, func(t *testing.T, root string) { + newSource := func() *builder.Source { + s := builder.NewSource("system") + if err := s.AddDir(builder.Dir{Path: root + "/src0"}); err != nil { + t.Fatal(err) + } + return s + } + + t.Run("nested layout when WithNestedDataFiles(true)", func(t *testing.T) { + buf := bytes.NewBuffer(nil) + if err := builder.New(). + WithSources([]*builder.Source{newSource()}). + WithNestedDataFiles(true). + WithOutput(buf). + Build(t.Context()); err != nil { + t.Fatal(err) + } + + tarEntries := tarPaths(t, buf.Bytes()) + + // Exactly one data.json per datasource, at their actual paths. + wantDataEntries := []string{ + "/teams/data.json", + "/org/admins/data.json", + "/us/east/servers/data.json", + } + for _, want := range wantDataEntries { + if !slices.Contains(tarEntries, want) { + t.Errorf("expected tar entry %q; got: %v", want, tarEntries) + } + } + + // No root /data.json — that is the flat layout we are replacing. + if slices.Contains(tarEntries, "/data.json") { + t.Errorf("nested layout must not contain root /data.json; got: %v", tarEntries) + } + + // No intermediate data.json files at parent paths — there is no + // datasource configured at "org/" or "us/" or "us/east/". + unwantedIntermediates := []string{ + "/org/data.json", + "/us/data.json", + "/us/east/data.json", + } + for _, unwanted := range unwantedIntermediates { + if slices.Contains(tarEntries, unwanted) { + t.Errorf("must not contain intermediate %q; got: %v", unwanted, tarEntries) + } + } + + // Count: exactly 3 data entries (one per datasource). + var dataEntries []string + for _, e := range tarEntries { + if strings.HasSuffix(e, "/data.json") || e == "/data.json" { + dataEntries = append(dataEntries, e) + } + } + if len(dataEntries) != 3 { + t.Errorf("expected exactly 3 data.json entries, got %d: %v", len(dataEntries), dataEntries) + } + + // Semantic correctness: round-trip through bundle.NewReader must + // produce the correct merged data document. + b, err := bundle.NewReader(bytes.NewReader(buf.Bytes())).Read() + if err != nil { + t.Fatal(err) + } + gotJSON, _ := json.Marshal(b.Data) + wantJSON := `{"org":{"admins":{"alice":true}},"teams":{"alice":"admin","bob":"viewer"},"us":{"east":{"servers":["web01","web02"]}}}` + if string(gotJSON) != wantJSON { + t.Errorf("merged data mismatch\ngot: %s\nwant: %s", gotJSON, wantJSON) + } + }) + + t.Run("flat layout by default (WithNestedDataFiles not called)", func(t *testing.T) { + buf := bytes.NewBuffer(nil) + if err := builder.New(). + WithSources([]*builder.Source{newSource()}). + WithOutput(buf). + Build(t.Context()); err != nil { + t.Fatal(err) + } + + tarEntries := tarPaths(t, buf.Bytes()) + + if !slices.Contains(tarEntries, "/data.json") { + t.Errorf("flat layout must contain root /data.json; got: %v", tarEntries) + } + for _, unwanted := range []string{ + "/teams/data.json", + "/org/admins/data.json", + "/us/east/servers/data.json", + } { + if slices.Contains(tarEntries, unwanted) { + t.Errorf("flat layout must not contain %q; got: %v", unwanted, tarEntries) + } + } + }) + }) +} + +// tarPaths returns all file paths found in a .tar.gz buffer. +func tarPaths(t *testing.T, data []byte) []string { + t.Helper() + gr, err := gzip.NewReader(bytes.NewReader(data)) + if err != nil { + t.Fatal(err) + } + defer gr.Close() + tr := tar.NewReader(gr) + var paths []string + for { + hdr, err := tr.Next() + if err != nil { + break + } + paths = append(paths, hdr.Name) + } + return paths +} + func trimLeadingWhitespace(input string) string { lines := strings.Split(input, "\n") for i, line := range lines { diff --git a/pkg/config/config.go b/pkg/config/config.go index dcd4ce7a..006037fe 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -179,6 +179,7 @@ type Options struct { NoDefaultStackMount bool `json:"no_default_stack_mount"` Optimization *Optimization `json:"optimization,omitempty"` Target string `json:"target,omitzero" enum:"rego,ir,plan,wasm"` + NestedDataFiles bool `json:"nested_data_files"` _ struct{} `additionalProperties:"false"` } diff --git a/pkg/service/worker.go b/pkg/service/worker.go index c4cd80f5..0d6f3218 100644 --- a/pkg/service/worker.go +++ b/pkg/service/worker.go @@ -219,6 +219,9 @@ func (w *BundleWorker) Execute(ctx context.Context) time.Time { if w.bundleConfig.Options.Optimization != nil { b = b.WithOptimizationLevel(w.bundleConfig.Options.Optimization.Level) } + if w.bundleConfig.Options.NestedDataFiles { + b = b.WithNestedDataFiles(true) + } if err := b.Build(ctx); err != nil { w.log.Warnf("failed to build a bundle %q: %v", w.bundleConfig.Name, err) diff --git a/schema.json b/schema.json index c65de2da..16eb1253 100755 --- a/schema.json +++ b/schema.json @@ -356,6 +356,9 @@ "ConfigOptions": { "additionalProperties": false, "properties": { + "nested_data_files": { + "type": "boolean" + }, "no_default_stack_mount": { "type": "boolean" },