Merge pull request #93 from eranra/k8s_labels_configurable

adding k8s labels based on  value and only if it is not empty

Author: eranra

README.md

@@ -304,10 +304,13 @@ All the geo-location fields will be named by appending `output` value
 (e.g., `CountryName`, `CountryLongName`, `RegionName`, `CityName` , `Longitude` and `Latitude`)
 
 The fifth rule `add_kubernetes` generates new fields with kubernetes information by
-matching the `input` value (`srcIP` in the example above) with k8s `nodes`, `pods` and `services` IPs.
+matching the `input` 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`,  `OwnerName`, `OwnerType` )
+(e.g., `Namespace`, `Name`, `Type`, `OwnerName`, `OwnerType` )
+
+In addition, if the `parameters` value is not empty, fields with kubernetes labels 
+will be generated, and named by appending `parameters` value to the label keys.   
 
 > Note: kubernetes connection is done using the first available method: 
 > 1. configuration parameter `KubeConfigPath` (in the example above `/tmp/config`) or

contrib/kubernetes/flowlogs2metrics.conf.yaml

@@ -251,7 +251,7 @@ pipeline:
       - input: srcIP
         output: srcK8S
         type: add_kubernetes
-        parameters: ""
+        parameters: srcK8S_labels
       - input: dstIP
         output: dstLocation
         type: add_location

network_definitions/egress_bandwidth_per_namespace.yaml

@@ -14,6 +14,7 @@ transform:
     - input: srcIP
       output: srcK8S
       type: add_kubernetes
+      parameters: srcK8S_labels
 extract:
   aggregates:
     - name: bandwidth_namespace

pkg/pipeline/transform/transform_network.go

@@ -135,8 +135,10 @@ func (n *Network) Transform(inputEntry config.GenericMap) config.GenericMap {
 			outputEntries[rule.Output+"_Type"] = kubeInfo.Type
 			outputEntries[rule.Output+"_OwnerName"] = kubeInfo.Owner.Name
 			outputEntries[rule.Output+"_OwnerType"] = kubeInfo.Owner.Type
-			for labelKey, labelValue := range kubeInfo.Labels {
-				outputEntries[rule.Output+"_Labels_"+labelKey] = labelValue
+			if rule.Parameters != "" {
+				for labelKey, labelValue := range kubeInfo.Labels {
+					outputEntries[rule.Parameters+"_"+labelKey] = labelValue
+				}
 			}
 		default:
 			log.Panicf("unknown type %s for transform.Network rule: %v", rule.Type, rule)