NETOBSERV-1203: Added option to add zones in k8s transform rule (#575)

* Added option to add zones in k8s transform rule

* Add zone feature to opentelemetry too

pkg/api/transform_network.go

@@ -69,6 +69,7 @@ type NetworkTransformRule struct {
 	Parameters      string        `yaml:"parameters,omitempty" json:"parameters,omitempty" doc:"parameters specific to type"`
 	Assignee        string        `yaml:"assignee,omitempty" json:"assignee,omitempty" doc:"value needs to assign to output field"`
 	KubernetesInfra *K8sInfraRule `yaml:"kubernetes_infra,omitempty" json:"kubernetes_infra,omitempty" doc:"Kubernetes infra rule specific configuration"`
+	Kubernetes      *K8sRule      `yaml:"kubernetes,omitempty" json:"kubernetes,omitempty" doc:"Kubernetes rule specific configuration"`
 }
 
 type K8sInfraRule struct {
@@ -77,6 +78,10 @@ type K8sInfraRule struct {
 	InfraPrefix string   `yaml:"infra_prefixes,omitempty" json:"infra_prefixes,omitempty" doc:"Namespace prefixes that will be tagged as infra"`
 }
 
+type K8sRule struct {
+	AddZone bool `yaml:"add_zone,omitempty" json:"add_zone,omitempty" doc:"If true the rule will add the zone"`
+}
+
 type NetworkTransformDirectionInfo struct {
 	ReporterIPField    string `yaml:"reporterIPField,omitempty" json:"reporterIPField,omitempty" doc:"field providing the reporter (agent) host IP"`
 	SrcHostField       string `yaml:"srcHostField,omitempty" json:"srcHostField,omitempty" doc:"source host field"`

pkg/pipeline/transform/kubernetes/kubernetes.go

@@ -52,6 +52,7 @@ const (
 
 type kubeDataInterface interface {
 	GetInfo(string) (*Info, error)
+	GetNodeInfo(string) (*Info, error)
 	InitFromConfig(string) error
 }
 
@@ -134,6 +135,16 @@ func infoForIP(idx cache.Indexer, ip string) (*Info, bool) {
 	return objs[0].(*Info), true
 }
 
+func (k *KubeData) GetNodeInfo(name string) (*Info, error) {
+	item, ok, err := k.nodes.GetIndexer().GetByKey(name)
+	if err != nil {
+		return nil, err
+	} else if ok {
+		return item.(*Info), nil
+	}
+	return nil, nil
+}
+
 func (k *KubeData) getOwner(info *Info) Owner {
 	if len(info.OwnerReferences) != 0 {
 		ownerReference := info.OwnerReferences[0]
@@ -244,9 +255,10 @@ func (k *KubeData) initPodInformer(informerFactory informers.SharedInformerFacto
 				Labels:          pod.Labels,
 				OwnerReferences: pod.OwnerReferences,
 			},
-			Type:   TypePod,
-			HostIP: pod.Status.HostIP,
-			ips:    ips,
+			Type:     TypePod,
+			HostIP:   pod.Status.HostIP,
+			HostName: pod.Spec.NodeName,
+			ips:      ips,
 		}, nil
 	}); err != nil {
 		return fmt.Errorf("can't set pods transform: %w", err)

pkg/pipeline/transform/transform_network.go

@@ -148,6 +148,7 @@ func fillInK8s(outputEntry config.GenericMap, rule api.NetworkTransformRule) {
 				outputEntry[rule.Output+"_HostName"] = kubeInfo.HostName
 			}
 		}
+		fillInK8sZone(outputEntry, rule, *kubeInfo, "_Zone")
 	} else {
 		// NOTE: Some of these fields are taken from opentelemetry specs.
 		// See https://opentelemetry.io/docs/specs/semconv/resource/k8s/
@@ -181,6 +182,41 @@ func fillInK8s(outputEntry config.GenericMap, rule api.NetworkTransformRule) {
 				outputEntry[rule.Output+"k8s.host.name"] = kubeInfo.HostName
 			}
 		}
+		fillInK8sZone(outputEntry, rule, *kubeInfo, "k8s.zone")
+	}
+}
+
+const nodeZoneLabelName = "topology.kubernetes.io/zone"
+
+func fillInK8sZone(outputEntry config.GenericMap, rule api.NetworkTransformRule, kubeInfo kubernetes.Info, zonePrefix string) {
+	if rule.Kubernetes == nil || !rule.Kubernetes.AddZone {
+		//Nothing to do
+		return
+	}
+	switch kubeInfo.Type {
+	case kubernetes.TypeNode:
+		zone, ok := kubeInfo.Labels[nodeZoneLabelName]
+		if ok {
+			outputEntry[rule.Output+zonePrefix] = zone
+		}
+		return
+	case kubernetes.TypePod:
+		nodeInfo, err := kubernetes.Data.GetNodeInfo(kubeInfo.HostName)
+		if err != nil {
+			logrus.WithError(err).Tracef("can't find nodes info for node %v", kubeInfo.HostName)
+			return
+		}
+		if nodeInfo != nil {
+			zone, ok := nodeInfo.Labels[nodeZoneLabelName]
+			if ok {
+				outputEntry[rule.Output+zonePrefix] = zone
+			}
+		}
+		return
+
+	case kubernetes.TypeService:
+		//A service is not assigned to a dedicated zone, skipping
+		return
 	}
 }
 

pkg/pipeline/transform/transform_network_test.go

@@ -242,6 +242,10 @@ func (*fakeKubeData) GetInfo(n string) (*kubernetes.Info, error) {
 	return nil, errors.New("notFound")
 }
 
+func (*fakeKubeData) GetNodeInfo(n string) (*kubernetes.Info, error) {
+	return nil, nil
+}
+
 func Test_Categorize(t *testing.T) {
 	entry := config.GenericMap{
 		"addr1": "10.1.2.3",