Fixing pod name indexer to use both namespace, pod name to frame inde…

…x key (elastic#4775)

* Fixing pod name indexer to use both namespace, pod name to frame index key

* Adding field pattern matcher to support pod name indexer matching

CHANGELOG.asciidoc

@@ -81,6 +81,8 @@ https://github.com/elastic/beats/compare/v6.0.0-beta1...master[Check the HEAD di
 
 *Affecting all Beats*
 
+- Fix pod name indexer to use both namespace, pod name to frame index key {pull}4775[4775]
+
 *Filebeat*
 
 *Heartbeat*

libbeat/processors/add_kubernetes_metadata/indexing.go

@@ -6,13 +6,19 @@ import (
 	"sync"
 
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/fmtstr"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/outputs/codec"
+	"github.com/elastic/beats/libbeat/outputs/codec/format"
+	"github.com/elastic/beats/libbeat/publisher/beat"
 )
 
 //Names of indexers and matchers that have been defined.
 const (
-	PodNameIndexerName   = "pod_name"
-	FieldMatcherName     = "fields"
-	ContainerIndexerName = "container"
+	ContainerIndexerName   = "container"
+	PodNameIndexerName     = "pod_name"
+	FieldMatcherName       = "fields"
+	FieldFormatMatcherName = "field_format"
 )
 
 // Indexing is the singleton Register instance where all Indexers and Matchers
@@ -238,14 +244,14 @@ func (p *PodNameIndexer) GetMetadata(pod *Pod) []MetadataIndex {
 	data := p.genMeta.GenerateMetaData(pod)
 	return []MetadataIndex{
 		{
-			Index: pod.Metadata.Name,
+			Index: fmt.Sprintf("%s/%s", pod.Metadata.Namespace, pod.Metadata.Name),
 			Data:  data,
 		},
 	}
 }
 
 func (p *PodNameIndexer) GetIndexes(pod *Pod) []string {
-	return []string{pod.Metadata.Name}
+	return []string{fmt.Sprintf("%s/%s", pod.Metadata.Namespace, pod.Metadata.Name)}
 }
 
 // ContainerIndexer indexes pods based on all their containers IDs
@@ -323,3 +329,43 @@ func (f *FieldMatcher) MetadataIndex(event common.MapStr) string {
 
 	return ""
 }
+
+type FieldFormatMatcher struct {
+	Codec codec.Codec
+}
+
+func NewFieldFormatMatcher(cfg common.Config) (Matcher, error) {
+	config := struct {
+		Format string `config:"format"`
+	}{}
+
+	err := cfg.Unpack(&config)
+	if err != nil {
+		return nil, fmt.Errorf("fail to unpack the `format` configuration of `field_format` matcher: %s", err)
+	}
+
+	if config.Format == "" {
+		return nil, fmt.Errorf("`format` of `field_format` matcher can't be empty")
+	}
+
+	return &FieldFormatMatcher{
+		Codec: format.New(fmtstr.MustCompileEvent(config.Format)),
+	}, nil
+
+}
+
+func (f *FieldFormatMatcher) MetadataIndex(event common.MapStr) string {
+	bytes, err := f.Codec.Encode("", &beat.Event{
+		Fields: event,
+	})
+
+	if err != nil {
+		logp.Debug("kubernetes", "Unable to apply field format pattern on event")
+	}
+
+	if len(bytes) == 0 {
+		return ""
+	}
+
+	return string(bytes)
+}

libbeat/processors/add_kubernetes_metadata/indexing_test.go

@@ -1,6 +1,7 @@
 package add_kubernetes_metadata
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -31,7 +32,7 @@ func TestPodIndexer(t *testing.T) {
 
 	indexers := podIndexer.GetMetadata(&pod)
 	assert.Equal(t, len(indexers), 1)
-	assert.Equal(t, indexers[0].Index, podName)
+	assert.Equal(t, indexers[0].Index, fmt.Sprintf("%s/%s", ns, podName))
 
 	expected := common.MapStr{
 		"pod": common.MapStr{
@@ -47,7 +48,7 @@ func TestPodIndexer(t *testing.T) {
 
 	indices := podIndexer.GetIndexes(&pod)
 	assert.Equal(t, len(indices), 1)
-	assert.Equal(t, indices[0], podName)
+	assert.Equal(t, indices[0], fmt.Sprintf("%s/%s", ns, podName))
 }
 
 func TestContainerIndexer(t *testing.T) {
@@ -209,3 +210,49 @@ func TestFilteredGenMeta(t *testing.T) {
 	ok, _ = annotationsMap.HasKey("a")
 	assert.Equal(t, ok, true)
 }
+
+func TestFieldFormatMatcher(t *testing.T) {
+	testCfg := map[string]interface{}{}
+	fieldCfg, err := common.NewConfigFrom(testCfg)
+
+	assert.Nil(t, err)
+	matcher, err := NewFieldFormatMatcher(*fieldCfg)
+	assert.NotNil(t, err)
+
+	testCfg["format"] = `%{[namespace]}/%{[pod]}`
+	fieldCfg, _ = common.NewConfigFrom(testCfg)
+
+	matcher, err = NewFieldFormatMatcher(*fieldCfg)
+	assert.NotNil(t, matcher)
+	assert.Nil(t, err)
+
+	event := common.MapStr{
+		"namespace": "foo",
+		"pod":       "bar",
+	}
+
+	out := matcher.MetadataIndex(event)
+	assert.Equal(t, "foo/bar", out)
+
+	event = common.MapStr{
+		"foo": "bar",
+	}
+	out = matcher.MetadataIndex(event)
+	assert.Empty(t, out)
+
+	testCfg["format"] = `%{[dimensions.namespace]}/%{[dimensions.pod]}`
+	fieldCfg, _ = common.NewConfigFrom(testCfg)
+	matcher, err = NewFieldFormatMatcher(*fieldCfg)
+	assert.NotNil(t, matcher)
+	assert.Nil(t, err)
+
+	event = common.MapStr{
+		"dimensions": common.MapStr{
+			"pod":       "bar",
+			"namespace": "foo",
+		},
+	}
+
+	out = matcher.MetadataIndex(event)
+	assert.Equal(t, "foo/bar", out)
+}