adjustments to confgenerator for Operator

cmd/confgenerator/main.go

@@ -63,7 +63,7 @@ func initConfig() {
 		if err != nil {
 			log.Fatal(err)
 		}
-		// Search config in home directory with name ".flpconfgen" (without extension).
+		// Search config in home directory with name ".confgen" (without extension).
 		v.AddConfigPath(home)
 		v.SetConfigName(defaultLogFileName)
 	}
@@ -132,7 +132,8 @@ func initFlags() {
 	rootCmd.PersistentFlags().StringVar(&confgen.Opt.DestConfFile, "destConfFile", "/tmp/flowlogs-pipeline.conf.yaml", "destination configuration file")
 	rootCmd.PersistentFlags().StringVar(&confgen.Opt.DestDocFile, "destDocFile", "/tmp/metrics.md", "destination documentation file (.md)")
 	rootCmd.PersistentFlags().StringVar(&confgen.Opt.DestGrafanaJsonnetFolder, "destGrafanaJsonnetFolder", "/tmp/jsonnet", "destination grafana jsonnet folder")
-	rootCmd.PersistentFlags().StringSliceVar(&confgen.Opt.SkipWithLabels, "skipWithLabels", nil, "Skip definitions with Labels")
+	rootCmd.PersistentFlags().StringSliceVar(&confgen.Opt.SkipWithTags, "skipWithTags", nil, "Skip definitions with Tags")
+	rootCmd.PersistentFlags().StringSliceVar(&confgen.Opt.GenerateStages, "generateStages", nil, "Produce only specified stages (ingest, transform_generic, transform_network, extract_aggregate, encode_prom, write_loki")
 }
 
 func main() {

docs/confGenerator.md

@@ -21,13 +21,14 @@ Usage:
 
 Flags:
       --config string                     config file (default is $HOME/.confgen)
-      --destConfFile string               destination configuration file (default "flowlogs-pipeline.conf.yaml")
-      --destGrafanaJsonnetFolder string   destination grafana jsonnet folder
+      --destConfFile string               destination configuration file (default "/tmp/flowlogs-pipeline.conf.yaml")
+      --destDocFile string                destination documentation file (.md) (default "/tmp/metrics.md")
+      --destGrafanaJsonnetFolder string   destination grafana jsonnet folder (default "/tmp/jsonnet")
+      --generateStages strings            Produce only specified stages (ingest, transform_generic, transform_network, extract_aggregate, encode_prom, write_loki
   -h, --help                              help for confgenerator
       --log-level string                  Log level: debug, info, warning, error (default "error")
-      --skipWithLabels strings            Skip definitions with Labels
+      --skipWithTags strings              Skip definitions with Tags
       --srcFolder string                  source folder (default "network_definitions")
-
 ```
 
 > Note: confgenerator is available also from `netobserv/flowlogs-pipeline` quay image. To use execute:  

network_definitions/bandwidth_per_network_service.yaml

@@ -5,7 +5,7 @@ details:
   Sum bytes for all traffic per network service
 usage:
   Evaluate network usage breakdown per network service
-labels:
+tags:
   - bandwidth
   - graph
   - rate

network_definitions/bandwidth_per_src_dest_subnet.yaml

@@ -5,7 +5,7 @@ details:
   Sum bandwidth bytes for all traffic per source / destination subnet pair
 usage:
   Evaluate network usage breakdown per source / destination subnet pair
-labels:
+tags:
   - bandwidth
   - graph
   - rate

network_definitions/bandwidth_per_src_subnet.yaml

@@ -5,7 +5,7 @@ details:
   Sum bytes for all traffic per source subnet
 usage:
   Evaluate network usage breakdown per source subnet
-labels:
+tags:
   - bandwidth
   - graph
   - rate

network_definitions/connection_rate_per_dest_subnet.yaml

@@ -5,7 +5,7 @@ details:
   Counts the number of connections per subnet with network prefix length /16 (using conn_tracking sum isNewFlow field)
 usage:
   Evaluate network connections per subnet
-labels:
+tags:
   - rate
   - subnet
 transform:

network_definitions/connection_rate_per_src_subnet.yaml

@@ -5,7 +5,7 @@ details:
   Counts the number of connections per subnet with network prefix length /16
 usage:
   Evaluate network connections per subnet
-labels:
+tags:
   - rate
   - subnet
 transform:

network_definitions/connection_rate_per_tcp_flags.yaml

@@ -5,7 +5,7 @@ details:
   Counts the number of connections per tcp flags
 usage:
   Evaluate difference in connections rate of different TCP Flags. Can be used, for example, to identify syn-attacks.
-labels:
+tags:
   - rate
   - TCPFlags
 extract:

network_definitions/connections_per_dst_as.yaml

@@ -5,7 +5,7 @@ details:
   Aggregates flow records by values of "DstAS" field and counts the number of entries in each aggregate with non zero value
 usage:
   Evaluate amount of connections targeted at different Autonomous Systems
-labels:
+tags:
   - rate
   - count
   - AS

network_definitions/connections_per_src_as.yaml

@@ -5,7 +5,7 @@ details:
   Aggregates flow records by values of "SrcAS" field and counts the number of entries in each aggregate with non zero value
 usage:
   Evaluate amount of connections initiated by different Autonomous Systems
-labels:
+tags:
   - rate
   - count
   - AS

network_definitions/count_per_src_dest_subnet.yaml

@@ -5,7 +5,7 @@ details:
   Count the number of distinct source / destination subnet pairs
 usage:
   Evaluate network usage breakdown per source / destination subnet pair
-labels:
+tags:
   - count
   - graph
   - rate

network_definitions/egress_bandwidth_per_dest_subnet.yaml

@@ -5,7 +5,7 @@ details:
   Sum egress bytes for all traffic per destination subnet
 usage:
   Evaluate network usage breakdown per destination subnet
-labels:
+tags:
   - bandwidth
   - graph
   - rate

network_definitions/egress_bandwidth_per_namespace.yaml

@@ -5,7 +5,7 @@ details:
   Sum egress bytes for all traffic per namespace
 usage:
   Evaluate network usage breakdown per namespace
-labels:
+tags:
   - kubernetes
   - bandwidth
   - graph

network_definitions/flows_length_histogram.yaml

@@ -5,7 +5,7 @@ details:
   Flows length distribution over time
 usage:
   Evaluate flows length behavior including mice/elephant use-case 
-labels:
+tags:
   - bandwidth
   - mice
   - elephant

network_definitions/geo-location_rate_per_dest.yaml

@@ -5,7 +5,7 @@ details:
   Counts the number of connections per geo-location based on destination IP
 usage:
   Evaluate network connections geo-location
-labels:
+tags:
   - rate
   - connections-count
   - geo-location

network_definitions/loki_bandwidth_per_namespace.yaml

@@ -5,7 +5,7 @@ details:
   Sum bytes for all traffic per source namespace
 usage:
   Evaluate network usage breakdown per source namespace
-labels:
+tags:
   - loki
   - graph
   - rate

network_definitions/loki_logs_per_sec.yaml

@@ -5,7 +5,7 @@ details:
   Rate of loki logs per sec
 usage:
   Evaluate loki service usage
-labels:
+tags:
   - loki
   - graph
   - rate

network_definitions/network_services_count.yaml

@@ -5,7 +5,7 @@ details:
   Counts the number of connections per network service based on destination port number and protocol
 usage:
   Evaluate network services
-labels:
+tags:
   - rate
   - network-services
   - destination-port

pkg/confgen/confgen.go

@@ -40,7 +40,7 @@ type Definition struct {
 	Description          string
 	Details              string
 	Usage                string
-	Labels               []string
+	Tags                 []string
 	TransformNetwork     *api.TransformNetwork
 	AggregateDefinitions *aggregate.Definitions
 	PromEncode           *api.PromEncode
@@ -62,7 +62,7 @@ type DefFile struct {
 	Description   string                 `yaml:"description"`
 	Details       string                 `yaml:"details"`
 	Usage         string                 `yaml:"usage"`
-	Labels        []string               `yaml:"labels"`
+	Tags          []string               `yaml:"tags"`
 	Transform     map[string]interface{} `yaml:"transform"`
 	Extract       map[string]interface{} `yaml:"extract"`
 	Encode        map[string]interface{} `yaml:"encode"`
@@ -88,10 +88,21 @@ func (cg *ConfGen) Run() error {
 
 	cg.dedupe()
 
-	err = cg.generateFlowlogs2PipelineConfig(Opt.DestConfFile)
-	if err != nil {
-		log.Debugf("cg.generateFlowlogs2PipelineConfig err: %v ", err)
-		return err
+	if len(Opt.GenerateStages) != 0 {
+		config := cg.GenerateTruncatedConfig(Opt.GenerateStages)
+		err = cg.writeConfigFile(Opt.DestConfFile, config)
+		if err != nil {
+			log.Debugf("cg.GenerateTruncatedConfig err: %v ", err)
+			return err
+		}
+		return nil
+	} else {
+		config := cg.GenerateFlowlogs2PipelineConfig()
+		err = cg.writeConfigFile(Opt.DestConfFile, config)
+		if err != nil {
+			log.Debugf("cg.GenerateFlowlogs2PipelineConfig err: %v ", err)
+			return err
+		}
 	}
 
 	err = cg.generateDoc(Opt.DestDocFile)
@@ -153,11 +164,11 @@ func (cg *ConfGen) parseFile(fileName string) error {
 		return err
 	}
 
-	//skip if there skip label match
-	for _, skipLabel := range Opt.SkipWithLabels {
-		for _, label := range defFile.Labels {
-			if skipLabel == label {
-				return fmt.Errorf("skipping definition %s due to skip label %s", fileName, label)
+	//skip if their skip tag match
+	for _, skipTag := range Opt.SkipWithTags {
+		for _, tag := range defFile.Tags {
+			if skipTag == tag {
+				return fmt.Errorf("skipping definition %s due to skip tag %s", fileName, tag)
 			}
 		}
 	}
@@ -168,7 +179,7 @@ func (cg *ConfGen) parseFile(fileName string) error {
 		Description: defFile.Description,
 		Details:     defFile.Details,
 		Usage:       defFile.Usage,
-		Labels:      defFile.Labels,
+		Tags:        defFile.Tags,
 	}
 
 	// parse transport

pkg/confgen/config.go

@@ -30,7 +30,8 @@ type Options struct {
 	DestDocFile              string
 	DestGrafanaJsonnetFolder string
 	SrcFolder                string
-	SkipWithLabels           []string
+	SkipWithTags             []string
+	GenerateStages           []string
 }
 
 var (

pkg/confgen/doc.go

@@ -70,7 +70,7 @@ func (cg *ConfGen) generateDoc(fileName string) error {
 		replacer := strings.NewReplacer("-", " ", "_", " ")
 		name := replacer.Replace(filepath.Base(metric.FileName[:len(metric.FileName)-len(filepath.Ext(metric.FileName))]))
 
-		labels := strings.Join(metric.Labels[:], ", ")
+		labels := strings.Join(metric.Tags[:], ", ")
 		// TODO: add support for multiple operations
 		operation := cg.generateOperationText(*metric.AggregateDefinitions)
 		expose := cg.generatePromEncodeText(metric.PromEncode.Metrics)
@@ -82,7 +82,7 @@ func (cg *ConfGen) generateDoc(fileName string) error {
 |:---|:---|
 | **Details** | %s | 
 | **Usage** | %s | 
-| **Labels** | %s |
+| **Tags** | %s |
 %s%s%s|||  
 
 `,

pkg/confgen/flowlogs2metrics_config.go

@@ -21,10 +21,11 @@ import (
 	"fmt"
 	"io/ioutil"
 
+	log "github.com/sirupsen/logrus"
 	"gopkg.in/yaml.v2"
 )
 
-func (cg *ConfGen) generateFlowlogs2PipelineConfig(fileName string) error {
+func (cg *ConfGen) GenerateFlowlogs2PipelineConfig() map[string]interface{} {
 	config := map[string]interface{}{
 		"log-level": "error",
 		"pipeline": []map[string]string{
@@ -97,7 +98,84 @@ func (cg *ConfGen) generateFlowlogs2PipelineConfig(fileName string) error {
 			},
 		},
 	}
+	return config
+}
+
+func (cg *ConfGen) GenerateTruncatedConfig(stages []string) map[string]interface{} {
+	parameters := make([]map[string]interface{}, len(stages))
+	for i, stage := range stages {
+		switch stage {
+		case "ingest":
+			parameters[i] = map[string]interface{}{
+				"name": "ingest_collector",
+				"ingest": map[string]interface{}{
+					"type": "collector",
+					"collector": map[string]interface{}{
+						"port":       cg.config.Ingest.Collector.Port,
+						"portLegacy": cg.config.Ingest.Collector.PortLegacy,
+						"hostname":   cg.config.Ingest.Collector.HostName,
+					},
+				},
+			}
+		case "transform_generic":
+			parameters[i] = map[string]interface{}{
+				"name": "transform_generic",
+				"transform": map[string]interface{}{
+					"type": "generic",
+					"generic": map[string]interface{}{
+						"policy": "replace_keys",
+						"rules":  cg.config.Transform.Generic.Rules,
+					},
+				},
+			}
+		case "transform_network":
+			parameters[i] = map[string]interface{}{
+				"name": "transform_network",
+				"transform": map[string]interface{}{
+					"type": "network",
+					"network": map[string]interface{}{
+						"rules": cg.transformRules,
+					},
+				},
+			}
+		case "extract_aggregate":
+			parameters[i] = map[string]interface{}{
+				"name": "extract_aggregate",
+				"extract": map[string]interface{}{
+					"type":       "aggregates",
+					"aggregates": cg.aggregateDefinitions,
+				},
+			}
+		case "encode_prom":
+			parameters[i] = map[string]interface{}{
+				"name": "encode_prom",
+				"encode": map[string]interface{}{
+					"type": "prom",
+					"prom": map[string]interface{}{
+						"port":    cg.config.Encode.Prom.Port,
+						"prefix":  cg.config.Encode.Prom.Prefix,
+						"metrics": cg.promMetrics,
+					},
+				},
+			}
+		case "write_loki":
+			parameters[i] = map[string]interface{}{
+				"name": "write_loki",
+				"write": map[string]interface{}{
+					"type": cg.config.Write.Type,
+					"loki": cg.config.Write.Loki,
+				},
+			}
+		}
+	}
+	log.Debugf("parameters = %v \n", parameters)
+	config := map[string]interface{}{
+		"parameters": parameters,
+	}
+	return config
+}
 
+func (cg *ConfGen) writeConfigFile(fileName string, config map[string]interface{}) error {
 	configData, err := yaml.Marshal(&config)
 	if err != nil {
 		return err