Proper yaml doc splitting

Signed-off-by: Kimmo Lehto <klehto@mirantis.com>

pkg/manifest/reader.go

@@ -1,6 +1,7 @@
 package manifest
 
 import (
+	"bufio"
 	"bytes"
 	"fmt"
 	"io"
@@ -10,7 +11,6 @@ import (
 	"strings"
 	"time"
 
-	log "github.com/sirupsen/logrus"
 	"gopkg.in/yaml.v2"
 )
 
@@ -64,6 +64,27 @@ func (rd *ResourceDefinition) Unmarshal(obj any) error {
 	return nil
 }
 
+func yamlDocumentSplit(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	if atEOF && len(data) == 0 {
+		return 0, nil, nil
+	}
+
+	// Look for the document separator
+	sepIndex := bytes.Index(data, []byte("\n---"))
+	if sepIndex >= 0 {
+		// Return everything up to the separator
+		return sepIndex + len("\n---"), data[:sepIndex], nil
+	}
+
+	// If at EOF, return the remaining data
+	if atEOF {
+		return len(data), data, nil
+	}
+
+	// Request more data
+	return 0, nil, nil
+}
+
 // Reader reads Kubernetes resource definitions from input streams.
 type Reader struct {
 	IgnoreErrors bool
@@ -79,32 +100,41 @@ func name(r io.Reader) string {
 
 // Parse parses Kubernetes resource definitions from the provided input stream. They are then available via the Resources() or GetResources(apiVersion, kind) methods.
 func (r *Reader) Parse(input io.Reader) error {
-	var buf bytes.Buffer
-	tee := io.TeeReader(input, &buf)
-	decoder := yaml.NewDecoder(tee)
-	for {
-		buf.Reset()
-		log.Debugf("reading next resource definition from %s", name(input))
+	scanner := bufio.NewScanner(input)
+	scanner.Split(yamlDocumentSplit)
+
+	for scanner.Scan() {
+		rawChunk := scanner.Bytes()
+
+		// Skip empty chunks
+		if len(rawChunk) == 0 {
+			continue
+		}
+
 		rd := &ResourceDefinition{}
-		if err := decoder.Decode(rd); err != nil {
-			if err == io.EOF {
-				break
-			}
+		if err := yaml.UnmarshalStrict(rawChunk, rd); err != nil {
 			if r.IgnoreErrors {
 				continue
 			}
-			return fmt.Errorf("encountered an error while parsing %s: %w", name(input), err)
+			return fmt.Errorf("failed to decode resource %s: %w", name(input), err)
 		}
+
 		if rd.APIVersion == "" || rd.Kind == "" {
 			if r.IgnoreErrors {
 				continue
 			}
-			return fmt.Errorf("missing apiVersion or kind in %s", name(input))
+			return fmt.Errorf("missing apiVersion or kind in resource %s", name(input))
 		}
-		rd.Raw = buf.Bytes()
-		rd.Origin = name(input)
+
+		// Store the raw chunk
+		rd.Raw = append([]byte{}, rawChunk...)
 		r.manifests = append(r.manifests, rd)
 	}
+
+	if err := scanner.Err(); err != nil {
+		return fmt.Errorf("error reading input: %w", err)
+	}
+
 	return nil
 }
 

pkg/manifest/reader_test.go

@@ -76,6 +76,7 @@ metadata:
 	service := r.Resources()[1]
 	assert.Equal(t, "v1", service.APIVersion, "Unexpected apiVersion for Service")
 	assert.Equal(t, "Service", service.Kind, "Unexpected kind for Service")
+	require.Len(t, service.Raw, len(input2))
 }
 
 func TestReader_FilterResources(t *testing.T) {
@@ -107,6 +108,8 @@ metadata:
 	assert.Len(t, v2Pods, 1, "Expected 1 v2 Pod to be returned")
 	assert.Equal(t, "pod1", v1Pods[0].Metadata.Name, "Unexpected name for v1 Pod")
 	assert.Equal(t, "pod2", v2Pods[0].Metadata.Name, "Unexpected name for v2 Pod")
+	assert.NotEmpty(t, v1Pods[0].Raw, "Expected raw data to be populated")
+	assert.NotEmpty(t, v2Pods[0].Raw, "Expected raw data to be populated")
 }
 
 func TestReader_GetResources(t *testing.T) {