Add owner object info to Kubernetes metadata (#7231)

This change adds object name for the Pod owner (if any) when enriching
with kubernetes metadata.

This information is really useful when mapping running containers to
their creator workload.

Resulting metadata will now include the name of the Deployment, ReplicaSet or ReplicaSet who created the Pod (if any), ie:

```
{
  "kubernetes":{
    "container":{
      "name":"kube-state-metrics"
    },
    "labels":{
      "k8s-app":"kube-state-metrics",
      "pod-template-hash":"2035844717"
    },
    "namespace":"kube-system",
    "node":{
      "name":"minikube"
    },
    "pod":{
      "name":"kube-state-metrics-6479d88c5c-5b6cl"
    },
    "replicaset":{
      "name":"kube-state-metrics-6479d88c5c"
    }
  }
}
```

CHANGELOG.asciidoc

@@ -145,6 +145,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Add support for loading a template.json file directly instead of using fields.yml. {pull}7039[7039]
 - Add support for keyword multifields in field.yml. {pull}7131[7131]
 - Add dissect processor. {pull}6925[6925]
+- Add owner object info to Kubernetes metadata. {pull}7231[7231]
 
 *Auditbeat*
 

libbeat/autodiscover/providers/kubernetes/config.go

@@ -16,11 +16,6 @@ type Config struct {
 	SyncPeriod     time.Duration `config:"sync_period"`
 	CleanupTimeout time.Duration `config:"cleanup_timeout"`
 
-	IncludeLabels      []string `config:"include_labels"`
-	ExcludeLabels      []string `config:"exclude_labels"`
-	IncludeAnnotations []string `config:"include_annotations"`
-	IncludePodUID      bool     `config:"include_pod_uid"`
-
 	Prefix       string                  `config:"prefix"`
 	HintsEnabled bool                    `config:"hints.enabled"`
 	Builders     []*common.Config        `config:"builders"`

libbeat/autodiscover/providers/kubernetes/kubernetes.go

@@ -43,7 +43,10 @@ func AutodiscoverBuilder(bus bus.Bus, c *common.Config) (autodiscover.Provider,
 		return nil, err
 	}
 
-	metagen := kubernetes.NewMetaGenerator(config.IncludeAnnotations, config.IncludeLabels, config.ExcludeLabels, config.IncludePodUID)
+	metagen, err := kubernetes.NewMetaGenerator(c)
+	if err != nil {
+		return nil, err
+	}
 
 	config.Host = kubernetes.DiscoverKubernetesNode(config.Host, config.InCluster, client)
 

libbeat/common/kubernetes/metadata.go

@@ -1,6 +1,8 @@
 package kubernetes
 
 import (
+	"strings"
+
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/safemapstr"
 )
@@ -15,39 +17,41 @@ type MetaGenerator interface {
 }
 
 type metaGenerator struct {
-	annotations   []string
-	labels        []string
-	labelsExclude []string
-	poduid        bool
+	IncludeLabels          []string `config:"include_labels"`
+	ExcludeLabels          []string `config:"exclude_labels"`
+	IncludeAnnotations     []string `config:"include_annotations"`
+	IncludePodUID          bool     `config:"include_pod_uid"`
+	IncludeCreatorMetadata bool     `config:"include_creator_metadata"`
 }
 
 // NewMetaGenerator initializes and returns a new kubernetes metadata generator
-func NewMetaGenerator(annotations, labels, labelsExclude []string, includePodUID bool) MetaGenerator {
-	return &metaGenerator{
-		annotations:   annotations,
-		labels:        labels,
-		labelsExclude: labelsExclude,
-		poduid:        includePodUID,
+func NewMetaGenerator(cfg *common.Config) (MetaGenerator, error) {
+	// default settings:
+	generator := metaGenerator{
+		IncludeCreatorMetadata: true,
 	}
+
+	err := cfg.Unpack(&generator)
+	return &generator, err
 }
 
 // PodMetadata generates metadata for the given pod taking to account certain filters
 func (g *metaGenerator) PodMetadata(pod *Pod) common.MapStr {
 	labelMap := common.MapStr{}
-	if len(g.labels) == 0 {
+	if len(g.IncludeLabels) == 0 {
 		for k, v := range pod.Metadata.Labels {
 			safemapstr.Put(labelMap, k, v)
 		}
 	} else {
-		labelMap = generateMapSubset(pod.Metadata.Labels, g.labels)
+		labelMap = generateMapSubset(pod.Metadata.Labels, g.IncludeLabels)
 	}
 
 	// Exclude any labels that are present in the exclude_labels config
-	for _, label := range g.labelsExclude {
+	for _, label := range g.ExcludeLabels {
 		delete(labelMap, label)
 	}
 
-	annotationsMap := generateMapSubset(pod.Metadata.Annotations, g.annotations)
+	annotationsMap := generateMapSubset(pod.Metadata.Annotations, g.IncludeAnnotations)
 	meta := common.MapStr{
 		"pod": common.MapStr{
 			"name": pod.Metadata.Name,
@@ -59,10 +63,25 @@ func (g *metaGenerator) PodMetadata(pod *Pod) common.MapStr {
 	}
 
 	// Add Pod UID metadata if enabled
-	if g.poduid {
+	if g.IncludePodUID {
 		safemapstr.Put(meta, "pod.uid", pod.Metadata.UID)
 	}
 
+	// Add controller metadata if present
+	if g.IncludeCreatorMetadata {
+		for _, ref := range pod.Metadata.OwnerReferences {
+			if ref.Controller {
+				switch ref.Kind {
+				// TODO grow this list as we keep adding more `state_*` metricsets
+				case "Deployment",
+					"ReplicaSet",
+					"StatefulSet":
+					safemapstr.Put(meta, strings.ToLower(ref.Kind)+".name", ref.Name)
+				}
+			}
+		}
+	}
+
 	if len(labelMap) != 0 {
 		meta["labels"] = labelMap
 	}

libbeat/common/kubernetes/metadata_test.go

@@ -9,10 +9,12 @@ import (
 )
 
 func TestPodMetadataDeDot(t *testing.T) {
+	withPodUID, _ := common.NewConfigFrom(map[string]interface{}{"include_pod_uid": true})
+
 	tests := []struct {
-		pod     *Pod
-		meta    common.MapStr
-		metaGen MetaGenerator
+		pod    *Pod
+		meta   common.MapStr
+		config *common.Config
 	}{
 		{
 			pod: &Pod{
@@ -27,7 +29,7 @@ func TestPodMetadataDeDot(t *testing.T) {
 				"node":      common.MapStr{"name": ""},
 				"labels":    common.MapStr{"a": common.MapStr{"value": "bar", "key": "foo"}},
 			},
-			metaGen: NewMetaGenerator(nil, nil, nil, false),
+			config: common.NewConfig(),
 		},
 		{
 			pod: &Pod{
@@ -42,11 +44,49 @@ func TestPodMetadataDeDot(t *testing.T) {
 				"node":      common.MapStr{"name": ""},
 				"labels":    common.MapStr{"a": common.MapStr{"value": "bar", "key": "foo"}},
 			},
-			metaGen: NewMetaGenerator(nil, nil, nil, true),
+			config: withPodUID,
+		},
+		{
+			pod: &Pod{
+				Metadata: ObjectMeta{
+					Labels: map[string]string{"a.key": "foo", "a": "bar"},
+					UID:    "005f3b90-4b9d-12f8-acf0-31020a840133",
+					OwnerReferences: []struct {
+						APIVersion string `json:"apiVersion"`
+						Controller bool   `json:"controller"`
+						Kind       string `json:"kind"`
+						Name       string `json:"name"`
+						UID        string `json:"uid"`
+					}{
+						{
+							Kind:       "Deployment",
+							Name:       "test",
+							Controller: true,
+						},
+						{
+							Kind:       "Replicaset",
+							Name:       "replicaset",
+							Controller: false,
+						},
+					},
+				},
+			},
+			meta: common.MapStr{
+				"pod":        common.MapStr{"name": ""},
+				"namespace":  "",
+				"node":       common.MapStr{"name": ""},
+				"labels":     common.MapStr{"a": common.MapStr{"value": "bar", "key": "foo"}},
+				"deployment": common.MapStr{"name": "test"},
+			},
+			config: common.NewConfig(),
 		},
 	}
 
 	for _, test := range tests {
-		assert.Equal(t, test.metaGen.PodMetadata(test.pod), test.meta)
+		metaGen, err := NewMetaGenerator(test.config)
+		if err != nil {
+			t.Fatal(err)
+		}
+		assert.Equal(t, metaGen.PodMetadata(test.pod), test.meta)
 	}
 }

libbeat/processors/add_kubernetes_metadata/config.go

@@ -14,15 +14,11 @@ type kubeAnnotatorConfig struct {
 	SyncPeriod time.Duration `config:"sync_period"`
 	// Annotations are kept after pod is removed, until they haven't been accessed
 	// for a full `cleanup_timeout`:
-	CleanupTimeout     time.Duration `config:"cleanup_timeout"`
-	Indexers           PluginConfig  `config:"indexers"`
-	Matchers           PluginConfig  `config:"matchers"`
-	DefaultMatchers    Enabled       `config:"default_matchers"`
-	DefaultIndexers    Enabled       `config:"default_indexers"`
-	IncludeLabels      []string      `config:"include_labels"`
-	ExcludeLabels      []string      `config:"exclude_labels"`
-	IncludeAnnotations []string      `config:"include_annotations"`
-	IncludePodUID      bool          `config:"include_pod_uid"`
+	CleanupTimeout  time.Duration `config:"cleanup_timeout"`
+	Indexers        PluginConfig  `config:"indexers"`
+	Matchers        PluginConfig  `config:"matchers"`
+	DefaultMatchers Enabled       `config:"default_matchers"`
+	DefaultIndexers Enabled       `config:"default_indexers"`
 }
 
 type Enabled struct {

libbeat/processors/add_kubernetes_metadata/indexers_test.go

@@ -10,7 +10,7 @@ import (
 	"github.com/elastic/beats/libbeat/common/kubernetes"
 )
 
-var metagen = kubernetes.NewMetaGenerator([]string{}, []string{}, []string{}, false)
+var metagen, _ = kubernetes.NewMetaGenerator(common.NewConfig())
 
 func TestPodIndexer(t *testing.T) {
 	var testConfig = common.NewConfig()
@@ -60,7 +60,8 @@ func TestPodIndexer(t *testing.T) {
 func TestPodUIDIndexer(t *testing.T) {
 	var testConfig = common.NewConfig()
 
-	metaGenWithPodUID := kubernetes.NewMetaGenerator([]string{}, []string{}, []string{}, true)
+	metaGenWithPodUID, err := kubernetes.NewMetaGenerator(common.NewConfig())
+	assert.Nil(t, err)
 
 	podUIDIndexer, err := NewPodUIDIndexer(*testConfig, metaGenWithPodUID)
 	assert.Nil(t, err)
@@ -89,7 +90,6 @@ func TestPodUIDIndexer(t *testing.T) {
 	expected := common.MapStr{
 		"pod": common.MapStr{
 			"name": "testpod",
-			"uid":  "005f3b90-4b9d-12f8-acf0-31020a840133",
 		},
 		"namespace": "testns",
 		"labels": common.MapStr{
@@ -220,7 +220,15 @@ func TestFilteredGenMeta(t *testing.T) {
 	rawAnnotations := indexers[0].Data["annotations"]
 	assert.Nil(t, rawAnnotations)
 
-	filteredGen := kubernetes.NewMetaGenerator([]string{"a"}, []string{"foo"}, []string{}, false)
+	config, err := common.NewConfigFrom(map[string]interface{}{
+		"include_annotations": []string{"a"},
+		"include_labels":      []string{"foo"},
+	})
+	assert.Nil(t, err)
+
+	filteredGen, err := kubernetes.NewMetaGenerator(config)
+	assert.Nil(t, err)
+
 	podIndexer, err = NewPodNameIndexer(*testConfig, filteredGen)
 	assert.Nil(t, err)
 
@@ -251,7 +259,14 @@ func TestFilteredGenMeta(t *testing.T) {
 func TestFilteredGenMetaExclusion(t *testing.T) {
 	var testConfig = common.NewConfig()
 
-	filteredGen := kubernetes.NewMetaGenerator([]string{}, []string{}, []string{"x"}, false)
+	config, err := common.NewConfigFrom(map[string]interface{}{
+		"exclude_labels": []string{"x"},
+	})
+	assert.Nil(t, err)
+
+	filteredGen, err := kubernetes.NewMetaGenerator(config)
+	assert.Nil(t, err)
+
 	podIndexer, err := NewPodNameIndexer(*testConfig, filteredGen)
 	assert.Nil(t, err)
 

libbeat/processors/add_kubernetes_metadata/kubernetes.go

@@ -66,7 +66,11 @@ func newKubernetesAnnotator(cfg *common.Config) (processors.Processor, error) {
 		Indexing.RUnlock()
 	}
 
-	metaGen := kubernetes.NewMetaGenerator(config.IncludeAnnotations, config.IncludeLabels, config.ExcludeLabels, config.IncludePodUID)
+	metaGen, err := kubernetes.NewMetaGenerator(cfg)
+	if err != nil {
+		return nil, err
+	}
+
 	indexers := NewIndexers(config.Indexers, metaGen)
 
 	matchers := NewMatchers(config.Matchers)