Conf gen updates (#262)

* return typed model instead of generic map

* exported a few more functions

* allowed missing config.yaml file

* added ConfigFileSgtruct

* restored output of Buckets

* changed config2 to config

Author: KalmanMeth

docs/api.md

@@ -145,7 +145,6 @@ Following is the supported API format for writing to standard output:
 <pre>
  stdout:
          format: the format of each line: printf (default - writes using golang's default map printing), fields (writes one key and value field per line) or json
-
 </pre>
 ## Aggregate metrics API
 Following is the supported API format for specifying metrics aggregations:

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/netobserv/loki-client-go v0.0.0-20211018150932-cb17208397a9
 	github.com/netobserv/netobserv-ebpf-agent v0.1.1-0.20220608092850-3fd4695b7cc2
 	github.com/netsampler/goflow2 v1.1.1-0.20220509155230-5300494e4785
+	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/common v0.32.1
 	github.com/segmentio/kafka-go v0.4.28
@@ -72,7 +73,6 @@ require (
 	github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f // indirect
 	github.com/pelletier/go-toml v1.9.4 // indirect
 	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect

pkg/api/encode_prom.go

@@ -18,10 +18,10 @@
 package api
 
 type PromEncode struct {
-	Metrics    PromMetricsItems `yaml:"metrics" json:"metrics,omitempty" doc:"list of prometheus metric definitions, each includes:"`
-	Port       int              `yaml:"port" json:"port,omitempty" doc:"port number to expose \"/metrics\" endpoint"`
-	Prefix     string           `yaml:"prefix" json:"prefix,omitempty" doc:"prefix added to each metric name"`
-	ExpiryTime int              `yaml:"expiryTime" json:"expiryTime,omitempty" doc:"seconds of no-flow to wait before deleting prometheus data item"`
+	Metrics    PromMetricsItems `yaml:"metrics,omitempty" json:"metrics,omitempty" doc:"list of prometheus metric definitions, each includes:"`
+	Port       int              `yaml:"port,omitempty" json:"port,omitempty" doc:"port number to expose \"/metrics\" endpoint"`
+	Prefix     string           `yaml:"prefix,omitempty" json:"prefix,omitempty" doc:"prefix added to each metric name"`
+	ExpiryTime int              `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"seconds of no-flow to wait before deleting prometheus data item"`
 }
 
 type PromEncodeOperationEnum struct {

pkg/confgen/confgen.go

@@ -71,13 +71,13 @@ type DefFile struct {
 
 func (cg *ConfGen) Run() error {
 	var err error
-	cg.config, err = cg.parseConfigFile(Opt.SrcFolder + "/" + configFileName)
+	cg.config, err = cg.ParseConfigFile(Opt.SrcFolder + "/" + configFileName)
 	if err != nil {
-		log.Debugf("cg.parseConfigFile err: %v ", err)
+		log.Debugf("cg.ParseConfigFile err: %v ", err)
 		return err
 	}
 
-	definitionFiles := cg.getDefinitionFiles(Opt.SrcFolder)
+	definitionFiles := cg.GetDefinitionFiles(Opt.SrcFolder)
 	for _, definitionFile := range definitionFiles {
 		err := cg.parseFile(definitionFile)
 		if err != nil {
@@ -86,7 +86,7 @@ func (cg *ConfGen) Run() error {
 		}
 	}
 
-	cg.dedupe()
+	cg.Dedupe()
 
 	if len(Opt.GenerateStages) != 0 {
 		config := cg.GenerateTruncatedConfig(Opt.GenerateStages)
@@ -215,7 +215,7 @@ func (cg *ConfGen) parseFile(fileName string) error {
 	return nil
 }
 
-func (*ConfGen) getDefinitionFiles(rootPath string) []string {
+func (*ConfGen) GetDefinitionFiles(rootPath string) []string {
 
 	var files []string
 

pkg/confgen/confgen_test.go

@@ -113,7 +113,7 @@ func Test_getDefinitionFiles(t *testing.T) {
 	require.NoError(t, err)
 	err = os.WriteFile(filepath.Join(dirPath, filename), []byte(networkDefinitionConfiguration), 0644)
 	require.NoError(t, err)
-	files := cg.getDefinitionFiles(dirPath)
+	files := cg.GetDefinitionFiles(dirPath)
 	require.Equal(t, 1, len(files))
 	expected := []string{path.Join(dirPath, filename)}
 	require.ElementsMatch(t, expected, files)

pkg/confgen/config.go

@@ -19,8 +19,10 @@ package confgen
 
 import (
 	"io/ioutil"
+	"os"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"gopkg.in/yaml.v2"
 )
@@ -68,9 +70,17 @@ type Config struct {
 	Visualization ConfigVisualization `yaml:"visualization"`
 }
 
-func (cg *ConfGen) parseConfigFile(fileName string) (*Config, error) {
+func (cg *ConfGen) ParseConfigFile(fileName string) (*Config, error) {
 	// parse config file yaml
+	// provide a minimal config for when config file is missing (as for Netobserv Openshift Operator)
 	var config Config
+	if _, err := os.Stat(fileName); errors.Is(err, os.ErrNotExist) {
+		if len(Opt.GenerateStages) == 0 {
+			log.Errorf("config file %s does not exist", fileName)
+			return nil, err
+		}
+		return &Config{}, nil
+	}
 	yamlFile, err := ioutil.ReadFile(fileName)
 	if err != nil {
 		log.Debugf("ioutil.ReadFile err: %v ", err)

pkg/confgen/config_test.go

@@ -62,7 +62,7 @@ func Test_parseConfigFile(t *testing.T) {
 	cg := getConfGen()
 	err := os.WriteFile(filename, []byte(testConfig), 0644)
 	require.Equal(t, err, nil)
-	config, err := cg.parseConfigFile(filename)
+	config, err := cg.ParseConfigFile(filename)
 	require.NoError(t, err)
 	require.Equal(t, config, expectedConfig())
 }

pkg/confgen/dedup.go

@@ -25,7 +25,7 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-func (cg *ConfGen) dedupe() {
+func (cg *ConfGen) Dedupe() {
 	cg.transformRules = dedupeNetworkTransformRules(cg.transformRules)
 	cg.aggregateDefinitions = dedupeAggregateDefinitions(cg.aggregateDefinitions)
 }
@@ -54,16 +54,16 @@ func dedupeNetworkTransformRules(rules api.NetworkTransformRules) api.NetworkTra
 // dedupeAggregateDefinitions is inefficient because we can't use a map to look for duplicates.
 // The reason is that aggregate.AggregateDefinition is not hashable due to its AggregateBy field which is a slice.
 func dedupeAggregateDefinitions(aggregateDefinitions aggregate.Definitions) aggregate.Definitions {
-	var dedpueSlice []api.AggregateDefinition
+	var dedupeSlice []api.AggregateDefinition
 	for i, aggregateDefinition := range aggregateDefinitions {
-		if containsAggregateDefinitions(dedpueSlice, aggregateDefinition) {
+		if containsAggregateDefinitions(dedupeSlice, aggregateDefinition) {
 			// duplicate aggregateDefinition
 			log.Debugf("Remove duplicate AggregateDefinitions %v at index %v", aggregateDefinition, i)
 			continue
 		}
-		dedpueSlice = append(dedpueSlice, aggregateDefinition)
+		dedupeSlice = append(dedupeSlice, aggregateDefinition)
 	}
-	return dedpueSlice
+	return dedupeSlice
 }
 
 func containsAggregateDefinitions(slice []api.AggregateDefinition, searchItem api.AggregateDefinition) bool {