Simplify yaml parsing and improve its preformance (#192)

* create decoder only once
* parse resources file-by-file
* a more accurate error message
* resourceFinder now accumulates resources it finds
* Using resource accumulation to simplify code
* inlineConfigMapRefs()  as resourceFinder method
* Fixes to documentation
* avoid redundant type conversions
* fully parsing each yaml doc in one iteration

Author: zivnevo

pkg/controller/error_types.go

@@ -56,7 +56,7 @@ func (err *NoYamlsFoundError) Error() string {
 }
 
 func (err *NoK8sResourcesFoundError) Error() string {
-	return "no relevant Kubernetes resources found"
+	return "could not find any Kubernetes workload resources"
 }
 
 func (err *ConfigMapNotFoundError) Error() string {

pkg/controller/policies_synthesizer.go

@@ -134,32 +134,29 @@ func (ps *PoliciesSynthesizer) ConnectionsFromFolderPaths(dirPaths []string) ([]
 func (ps *PoliciesSynthesizer) extractConnections(dirPaths []string) ([]common.Resource, []*common.Connections, []FileProcessingError) {
 	// 1. Get all relevant resources from the repo
 	resFinder := resourceFinder{logger: ps.logger, stopOn1stErr: ps.stopOnError, walkFn: ps.walkFn}
-	rawResources := []rawResourcesInFile{}
 	fileErrors := []FileProcessingError{}
 	for _, dirPath := range dirPaths {
-		dObjs, errs := resFinder.getRelevantK8sResources(dirPath)
-		rawResources = append(rawResources, dObjs...)
+		errs := resFinder.getRelevantK8sResources(dirPath)
 		fileErrors = append(fileErrors, errs...)
 		if stopProcessing(ps.stopOnError, errs) {
 			return nil, nil, fileErrors
 		}
 	}
-	if len(rawResources) == 0 {
+	if len(resFinder.workloads) == 0 {
 		fileErrors = appendAndLogNewError(fileErrors, noK8sResourcesFound(), ps.logger)
 		return []common.Resource{}, []*common.Connections{}, fileErrors
 	}
 
-	// 2. Parse them into internal structs
-	resParser := resourceParser{logger: ps.logger}
-	resources, links, parseErrors := resParser.parseResources(rawResources)
-	fileErrors = append(fileErrors, parseErrors...)
+	// 2. Inline configmaps values as workload envs
+	errs := resFinder.inlineConfigMapRefsAsEnvs()
+	fileErrors = append(fileErrors, errs...)
 	if stopProcessing(ps.stopOnError, fileErrors) {
 		return nil, nil, fileErrors
 	}
 
 	// 3. Discover all connections between resources
-	connections := discoverConnections(resources, links, ps.logger)
-	return resources, connections, fileErrors
+	connections := discoverConnections(resFinder.workloads, resFinder.services, ps.logger)
+	return resFinder.workloads, connections, fileErrors
 }
 
 func hasFatalError(errs []FileProcessingError) error {

pkg/controller/resource_finder.go

@@ -10,6 +10,9 @@ import (
 
 	"gopkg.in/yaml.v3"
 	"k8s.io/client-go/kubernetes/scheme"
+
+	"github.com/np-guard/cluster-topology-analyzer/pkg/analyzer"
+	"github.com/np-guard/cluster-topology-analyzer/pkg/common"
 )
 
 // K8s resources that are relevant for connectivity analysis
@@ -18,8 +21,8 @@ const (
 	replicaSet            string = "ReplicaSet"
 	replicationController string = "ReplicationController"
 	deployment            string = "Deployment"
-	statefulset           string = "StatefulSet"
-	daemonset             string = "DaemonSet"
+	statefulSet           string = "StatefulSet"
+	daemonSet             string = "DaemonSet"
 	job                   string = "Job"
 	cronJob               string = "CronTab"
 	service               string = "Service"
@@ -28,57 +31,48 @@ const (
 
 var (
 	acceptedK8sTypesRegex = fmt.Sprintf("(^%s$|^%s$|^%s$|^%s$|^%s$|^%s$|^%s$|^%s$|^%s$|^%s$)",
-		pod, replicaSet, replicationController, deployment, daemonset, statefulset, job, cronJob, service, configmap)
-	acceptedK8sTypes = regexp.MustCompile(acceptedK8sTypesRegex)
-	yamlSuffix       = regexp.MustCompile(".ya?ml$")
+		pod, replicaSet, replicationController, deployment, daemonSet, statefulSet, job, cronJob, service, configmap)
+	acceptedK8sTypes   = regexp.MustCompile(acceptedK8sTypesRegex)
+	yamlSuffix         = regexp.MustCompile(".ya?ml$")
+	k8sResourceDecoder = scheme.Codecs.UniversalDeserializer()
 )
 
-// rawResourcesInFile represents a single YAML file with multiple K8s resources
-type rawResourcesInFile struct {
-	ManifestFilepath string
-	rawK8sResources  []rawK8sResource
-}
-
-// rawK8sResource stores a raw K8s resource and its kind for later parsing
-type rawK8sResource struct {
-	GroupKind     string
-	RuntimeObject []byte
-}
-
-// resourceFinder is used to locate all relevant K8s resources in a given file-system directory
+// resourceFinder is used to locate all relevant K8s resources in given file-system directories
+// and to convert them into the internal structs, used for later processing.
 type resourceFinder struct {
 	logger       Logger
 	stopOn1stErr bool
 	walkFn       WalkFunction // for customizing directory scan
+
+	workloads  []common.Resource // accumulates all workload resources found
+	services   []common.Service  // accumulates all service resources found
+	configmaps []common.CfgMap   // accumulates all ConfigMap resources found
 }
 
 // getRelevantK8sResources is the main function of resourceFinder.
 // It scans a given directory using walkFn, looking for all yaml files. It then breaks each yaml into its documents
 // and extracts all K8s resources that are relevant for connectivity analysis.
-// The result is stored in a slice of rawResourcesInFile (one per yaml file), each containing a slice of rawK8sResource
-func (rf *resourceFinder) getRelevantK8sResources(repoDir string) ([]rawResourcesInFile, []FileProcessingError) {
+// The resources are stored in the struct, separated to workloads, services and configmaps
+func (rf *resourceFinder) getRelevantK8sResources(repoDir string) []FileProcessingError {
 	manifestFiles, fileScanErrors := rf.searchForManifests(repoDir)
 	if stopProcessing(rf.stopOn1stErr, fileScanErrors) {
-		return nil, fileScanErrors
+		return fileScanErrors
 	}
 	if len(manifestFiles) == 0 {
 		fileScanErrors = appendAndLogNewError(fileScanErrors, noYamlsFound(), rf.logger)
-		return nil, fileScanErrors
+		return fileScanErrors
 	}
 
-	parsedObjs := []rawResourcesInFile{}
 	for _, mfp := range manifestFiles {
-		rawK8sResources, err := rf.parseK8sYaml(mfp)
-		fileScanErrors = append(fileScanErrors, err...)
+		relMfp := pathWithoutBaseDir(mfp, repoDir)
+		errs := rf.parseK8sYaml(mfp, relMfp)
+		fileScanErrors = append(fileScanErrors, errs...)
 		if stopProcessing(rf.stopOn1stErr, fileScanErrors) {
-			return nil, fileScanErrors
-		}
-		if len(rawK8sResources) > 0 {
-			manifestFilePath := pathWithoutBaseDir(mfp, repoDir)
-			parsedObjs = append(parsedObjs, rawResourcesInFile{rawK8sResources: rawK8sResources, ManifestFilepath: manifestFilePath})
+			return fileScanErrors
 		}
 	}
-	return parsedObjs, fileScanErrors
+
+	return fileScanErrors
 }
 
 // searchForManifests returns a list of YAML files under a given directory (recursively)
@@ -105,14 +99,14 @@ func (rf *resourceFinder) searchForManifests(repoDir string) ([]string, []FilePr
 }
 
 // splitByYamlDocuments takes a YAML file and returns a slice containing its documents as raw text
-func (rf *resourceFinder) splitByYamlDocuments(mfp string) ([]string, []FileProcessingError) {
+func (rf *resourceFinder) splitByYamlDocuments(mfp string) ([][]byte, []FileProcessingError) {
 	fileBuf, err := os.ReadFile(mfp)
 	if err != nil {
-		return []string{}, appendAndLogNewError(nil, failedReadingFile(mfp, err), rf.logger)
+		return nil, appendAndLogNewError(nil, failedReadingFile(mfp, err), rf.logger)
 	}
 
 	decoder := yaml.NewDecoder(bytes.NewBuffer(fileBuf))
-	documents := []string{}
+	documents := [][]byte{}
 	documentID := 0
 	for {
 		var doc yaml.Node
@@ -127,39 +121,67 @@ func (rf *resourceFinder) splitByYamlDocuments(mfp string) ([]string, []FileProc
 			if err != nil {
 				return documents, appendAndLogNewError(nil, malformedYamlDoc(mfp, doc.Line, documentID, err), rf.logger)
 			}
-			documents = append(documents, string(out))
+			documents = append(documents, out)
 		}
 		documentID += 1
 	}
 	return documents, nil
 }
 
-// parseK8sYaml takes a YAML document and checks if it stands for a relevant K8s resource.
-// If yes, it puts it into a rawK8sResource and appends it to the result.
-func (rf *resourceFinder) parseK8sYaml(mfp string) ([]rawK8sResource, []FileProcessingError) {
-	dObjs := []rawK8sResource{}
-	sepYamlFiles, fileProcessingErrors := rf.splitByYamlDocuments(mfp)
+// parseK8sYaml takes a YAML file and attempts to parse each of its documents into
+// one of the relevant k8s resources
+func (rf *resourceFinder) parseK8sYaml(mfp, relMfp string) []FileProcessingError {
+	yamlDocs, fileProcessingErrors := rf.splitByYamlDocuments(mfp)
 	if stopProcessing(rf.stopOn1stErr, fileProcessingErrors) {
-		return nil, fileProcessingErrors
+		return fileProcessingErrors
 	}
 
-	for docID, doc := range sepYamlFiles {
-		decode := scheme.Codecs.UniversalDeserializer().Decode
-		_, groupVersionKind, err := decode([]byte(doc), nil, nil)
+	for docID, doc := range yamlDocs {
+		_, groupVersionKind, err := k8sResourceDecoder.Decode(doc, nil, nil)
 		if err != nil {
-			fileProcessingErrors = appendAndLogNewError(fileProcessingErrors, notK8sResource(mfp, docID, err), rf.logger)
+			fileProcessingErrors = appendAndLogNewError(fileProcessingErrors, notK8sResource(relMfp, docID, err), rf.logger)
 			continue
 		}
 		if !acceptedK8sTypes.MatchString(groupVersionKind.Kind) {
-			rf.logger.Infof("in file: %s, document: %d, skipping object with type: %s", mfp, docID, groupVersionKind.Kind)
+			rf.logger.Infof("in file: %s, document: %d, skipping object with type: %s", relMfp, docID, groupVersionKind.Kind)
 		} else {
-			d := rawK8sResource{}
-			d.GroupKind = groupVersionKind.Kind
-			d.RuntimeObject = []byte(doc)
-			dObjs = append(dObjs, d)
+			kind := groupVersionKind.Kind
+			err = rf.parseResource(kind, doc, relMfp)
+			if err != nil {
+				fileProcessingErrors = appendAndLogNewError(fileProcessingErrors, failedScanningResource(kind, relMfp, err), rf.logger)
+			}
+		}
+	}
+	return fileProcessingErrors
+}
+
+// parseResource takes a yaml document, parses it into a K8s resource and puts it into one of the 3 struct slices:
+// the workload resource slice, the Service resource slice, and the ConfigMaps resource slice
+func (rf *resourceFinder) parseResource(kind string, yamlDoc []byte, manifestFilePath string) error {
+	switch kind {
+	case service:
+		res, err := analyzer.ScanK8sServiceObject(kind, yamlDoc)
+		if err != nil {
+			return err
 		}
+		res.Resource.FilePath = manifestFilePath
+		rf.services = append(rf.services, res)
+	case configmap:
+		res, err := analyzer.ScanK8sConfigmapObject(kind, yamlDoc)
+		if err != nil {
+			return err
+		}
+		rf.configmaps = append(rf.configmaps, res)
+	default:
+		res, err := analyzer.ScanK8sWorkloadObject(kind, yamlDoc)
+		if err != nil {
+			return err
+		}
+		res.Resource.FilePath = manifestFilePath
+		rf.workloads = append(rf.workloads, res)
 	}
-	return dObjs, fileProcessingErrors
+
+	return nil
 }
 
 // returns a file path without its prefix base dir
@@ -174,3 +196,49 @@ func pathWithoutBaseDir(path, baseDir string) string {
 	}
 	return relPath
 }
+
+// inlineConfigMapRefsAsEnvs appends to the Envs of each given resource the ConfigMap values it is referring to
+// It should only be called after ALL calls to getRelevantK8sResources successfully returned
+func (rf *resourceFinder) inlineConfigMapRefsAsEnvs() []FileProcessingError {
+	cfgMapsByName := map[string]*common.CfgMap{}
+	for cm := range rf.configmaps {
+		cfgMapsByName[rf.configmaps[cm].FullName] = &rf.configmaps[cm]
+	}
+
+	parseErrors := []FileProcessingError{}
+	for idx := range rf.workloads {
+		res := &rf.workloads[idx]
+
+		// inline the envFrom field in PodSpec->containers
+		for _, cfgMapRef := range res.Resource.ConfigMapRefs {
+			configmapFullName := res.Resource.Namespace + "/" + cfgMapRef
+			if cfgMap, ok := cfgMapsByName[configmapFullName]; ok {
+				for _, v := range cfgMap.Data {
+					if analyzer.IsNetworkAddressValue(v) {
+						res.Resource.NetworkAddrs = append(res.Resource.NetworkAddrs, v)
+					}
+				}
+			} else {
+				parseErrors = appendAndLogNewError(parseErrors, configMapNotFound(configmapFullName, res.Resource.Name), rf.logger)
+			}
+		}
+
+		// inline PodSpec->container->env->valueFrom->configMapKeyRef
+		for _, cfgMapKeyRef := range res.Resource.ConfigMapKeyRefs {
+			configmapFullName := res.Resource.Namespace + "/" + cfgMapKeyRef.Name
+			if cfgMap, ok := cfgMapsByName[configmapFullName]; ok {
+				if val, ok := cfgMap.Data[cfgMapKeyRef.Key]; ok {
+					if analyzer.IsNetworkAddressValue(val) {
+						res.Resource.NetworkAddrs = append(res.Resource.NetworkAddrs, val)
+					}
+				} else {
+					err := configMapKeyNotFound(cfgMapKeyRef.Name, cfgMapKeyRef.Key, res.Resource.Name)
+					parseErrors = appendAndLogNewError(parseErrors, err, rf.logger)
+				}
+			} else {
+				parseErrors = appendAndLogNewError(parseErrors, configMapNotFound(configmapFullName, res.Resource.Name), rf.logger)
+			}
+		}
+	}
+	return parseErrors
+}

pkg/controller/resource_finder_test.go

@@ -13,7 +13,7 @@ import (
 func TestGetRelevantK8sResourcesBadYamlDocument(t *testing.T) {
 	dirPath := filepath.Join(getTestsDir(), "bad_yamls", "document_with_syntax_error.yaml")
 	resFinder := resourceFinder{logger: NewDefaultLogger(), stopOn1stErr: false, walkFn: filepath.WalkDir}
-	objs, errs := resFinder.getRelevantK8sResources(dirPath)
+	errs := resFinder.getRelevantK8sResources(dirPath)
 	require.Len(t, errs, 1)
 	badDoc := &MalformedYamlDocError{}
 	require.True(t, errors.As(errs[0].Error(), &badDoc))
@@ -22,14 +22,15 @@ func TestGetRelevantK8sResourcesBadYamlDocument(t *testing.T) {
 	require.Equal(t, 6, docID)
 	require.Nil(t, err)
 
-	require.Len(t, objs, 1)
-	require.Len(t, objs[0].rawK8sResources, 6)
+	require.Len(t, resFinder.workloads, 3)
+	require.Len(t, resFinder.services, 3)
+	require.Empty(t, resFinder.configmaps)
 }
 
 func TestGetRelevantK8sResourcesBadYamlDocumentFailFast(t *testing.T) {
 	dirPath := filepath.Join(getTestsDir(), "bad_yamls", "document_with_syntax_error.yaml")
 	resFinder := resourceFinder{logger: NewDefaultLogger(), stopOn1stErr: true, walkFn: filepath.WalkDir}
-	objs, errs := resFinder.getRelevantK8sResources(dirPath)
+	errs := resFinder.getRelevantK8sResources(dirPath)
 	require.Len(t, errs, 1)
 	badDoc := &MalformedYamlDocError{}
 	require.True(t, errors.As(errs[0].Error(), &badDoc))
@@ -38,48 +39,57 @@ func TestGetRelevantK8sResourcesBadYamlDocumentFailFast(t *testing.T) {
 	require.Equal(t, 6, docID)
 	require.Nil(t, err)
 
-	require.Empty(t, objs)
+	require.Empty(t, resFinder.workloads)
+	require.Empty(t, resFinder.services)
+	require.Empty(t, resFinder.configmaps)
 }
 
 func TestGetRelevantK8sResourcesNoK8sResource(t *testing.T) {
 	dirPath := filepath.Join(getTestsDir(), "bad_yamls", "not_a_k8s_resource.yaml")
 	resFinder := resourceFinder{logger: NewDefaultLogger(), stopOn1stErr: false, walkFn: filepath.WalkDir}
-	objs, errs := resFinder.getRelevantK8sResources(dirPath)
+	errs := resFinder.getRelevantK8sResources(dirPath)
 	require.Len(t, errs, 1)
 	notK8sRes := &NotK8sResourceError{}
 	require.True(t, errors.As(errs[0].Error(), &notK8sRes))
-	require.Len(t, objs, 1)
-	require.Len(t, objs[0].rawK8sResources, 1)
+	require.Empty(t, resFinder.workloads)
+	require.Len(t, resFinder.services, 1)
+	require.Empty(t, resFinder.configmaps)
 }
 
 func TestGetRelevantK8sResourcesNoYAMLs(t *testing.T) {
 	dirPath := filepath.Join(getTestsDir(), "bad_yamls", "subdir2")
 	resFinder := resourceFinder{logger: NewDefaultLogger(), stopOn1stErr: false, walkFn: filepath.WalkDir}
-	objs, errs := resFinder.getRelevantK8sResources(dirPath)
+	errs := resFinder.getRelevantK8sResources(dirPath)
 	require.Len(t, errs, 1)
 	noYamls := &NoYamlsFoundError{}
 	require.True(t, errors.As(errs[0].Error(), &noYamls))
-	require.Empty(t, objs)
+	require.Empty(t, resFinder.workloads)
+	require.Empty(t, resFinder.services)
+	require.Empty(t, resFinder.configmaps)
 }
 
 func TestGetRelevantK8sResourcesBadDir(t *testing.T) {
 	dirPath := filepath.Join(getTestsDir(), "bad_yamls", "subdir3") // doesn't exist
 	resFinder := resourceFinder{logger: NewDefaultLogger(), stopOn1stErr: false, walkFn: filepath.WalkDir}
-	objs, errs := resFinder.getRelevantK8sResources(dirPath)
+	errs := resFinder.getRelevantK8sResources(dirPath)
 	require.Len(t, errs, 1)
 	badDir := &FailedAccessingDirError{}
 	require.True(t, errors.As(errs[0].Error(), &badDir))
-	require.Empty(t, objs)
+	require.Empty(t, resFinder.workloads)
+	require.Empty(t, resFinder.services)
+	require.Empty(t, resFinder.configmaps)
 }
 
 func TestGetRelevantK8sResourcesBadDirFailFast(t *testing.T) {
 	dirPath := filepath.Join(getTestsDir(), "bad_yamls", "subdir3") // doesn't exist
 	resFinder := resourceFinder{logger: NewDefaultLogger(), stopOn1stErr: true, walkFn: filepath.WalkDir}
-	objs, errs := resFinder.getRelevantK8sResources(dirPath)
+	errs := resFinder.getRelevantK8sResources(dirPath)
 	require.Len(t, errs, 1)
 	badDir := &FailedAccessingDirError{}
 	require.True(t, errors.As(errs[0].Error(), &badDir))
-	require.Empty(t, objs)
+	require.Empty(t, resFinder.workloads)
+	require.Empty(t, resFinder.services)
+	require.Empty(t, resFinder.configmaps)
 }
 
 func TestSearchForManifests(t *testing.T) {