Merge pull request #91 from eranra/add_k8s_owner

add k8s owner

Author: eranra

README.md

@@ -307,7 +307,8 @@ The fifth rule `add_kubernetes` generates new fields with kubernetes information
 matching the `input` value (`srcIP` in the example above) with k8s `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., `Type`, `Name` and `Namespace`)
+(e.g., `Namespace`, `Name`, `Type`,  `OwnerName`, `OwnerType` )
+
 > Note: kubernetes connection is done using the first available method: 
 > 1. configuration parameter `KubeConfigPath` (in the example above `/tmp/config`) or
 > 2. using `KUBECONFIG` environment variable

go.mod

@@ -23,6 +23,7 @@ require (
 	gopkg.in/yaml.v2 v2.4.0
 	honnef.co/go/netdb v0.0.0-20210921115105-e902e863d85d
 	k8s.io/api v0.23.2
+	k8s.io/apimachinery v0.23.2
 	k8s.io/client-go v0.23.2
 )
 
@@ -75,7 +76,6 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.66.2 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
-	k8s.io/apimachinery v0.23.2 // indirect
 	k8s.io/klog/v2 v2.30.0 // indirect
 	k8s.io/kube-openapi v0.0.0-20211115234752-e816edb12b65 // indirect
 	k8s.io/utils v0.0.0-20210930125809-cb0fa318a74b // indirect

pkg/pipeline/transform/kubernetes/kubernetes.go

@@ -20,7 +20,9 @@ package kubernetes
 import (
 	"fmt"
 	log "github.com/sirupsen/logrus"
+	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -37,60 +39,104 @@ const (
 	kubeConfigEnvVariable = "KUBECONFIG"
 	syncTime              = 10 * time.Minute
 	IndexIP               = "byIP"
-	typeNode              = "node"
-	typePod               = "pod"
-	typeService           = "service"
+	typeNode              = "Node"
+	typePod               = "Pod"
+	typeService           = "Service"
 )
 
 type KubeData struct {
-	informers map[string]cache.SharedIndexInformer
-	stopChan  chan struct{}
+	ipInformers        map[string]cache.SharedIndexInformer
+	replicaSetInformer cache.SharedIndexInformer
+	stopChan           chan struct{}
+}
+
+type Owner struct {
+	Type string
+	Name string
 }
 
 type Info struct {
-	Type      string
-	Name      string
-	Namespace string
-	Labels    map[string]string
+	Type            string
+	Name            string
+	Namespace       string
+	Labels          map[string]string
+	OwnerReferences []metav1.OwnerReference
+	Owner           Owner
 }
 
-func (k KubeData) GetInfo(ip string) (*Info, error) {
-	for objType, informer := range k.informers {
+func (k *KubeData) GetInfo(ip string) (*Info, error) {
+	for objType, informer := range k.ipInformers {
 		objs, err := informer.GetIndexer().ByIndex(IndexIP, ip)
 		if err == nil && len(objs) > 0 {
+			var info *Info
 			switch objType {
 			case typePod:
 				pod := objs[0].(*v1.Pod)
-				return &Info{
-					Type:      typePod,
-					Name:      pod.Name,
-					Namespace: pod.Namespace,
-					Labels:    pod.Labels,
-				}, nil
+				info = &Info{
+					Type:            typePod,
+					Name:            pod.Name,
+					Namespace:       pod.Namespace,
+					Labels:          pod.Labels,
+					OwnerReferences: pod.OwnerReferences,
+				}
 			case typeNode:
 				node := objs[0].(*v1.Node)
-				return &Info{
+				info = &Info{
 					Type:      typeNode,
 					Name:      node.Name,
 					Namespace: node.Namespace,
 					Labels:    node.Labels,
-				}, nil
+				}
 			case typeService:
 				service := objs[0].(*v1.Service)
-				return &Info{
+				info = &Info{
 					Type:      typeService,
 					Name:      service.Name,
 					Namespace: service.Namespace,
 					Labels:    service.Labels,
-				}, nil
+				}
 			}
+
+			info.Owner = k.getOwner(info)
+			return info, nil
 		}
 	}
 
 	return nil, fmt.Errorf("can't find ip")
 }
 
-func (k KubeData) NewNodeInformer(informerFactory informers.SharedInformerFactory) error {
+func (k *KubeData) getOwner(info *Info) Owner {
+	if info.OwnerReferences != nil && len(info.OwnerReferences) > 0 {
+		ownerReference := info.OwnerReferences[0]
+		if ownerReference.Kind == "ReplicaSet" {
+			item, ok, err := k.replicaSetInformer.GetIndexer().GetByKey(info.Namespace + "/" + ownerReference.Name)
+			if err != nil {
+				panic(err)
+			}
+			if ok {
+				replicaSet := item.(*appsv1.ReplicaSet)
+				if len(replicaSet.OwnerReferences) > 0 {
+					return Owner{
+						Name: replicaSet.OwnerReferences[0].Name,
+						Type: replicaSet.OwnerReferences[0].Kind,
+					}
+				}
+			}
+		} else {
+			return Owner{
+				Name: ownerReference.Name,
+				Type: ownerReference.Kind,
+			}
+		}
+	}
+
+	return Owner{
+		Name: info.Name,
+		Type: info.Type,
+	}
+}
+
+func (k *KubeData) NewNodeInformer(informerFactory informers.SharedInformerFactory) error {
 	nodes := informerFactory.Core().V1().Nodes().Informer()
 	err := nodes.AddIndexers(map[string]cache.IndexFunc{
 		IndexIP: func(obj interface{}) ([]string, error) {
@@ -106,11 +152,11 @@ func (k KubeData) NewNodeInformer(informerFactory informers.SharedInformerFactor
 		},
 	})
 
-	k.informers[typeNode] = nodes
+	k.ipInformers[typeNode] = nodes
 	return err
 }
 
-func (k KubeData) NewPodInformer(informerFactory informers.SharedInformerFactory) error {
+func (k *KubeData) NewPodInformer(informerFactory informers.SharedInformerFactory) error {
 	pods := informerFactory.Core().V1().Pods().Informer()
 	err := pods.AddIndexers(map[string]cache.IndexFunc{
 		IndexIP: func(obj interface{}) ([]string, error) {
@@ -126,11 +172,11 @@ func (k KubeData) NewPodInformer(informerFactory informers.SharedInformerFactory
 		},
 	})
 
-	k.informers[typePod] = pods
+	k.ipInformers[typePod] = pods
 	return err
 }
 
-func (k KubeData) NewServiceInformer(informerFactory informers.SharedInformerFactory) error {
+func (k *KubeData) NewServiceInformer(informerFactory informers.SharedInformerFactory) error {
 	services := informerFactory.Core().V1().Services().Informer()
 	err := services.AddIndexers(map[string]cache.IndexFunc{
 		IndexIP: func(obj interface{}) ([]string, error) {
@@ -143,14 +189,23 @@ func (k KubeData) NewServiceInformer(informerFactory informers.SharedInformerFac
 		},
 	})
 
-	k.informers[typeService] = services
+	k.ipInformers[typeService] = services
 	return err
 }
 
-func (k KubeData) InitFromConfig(kubeConfigPath string) error {
+func (k *KubeData) NewReplicaSetInformer(informerFactory informers.SharedInformerFactory) error {
+	k.replicaSetInformer = informerFactory.Apps().V1().ReplicaSets().Informer()
+	return nil
+}
+
+func (k *KubeData) InitFromConfig(kubeConfigPath string) error {
 	var config *rest.Config
 	var err error
 
+	// Initialization variables
+	k.stopChan = make(chan struct{})
+	k.ipInformers = map[string]cache.SharedIndexInformer{}
+
 	if kubeConfigPath != "" {
 		config, err = clientcmd.BuildConfigFromFlags("", kubeConfigPath)
 		if err != nil {
@@ -189,9 +244,7 @@ func (k KubeData) InitFromConfig(kubeConfigPath string) error {
 	return nil
 }
 
-func (k KubeData) initInformers(client kubernetes.Interface) error {
-	//defer close(stopChan)
-
+func (k *KubeData) initInformers(client kubernetes.Interface) error {
 	informerFactory := informers.NewSharedInformerFactory(client, syncTime)
 	err := k.NewNodeInformer(informerFactory)
 	if err != nil {
@@ -205,18 +258,15 @@ func (k KubeData) initInformers(client kubernetes.Interface) error {
 	if err != nil {
 		return err
 	}
+	err = k.NewReplicaSetInformer(informerFactory)
+	if err != nil {
+		return err
+	}
 
-	log.Debugf("starting kubernetes informer, waiting for syncronization")
+	log.Debugf("starting kubernetes informers, waiting for syncronization")
 	informerFactory.Start(k.stopChan)
 	informerFactory.WaitForCacheSync(k.stopChan)
 	log.Debugf("kubernetes informers started")
 
 	return nil
 }
-
-func init() {
-	Data = KubeData{
-		informers: map[string]cache.SharedIndexInformer{},
-		stopChan:  make(chan struct{}),
-	}
-}

pkg/pipeline/transform/kubernetes/kubernetes_test.go

@@ -0,0 +1,89 @@
+/*
+ * Copyright (C) 2022 IBM, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package kubernetes
+
+import (
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+	"testing"
+)
+
+type IndexerMock struct {
+	mock.Mock
+	cache.Indexer
+}
+
+type InformerMock struct {
+	mock.Mock
+	InformerInterface
+}
+
+type InformerInterface interface {
+	cache.SharedInformer
+	AddIndexers(indexers cache.Indexers) error
+	GetIndexer() cache.Indexer
+}
+
+func (indexMock *IndexerMock) ByIndex(indexName, indexedValue string) ([]interface{}, error) {
+	pod := fakePod("podName", "podNamespace", "podHost")
+	podInterface := interface{}(pod)
+	return []interface{}{podInterface}, nil
+}
+
+func fakePod(name, ns, host string) *v1.Pod {
+	return &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+		},
+		Status: v1.PodStatus{
+			HostIP: host,
+		},
+	}
+}
+
+func (informerMock *InformerMock) GetIndexer() cache.Indexer {
+	informerMock.Mock.Called()
+	return &IndexerMock{}
+}
+
+func TestKubeData_getInfo(t *testing.T) {
+	// Test with no informer
+	kubeData := KubeData{}
+	info, err := kubeData.GetInfo("1.2.3.4")
+	require.EqualError(t, err, "can't find ip")
+	require.Nil(t, info)
+
+	// Test with mock pod informer
+	expectedInfo := &Info{
+		Type:      "Pod",
+		Name:      "podName",
+		Namespace: "podNamespace",
+		Owner:     Owner{Name: "podName", Type: "Pod"},
+	}
+	kubeData = KubeData{ipInformers: map[string]cache.SharedIndexInformer{}}
+	informerMock := &InformerMock{}
+	informerMock.On("GetIndexer", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	kubeData.ipInformers["Pod"] = informerMock
+	info, err = kubeData.GetInfo("1.2.3.4")
+	require.NoError(t, err)
+	require.Equal(t, info, expectedInfo)
+}

pkg/pipeline/transform/transform_network.go

@@ -130,9 +130,11 @@ func (n *Network) Transform(inputEntry config.GenericMap) config.GenericMap {
 				log.Infof("Can't find kubernetes info for IP %v err %v", outputEntries[rule.Input], err)
 				continue
 			}
-			outputEntries[rule.Output+"_Type"] = kubeInfo.Type
-			outputEntries[rule.Output+"_Name"] = kubeInfo.Name
 			outputEntries[rule.Output+"_Namespace"] = kubeInfo.Namespace
+			outputEntries[rule.Output+"_Name"] = kubeInfo.Name
+			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
 			}