add WalkMetaFS function

This function is like WalkFS except that callers provide a function
that handles a Meta object at a time, rather than an entire file's worth
of FBC objects.

For callers that can operate on a single meta at a time, this function
will provide significant reductions in memory usage.

Signed-off-by: Joe Lanford <[email protected]>

Author: joelanford

alpha/declcfg/declcfg.go

@@ -84,6 +84,7 @@ type RelatedImage struct {
 type Meta struct {
 	Schema  string
 	Package string
+	Name    string
 
 	Blob json.RawMessage
 }
@@ -96,6 +97,7 @@ func (m *Meta) UnmarshalJSON(blob []byte) error {
 	type tmp struct {
 		Schema     string              `json:"schema"`
 		Package    string              `json:"package,omitempty"`
+		Name       string              `json:"name,omitempty"`
 		Properties []property.Property `json:"properties,omitempty"`
 	}
 	var t tmp
@@ -104,6 +106,7 @@ func (m *Meta) UnmarshalJSON(blob []byte) error {
 	}
 	m.Schema = t.Schema
 	m.Package = t.Package
+	m.Name = t.Name
 	m.Blob = blob
 	return nil
 }

alpha/declcfg/load.go

@@ -22,13 +22,48 @@ const (
 	indexIgnoreFilename = ".indexignore"
 )
 
+type WalkMetasFunc func(path string, meta *Meta, err error) error
+
+func WalkMetasFS(root fs.FS, walkFn WalkMetasFunc) error {
+	return walkFiles(root, func(root fs.FS, path string, err error) error {
+		if err != nil {
+			return walkFn(path, nil, err)
+		}
+
+		f, err := root.Open(path)
+		if err != nil {
+			return walkFn(path, nil, err)
+		}
+		defer f.Close()
+
+		return walkMetasReader(f, func(meta *Meta) error {
+			return walkFn(path, meta, nil)
+		})
+	})
+}
+
 type WalkFunc func(path string, cfg *DeclarativeConfig, err error) error
 
 // WalkFS walks root using a gitignore-style filename matcher to skip files
 // that match patterns found in .indexignore files found throughout the filesystem.
 // It calls walkFn for each declarative config file it finds. If WalkFS encounters
 // an error loading or parsing any file, the error will be immediately returned.
 func WalkFS(root fs.FS, walkFn WalkFunc) error {
+	return walkFiles(root, func(root fs.FS, path string, err error) error {
+		if err != nil {
+			return walkFn(path, nil, err)
+		}
+
+		cfg, err := LoadFile(root, path)
+		if err != nil {
+			return walkFn(path, cfg, err)
+		}
+
+		return walkFn(path, cfg, nil)
+	})
+}
+
+func walkFiles(root fs.FS, fn func(root fs.FS, path string, err error) error) error {
 	if root == nil {
 		return fmt.Errorf("no declarative config filesystem provided")
 	}
@@ -40,20 +75,15 @@ func WalkFS(root fs.FS, walkFn WalkFunc) error {
 
 	return fs.WalkDir(root, ".", func(path string, info fs.DirEntry, err error) error {
 		if err != nil {
-			return walkFn(path, nil, err)
+			return fn(root, path, err)
 		}
 		// avoid validating a directory, an .indexignore file, or any file that matches
 		// an ignore pattern outlined in a .indexignore file.
 		if info.IsDir() || info.Name() == indexIgnoreFilename || matcher.Match(path, false) {
 			return nil
 		}
 
-		cfg, err := LoadFile(root, path)
-		if err != nil {
-			return walkFn(path, cfg, err)
-		}
-
-		return walkFn(path, cfg, err)
+		return fn(root, path, nil)
 	})
 }
 
@@ -123,46 +153,35 @@ func extractCSV(objs []string) string {
 // Path references will not be de-referenced so callers are responsible for de-referencing if necessary.
 func LoadReader(r io.Reader) (*DeclarativeConfig, error) {
 	cfg := &DeclarativeConfig{}
-	dec := yaml.NewYAMLOrJSONDecoder(r, 4096)
-	for {
-		doc := json.RawMessage{}
-		if err := dec.Decode(&doc); err != nil {
-			if errors.Is(err, io.EOF) {
-				break
-			}
-			return nil, err
-		}
-		doc = []byte(strings.NewReplacer(`\u003c`, "<", `\u003e`, ">", `\u0026`, "&").Replace(string(doc)))
-
-		var in Meta
-		if err := json.Unmarshal(doc, &in); err != nil {
-			return nil, fmt.Errorf("unmarshal error: %s", resolveUnmarshalErr(doc, err))
-		}
 
+	if err := walkMetasReader(r, func(in *Meta) error {
 		switch in.Schema {
 		case SchemaPackage:
 			var p Package
-			if err := json.Unmarshal(doc, &p); err != nil {
-				return nil, fmt.Errorf("parse package: %v", err)
+			if err := json.Unmarshal(in.Blob, &p); err != nil {
+				return fmt.Errorf("parse package: %v", err)
 			}
 			cfg.Packages = append(cfg.Packages, p)
 		case SchemaChannel:
 			var c Channel
-			if err := json.Unmarshal(doc, &c); err != nil {
-				return nil, fmt.Errorf("parse channel: %v", err)
+			if err := json.Unmarshal(in.Blob, &c); err != nil {
+				return fmt.Errorf("parse channel: %v", err)
 			}
 			cfg.Channels = append(cfg.Channels, c)
 		case SchemaBundle:
 			var b Bundle
-			if err := json.Unmarshal(doc, &b); err != nil {
-				return nil, fmt.Errorf("parse bundle: %v", err)
+			if err := json.Unmarshal(in.Blob, &b); err != nil {
+				return fmt.Errorf("parse bundle: %v", err)
 			}
 			cfg.Bundles = append(cfg.Bundles, b)
 		case "":
-			return nil, fmt.Errorf("object '%s' is missing root schema field", string(doc))
+			return fmt.Errorf("object '%s' is missing root schema field", string(in.Blob))
 		default:
-			cfg.Others = append(cfg.Others, in)
+			cfg.Others = append(cfg.Others, *in)
 		}
+		return nil
+	}); err != nil {
+		return nil, err
 	}
 	return cfg, nil
 }
@@ -188,6 +207,30 @@ func LoadFile(root fs.FS, path string) (*DeclarativeConfig, error) {
 	return cfg, nil
 }
 
+func walkMetasReader(r io.Reader, fn func(*Meta) error) error {
+	dec := yaml.NewYAMLOrJSONDecoder(r, 4096)
+	for {
+		doc := json.RawMessage{}
+		if err := dec.Decode(&doc); err != nil {
+			if errors.Is(err, io.EOF) {
+				break
+			}
+			return err
+		}
+		doc = []byte(strings.NewReplacer(`\u003c`, "<", `\u003e`, ">", `\u0026`, "&").Replace(string(doc)))
+
+		var in Meta
+		if err := json.Unmarshal(doc, &in); err != nil {
+			return fmt.Errorf("unmarshal error: %s", resolveUnmarshalErr(doc, err))
+		}
+
+		if err := fn(&in); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func resolveUnmarshalErr(data []byte, err error) string {
 	var te *json.UnmarshalTypeError
 	if errors.As(err, &te) {
@@ -213,10 +256,10 @@ func formatUnmarshallErrorString(data []byte, errmsg string, offset int64) strin
 		var pOffset, origOffset int64
 		origOffset = 0
 		for origOffset = 0; origOffset < offset; {
-			pOffset++
 			if pString[pOffset] != '\n' && pString[pOffset] != ' ' {
 				origOffset++
 			}
+			pOffset++
 		}
 		_, _ = sb.WriteString(pString[:pOffset])
 		_, _ = sb.WriteString(" <== ")

alpha/declcfg/load_test.go

@@ -92,6 +92,73 @@ func TestLoadReader(t *testing.T) {
 	}
 }
 
+func TestWalkMetasFS(t *testing.T) {
+	type spec struct {
+		name              string
+		fsys              fs.FS
+		assertion         require.ErrorAssertionFunc
+		expectNumPackages int
+		expectNumChannels int
+		expectNumBundles  int
+		expectNumOthers   int
+	}
+	specs := []spec{
+		{
+			name:      "Error/NilFS",
+			fsys:      nil,
+			assertion: require.Error,
+		},
+		{
+			name:      "Error/NonExistentDir",
+			fsys:      os.DirFS("non/existent/dir/"),
+			assertion: require.Error,
+		},
+		{
+			name:      "Error/Invalid",
+			fsys:      invalidFS,
+			assertion: require.Error,
+		},
+		{
+			name:              "Success/ValidDir",
+			fsys:              validFS,
+			assertion:         require.NoError,
+			expectNumPackages: 3,
+			expectNumChannels: 0,
+			expectNumBundles:  12,
+			expectNumOthers:   1,
+		},
+	}
+
+	for _, s := range specs {
+		t.Run(s.name, func(t *testing.T) {
+			numPackages, numChannels, numBundles, numOthers := 0, 0, 0, 0
+			err := WalkMetasFS(s.fsys, func(path string, meta *Meta, err error) error {
+				if err != nil {
+					return err
+				}
+				switch meta.Schema {
+				case SchemaPackage:
+					numPackages++
+				case SchemaChannel:
+					numChannels++
+				case SchemaBundle:
+					numBundles++
+				default:
+					numOthers++
+				}
+				return nil
+			})
+			s.assertion(t, err)
+			if err == nil {
+				assert.Equal(t, s.expectNumPackages, numPackages, "unexpected package count")
+				assert.Equal(t, s.expectNumChannels, numChannels, "unexpected channel count")
+				assert.Equal(t, s.expectNumBundles, numBundles, "unexpected bundle count")
+				assert.Equal(t, s.expectNumOthers, numOthers, "unexpected others count")
+			}
+		})
+	}
+}
+
 func TestLoadFS(t *testing.T) {
 	type spec struct {
 		name      string