NETOBSERV-1875: Make enrichment indexes configurable

README.md

@@ -422,7 +422,7 @@ parameters:
               output: dstLocation
           - type: add_kubernetes
             kubernetes:
-              input: srcIP
+              ipField: srcIP
               output: srcK8S
 ```
 
@@ -443,7 +443,7 @@ All the geo-location fields will be named by appending `output` value
 (e.g., `CountryName`, `CountryLongName`, `RegionName`, `CityName` , `Longitude` and `Latitude`)
 
 The rule `add_kubernetes` generates new fields with kubernetes information by
-matching the `input` value (`srcIP` in the example above) with kubernetes `nodes`, `pods` and `services` IPs.
+matching the `ipField` value (`srcIP` in the example above) with kubernetes `nodes`, `pods` and `services` IPs.
 All the kubernetes fields will be named by appending `output` value
 (`srcK8S` in the example above) to the kubernetes metadata field names
 (e.g., `Namespace`, `Name`, `Type`, `HostIP`, `OwnerName`, `OwnerType` )

cmd/flowlogs-pipeline/main_test.go

@@ -52,7 +52,7 @@ func TestPipelineConfigSetup(t *testing.T) {
 
 	js := `{
     "PipeLine": "[{\"name\":\"grpc\"},{\"follows\":\"grpc\",\"name\":\"enrich\"},{\"follows\":\"enrich\",\"name\":\"loki\"},{\"follows\":\"enrich\",\"name\":\"prometheus\"}]",
-    "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"kubernetes\":{\"input\":\"SrcAddr\",\"output\":\"SrcK8S\"},\"type\":\"add_kubernetes\"},{\"kubernetes\":{\"input\":\"DstAddr\",\"output\":\"DstK8S\"},\"type\":\"add_kubernetes\"},{\"add_service\":{\"input\":\"DstPort\",\"output\":\"Service\",\"protocol\":\"Proto\"},\"type\":\"add_service\"},{\"add_subnet\":{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"subnet_mask\":\"/16\"},\"type\":\"add_subnet\"}]},\"type\":\"network\"}},{\"name\":\"loki\",\"write\":{\"loki\":{\"batchSize\":102400,\"batchWait\":\"1s\",\"clientConfig\":{\"follow_redirects\":false,\"proxy_url\":null,\"tls_config\":{\"insecure_skip_verify\":false}},\"labels\":[\"SrcK8S_Namespace\",\"SrcK8S_OwnerName\",\"DstK8S_Namespace\",\"DstK8S_OwnerName\",\"FlowDirection\"],\"maxBackoff\":\"5m0s\",\"maxRetries\":10,\"minBackoff\":\"1s\",\"staticLabels\":{\"app\":\"netobserv-flowcollector\"},\"tenantID\":\"netobserv\",\"timeout\":\"10s\",\"timestampLabel\":\"TimeFlowEndMs\",\"timestampScale\":\"1ms\",\"url\":\"http://loki.netobserv.svc:3100/\"},\"type\":\"loki\"}},{\"encode\":{\"prom\":{\"metrics\":[{\"buckets\":null,\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
+    "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"kubernetes\":{\"ipField\":\"SrcAddr\",\"output\":\"SrcK8S\"},\"type\":\"add_kubernetes\"},{\"kubernetes\":{\"ipField\":\"DstAddr\",\"output\":\"DstK8S\"},\"type\":\"add_kubernetes\"},{\"add_service\":{\"input\":\"DstPort\",\"output\":\"Service\",\"protocol\":\"Proto\"},\"type\":\"add_service\"},{\"add_subnet\":{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"subnet_mask\":\"/16\"},\"type\":\"add_subnet\"}]},\"type\":\"network\"}},{\"name\":\"loki\",\"write\":{\"loki\":{\"batchSize\":102400,\"batchWait\":\"1s\",\"clientConfig\":{\"follow_redirects\":false,\"proxy_url\":null,\"tls_config\":{\"insecure_skip_verify\":false}},\"labels\":[\"SrcK8S_Namespace\",\"SrcK8S_OwnerName\",\"DstK8S_Namespace\",\"DstK8S_OwnerName\",\"FlowDirection\"],\"maxBackoff\":\"5m0s\",\"maxRetries\":10,\"minBackoff\":\"1s\",\"staticLabels\":{\"app\":\"netobserv-flowcollector\"},\"tenantID\":\"netobserv\",\"timeout\":\"10s\",\"timestampLabel\":\"TimeFlowEndMs\",\"timestampScale\":\"1ms\",\"url\":\"http://loki.netobserv.svc:3100/\"},\"type\":\"loki\"}},{\"encode\":{\"prom\":{\"metrics\":[{\"buckets\":null,\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
     "Health": {
         "Port": "8080"
     },

contrib/kubernetes/flowlogs-pipeline.conf.yaml

@@ -144,7 +144,7 @@ parameters:
           subnet_mask: /16
       - type: add_kubernetes
         kubernetes:
-          input: srcIP
+          ipField: srcIP
           output: srcK8S
           labels_prefix: srcK8S_labels
       - type: add_location

docs/api.md

@@ -250,8 +250,9 @@ Following is the supported API format for network transformations:
                              name: name of the object
                              namespace: namespace of the object
                  kubernetes: Kubernetes rule configuration
-                     input: entry IP input field
-                     mac-input: Optional entry MAC input field
+                     ipField: entry IP input field
+                     interfacesField: entry Interfaces input field
+                     macField: entry MAC input field
                      output: entry output field
                      assignee: value needs to assign to output field
                      labels_prefix: labels prefix to use to copy input lables, if empty labels will not be copied
@@ -272,7 +273,10 @@ Following is the supported API format for network transformations:
                      protocol: entry protocol field
          kubeConfig: global configuration related to Kubernetes (optional)
              configPath: path to kubeconfig file (optional)
-             managedCNI: a list of CNI (network plugins) to manage, for detecting additional interfaces. Currently supported: ovn, multus
+             secondaryNetworks: configuration for secondary networks
+                     name: name of the secondary network, as mentioned in the annotation 'k8s.v1.cni.cncf.io/network-status'
+                     index: fields to use for indexing, must be any combination of 'mac', 'ip', 'interface'
+             managedCNI: a list of CNI (network plugins) to manage, for detecting additional interfaces. Currently supported: ovn
          servicesFile: path to services file (optional, default: /etc/services)
          protocolsFile: path to protocols file (optional, default: /etc/protocols)
          subnetLabels: configure subnet and IPs custom labels

docs/operational-metrics.md

@@ -143,6 +143,14 @@ Each table below provides documentation for an exported flowlogs-pipeline operat
 | **Labels** | stage | 
 
 
+### secondary_network_indexer_hit
+| **Name** | secondary_network_indexer_hit | 
+|:---|:---|
+| **Description** | Counter of hits per secondary network index for Kubernetes enrichment | 
+| **Type** | counter | 
+| **Labels** | kind, network, warning | 
+
+
 ### stage_duration_ms
 | **Name** | stage_duration_ms | 
 |:---|:---|

pkg/api/transform_network.go

@@ -39,13 +39,13 @@ func (tn *TransformNetwork) GetServiceFiles() (string, string) {
 }
 
 const (
-	OVN    = "ovn"
-	Multus = "multus"
+	OVN = "ovn"
 )
 
 type NetworkTransformKubeConfig struct {
-	ConfigPath string   `yaml:"configPath,omitempty" json:"configPath,omitempty" doc:"path to kubeconfig file (optional)"`
-	ManagedCNI []string `yaml:"managedCNI,omitempty" json:"managedCNI,omitempty" doc:"a list of CNI (network plugins) to manage, for detecting additional interfaces. Currently supported: ovn, multus"`
+	ConfigPath        string             `yaml:"configPath,omitempty" json:"configPath,omitempty" doc:"path to kubeconfig file (optional)"`
+	SecondaryNetworks []SecondaryNetwork `yaml:"secondaryNetworks,omitempty" json:"secondaryNetworks,omitempty" doc:"configuration for secondary networks"`
+	ManagedCNI        []string           `yaml:"managedCNI,omitempty" json:"managedCNI,omitempty" doc:"a list of CNI (network plugins) to manage, for detecting additional interfaces. Currently supported: ovn"`
 }
 
 type TransformNetworkOperationEnum string
@@ -84,12 +84,18 @@ type K8sReference struct {
 }
 
 type K8sRule struct {
-	Input        string `yaml:"input,omitempty" json:"input,omitempty" doc:"entry IP input field"`
-	MacInput     string `yaml:"mac-input,omitempty" json:"mac-input,omitempty" doc:"Optional entry MAC input field"`
-	Output       string `yaml:"output,omitempty" json:"output,omitempty" doc:"entry output field"`
-	Assignee     string `yaml:"assignee,omitempty" json:"assignee,omitempty" doc:"value needs to assign to output field"`
-	LabelsPrefix string `yaml:"labels_prefix,omitempty" json:"labels_prefix,omitempty" doc:"labels prefix to use to copy input lables, if empty labels will not be copied"`
-	AddZone      bool   `yaml:"add_zone,omitempty" json:"add_zone,omitempty" doc:"if true the rule will add the zone"`
+	IPField         string `yaml:"ipField,omitempty" json:"ipField,omitempty" doc:"entry IP input field"`
+	InterfacesField string `yaml:"interfacesField,omitempty" json:"interfacesField,omitempty" doc:"entry Interfaces input field"`
+	MACField        string `yaml:"macField,omitempty" json:"macField,omitempty" doc:"entry MAC input field"`
+	Output          string `yaml:"output,omitempty" json:"output,omitempty" doc:"entry output field"`
+	Assignee        string `yaml:"assignee,omitempty" json:"assignee,omitempty" doc:"value needs to assign to output field"`
+	LabelsPrefix    string `yaml:"labels_prefix,omitempty" json:"labels_prefix,omitempty" doc:"labels prefix to use to copy input lables, if empty labels will not be copied"`
+	AddZone         bool   `yaml:"add_zone,omitempty" json:"add_zone,omitempty" doc:"if true the rule will add the zone"`
+}
+
+type SecondaryNetwork struct {
+	Name  string         `yaml:"name,omitempty" json:"name,omitempty" doc:"name of the secondary network, as mentioned in the annotation 'k8s.v1.cni.cncf.io/network-status'"`
+	Index map[string]any `yaml:"index,omitempty" json:"index,omitempty" doc:"fields to use for indexing, must be any combination of 'mac', 'ip', 'interface'"`
 }
 
 type NetworkGenericRule struct {

pkg/confgen/dedup_test.go

@@ -26,15 +26,15 @@ import (
 
 func Test_dedupeNetworkTransformRules(t *testing.T) {
 	slice := api.NetworkTransformRules{
-		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{Input: "i1", Output: "o1"}},
-		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{Input: "i2", Output: "o2"}},
-		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{Input: "i3", Output: "o3"}},
-		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{Input: "i2", Output: "o2"}},
+		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{IPField: "i1", Output: "o1"}},
+		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{IPField: "i2", Output: "o2"}},
+		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{IPField: "i3", Output: "o3"}},
+		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{IPField: "i2", Output: "o2"}},
 	}
 	expected := api.NetworkTransformRules{
-		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{Input: "i1", Output: "o1"}},
-		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{Input: "i2", Output: "o2"}},
-		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{Input: "i3", Output: "o3"}},
+		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{IPField: "i1", Output: "o1"}},
+		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{IPField: "i2", Output: "o2"}},
+		api.NetworkTransformRule{Type: "add_kubernetes", Kubernetes: &api.K8sRule{IPField: "i3", Output: "o3"}},
 	}
 	actual := dedupeNetworkTransformRules(slice)
 

