Include owner object reference info

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.

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/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,15 +17,20 @@ type MetaGenerator interface {
 }
 
 type metaGenerator struct {
-	IncludeLabels      []string `config:"include_labels"`
-	ExcludeLabels      []string `config:"exclude_labels"`
-	IncludeAnnotations []string `config:"include_annotations"`
-	IncludePodUID      bool     `config:"include_pod_uid"`
+	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(cfg *common.Config) (MetaGenerator, error) {
-	generator := metaGenerator{}
+	// default settings:
+	generator := metaGenerator{
+		IncludeCreatorMetadata: true,
+	}
+
 	err := cfg.Unpack(&generator)
 	return &generator, err
 }
@@ -60,6 +67,21 @@ func (g *metaGenerator) PodMetadata(pod *Pod) common.MapStr {
 		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

@@ -46,6 +46,40 @@ func TestPodMetadataDeDot(t *testing.T) {
 			},
 			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 {