Add owner object info to Kubernetes metadata

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)