pkg/config/pipeline_builder_test.go

@@ -31,14 +31,14 @@ func TestLokiPipeline(t *testing.T) {
 	pl = pl.TransformNetwork("enrich", api.TransformNetwork{Rules: api.NetworkTransformRules{{
 		Type: api.NetworkAddKubernetes,
 		Kubernetes: &api.K8sRule{
-			Input:  "SrcAddr",
-			Output: "SrcK8S",
+			IPField: "SrcAddr",
+			Output:  "SrcK8S",
 		},
 	}, {
 		Type: api.NetworkAddKubernetes,
 		Kubernetes: &api.K8sRule{
-			Input:  "DstAddr",
-			Output: "DstK8S",
+			IPField: "DstAddr",
+			Output:  "DstK8S",
 		},
 	}}})
 	pl = pl.WriteLoki("loki", api.WriteLoki{URL: "http://loki:3100/"})
@@ -58,7 +58,7 @@ func TestLokiPipeline(t *testing.T) {
 
 	b, err = json.Marshal(params[1])
 	require.NoError(t, err)
-	require.JSONEq(t, `{"name":"enrich","transform":{"type":"network","network":{"directionInfo":{},"kubeConfig":{},"rules":[{"kubernetes":{"input":"SrcAddr","output":"SrcK8S"},"type":"add_kubernetes"},{"kubernetes":{"input":"DstAddr","output":"DstK8S"},"type":"add_kubernetes"}]}}}`, string(b))
+	require.JSONEq(t, `{"name":"enrich","transform":{"type":"network","network":{"directionInfo":{},"kubeConfig":{},"rules":[{"kubernetes":{"ipField":"SrcAddr","output":"SrcK8S"},"type":"add_kubernetes"},{"kubernetes":{"ipField":"DstAddr","output":"DstK8S"},"type":"add_kubernetes"}]}}}`, string(b))
 
 	b, err = json.Marshal(params[2])
 	require.NoError(t, err)
@@ -206,14 +206,14 @@ func TestIPFIXPipeline(t *testing.T) {
 	pl = pl.TransformNetwork("enrich", api.TransformNetwork{Rules: api.NetworkTransformRules{{
 		Type: api.NetworkAddKubernetes,
 		Kubernetes: &api.K8sRule{
-			Input:  "SrcAddr",
-			Output: "SrcK8S",
+			IPField: "SrcAddr",
+			Output:  "SrcK8S",
 		},
 	}, {
 		Type: api.NetworkAddKubernetes,
 		Kubernetes: &api.K8sRule{
-			Input:  "DstAddr",
-			Output: "DstK8S",
+			IPField: "DstAddr",
+			Output:  "DstK8S",
 		},
 	}}})
 	pl = pl.WriteIpfix("ipfix", api.WriteIpfix{
@@ -238,7 +238,7 @@ func TestIPFIXPipeline(t *testing.T) {
 
 	b, err = json.Marshal(params[1])
 	require.NoError(t, err)
-	require.JSONEq(t, `{"name":"enrich","transform":{"type":"network","network":{"directionInfo":{},"kubeConfig":{},"rules":[{"kubernetes":{"input":"SrcAddr","output":"SrcK8S"},"type":"add_kubernetes"},{"kubernetes":{"input":"DstAddr","output":"DstK8S"},"type":"add_kubernetes"}]}}}`, string(b))
+	require.JSONEq(t, `{"name":"enrich","transform":{"type":"network","network":{"directionInfo":{},"kubeConfig":{},"rules":[{"kubernetes":{"ipField":"SrcAddr","output":"SrcK8S"},"type":"add_kubernetes"},{"kubernetes":{"ipField":"DstAddr","output":"DstK8S"},"type":"add_kubernetes"}]}}}`, string(b))
 
 	b, err = json.Marshal(params[2])
 	require.NoError(t, err)

pkg/operational/metrics.go

@@ -80,6 +80,15 @@ var (
 		TypeHistogram,
 		"stage",
 	)
+	indexerHit = DefineMetric(
+		"secondary_network_indexer_hit",
+		"Counter of hits per secondary network index for Kubernetes enrichment",
+		TypeCounter,
+		"kind",
+		"namespace",
+		"network",
+		"warning",
+	)
 )
 
 func (def *MetricDefinition) mapLabels(labels []string) prometheus.Labels {
@@ -241,6 +250,10 @@ func (o *Metrics) GetOrCreateStageDurationHisto() *prometheus.HistogramVec {
 	return o.stageDurationHisto
 }
 
+func (o *Metrics) CreateIndexerHitCounter() *prometheus.CounterVec {
+	return o.NewCounterVec(&indexerHit)
+}
+
 func GetDocumentation() string {
 	doc := ""
 	sort.Slice(allMetrics, func(i, j int) bool {

pkg/pipeline/pipeline_builder.go

@@ -434,7 +434,7 @@ func getWriter(opMetrics *operational.Metrics, params config.StageParam) (write.
 	return writer, err
 }
 
-func getTransformer(_ *operational.Metrics, params config.StageParam) (transform.Transformer, error) {
+func getTransformer(opMetrics *operational.Metrics, params config.StageParam) (transform.Transformer, error) {
 	var transformer transform.Transformer
 	var err error
 	switch params.Transform.Type {
@@ -443,7 +443,7 @@ func getTransformer(_ *operational.Metrics, params config.StageParam) (transform
 	case api.FilterType:
 		transformer, err = transform.NewTransformFilter(params)
 	case api.NetworkType:
-		transformer, err = transform.NewTransformNetwork(params)
+		transformer, err = transform.NewTransformNetwork(params, opMetrics)
 	case api.NoneType:
 		transformer, err = transform.NewTransformNone()
 	default:

pkg/pipeline/transform/kubernetes/cni/cni.go

@@ -6,5 +6,4 @@ import (
 
 type Plugin interface {
 	GetNodeIPs(node *v1.Node) []string
-	GetPodIPsAndMACs(pod *v1.Pod) ([]string, []string)
 }