Refactor metadata generator to support adding metadata across resourc…

…es (elastic#14875)

* Refactor metagen to allow multiple resources to be enriched

CHANGELOG.next.asciidoc

@@ -29,6 +29,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Libbeat: Do not overwrite agent.*, ecs.version, and host.name. {pull}14407[14407]
 - Libbeat: Cleanup the x-pack licenser code to use the new license endpoint and the new format. {pull}15091[15091]
 - Users can now specify `monitoring.cloud.*` to override `monitoring.elasticsearch.*` settings. {issue}14399[14399] {pull}15254[15254]
+- Refactor metadata generator to support adding metadata across resources {pull}14875[14875]
 - Update to ECS 1.4.0. {pull}14844[14844]
 
 *Auditbeat*

libbeat/autodiscover/providers/kubernetes/config.go

@@ -23,6 +23,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/elastic/beats/libbeat/common/kubernetes/metadata"
+
 	"github.com/elastic/beats/libbeat/autodiscover/template"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
@@ -48,6 +50,8 @@ type Config struct {
 	Builders  []*common.Config        `config:"builders"`
 	Appenders []*common.Config        `config:"appenders"`
 	Templates template.MapperSettings `config:"templates"`
+
+	AddResourceMetadata *metadata.AddResourceMetadataConfig `config:"add_resource_metadata"`
 }
 
 func defaultConfig() *Config {

libbeat/autodiscover/providers/kubernetes/node.go

@@ -29,30 +29,26 @@ import (
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/bus"
 	"github.com/elastic/beats/libbeat/common/kubernetes"
+	"github.com/elastic/beats/libbeat/common/kubernetes/metadata"
 	"github.com/elastic/beats/libbeat/common/safemapstr"
 	"github.com/elastic/beats/libbeat/logp"
 )
 
 type node struct {
 	uuid    uuid.UUID
 	config  *Config
-	metagen kubernetes.MetaGenerator
+	metagen metadata.MetaGen
 	logger  *logp.Logger
 	publish func(bus.Event)
 	watcher kubernetes.Watcher
 }
 
 // NewNodeEventer creates an eventer that can discover and process node objects
 func NewNodeEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface, publish func(event bus.Event)) (Eventer, error) {
-	metagen, err := kubernetes.NewMetaGenerator(cfg)
-	if err != nil {
-		return nil, err
-	}
-
 	logger := logp.NewLogger("autodiscover.node")
 
 	config := defaultConfig()
-	err = cfg.Unpack(&config)
+	err := cfg.Unpack(&config)
 	if err != nil {
 		return nil, err
 	}
@@ -70,7 +66,7 @@ func NewNodeEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface, pu
 	watcher, err := kubernetes.NewWatcher(client, &kubernetes.Node{}, kubernetes.WatchOptions{
 		SyncTimeout: config.SyncPeriod,
 		Node:        config.Node,
-	})
+	}, nil)
 
 	if err != nil {
 		return nil, fmt.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Node{}, err)
@@ -80,7 +76,7 @@ func NewNodeEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface, pu
 		config:  config,
 		uuid:    uuid,
 		publish: publish,
-		metagen: metagen,
+		metagen: metadata.NewNodeMetadataGenerator(cfg, watcher.Store()),
 		logger:  logger,
 		watcher: watcher,
 	}
@@ -172,11 +168,7 @@ func (n *node) emit(node *kubernetes.Node, flag string) {
 	}
 
 	eventID := fmt.Sprint(node.GetObjectMeta().GetUID())
-	meta := n.metagen.ResourceMetadata(node)
-
-	// TODO: Refactor metagen to make sure that this is seamless
-	meta.Put("node.name", node.Name)
-	meta.Put("node.uid", string(node.GetObjectMeta().GetUID()))
+	meta := n.metagen.Generate(node)
 
 	kubemeta := meta.Clone()
 	// Pass annotations to all events so that it can be used in templating and by annotation builders.

libbeat/autodiscover/providers/kubernetes/node_test.go

@@ -21,6 +21,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/elastic/beats/libbeat/common/kubernetes/metadata"
+
 	"github.com/gofrs/uuid"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
@@ -114,6 +116,11 @@ func TestEmitEvent_Node(t *testing.T) {
 	nodeIP := "192.168.0.1"
 	uid := "005f3b90-4b9d-12f8-acf0-31020a840133"
 	UUID, err := uuid.NewV4()
+
+	typeMeta := metav1.TypeMeta{
+		Kind:       "Node",
+		APIVersion: "v1",
+	}
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -134,6 +141,7 @@ func TestEmitEvent_Node(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
+				TypeMeta: typeMeta,
 				Status: v1.NodeStatus{
 					Addresses: []v1.NodeAddress{
 						{
@@ -180,7 +188,8 @@ func TestEmitEvent_Node(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
-				Status: v1.NodeStatus{},
+				TypeMeta: typeMeta,
+				Status:   v1.NodeStatus{},
 			},
 			Expected: nil,
 		},
@@ -194,6 +203,7 @@ func TestEmitEvent_Node(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
+				TypeMeta: typeMeta,
 				Status: v1.NodeStatus{
 					Addresses: []v1.NodeAddress{},
 					Conditions: []v1.NodeCondition{
@@ -236,11 +246,7 @@ func TestEmitEvent_Node(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			metaGen, err := kubernetes.NewMetaGenerator(common.NewConfig())
-			if err != nil {
-				t.Fatal(err)
-			}
-
+			metaGen := metadata.NewNodeMetadataGenerator(common.NewConfig(), nil)
 			p := &Provider{
 				config:    defaultConfig(),
 				bus:       bus.New("test"),

libbeat/autodiscover/providers/kubernetes/pod.go

@@ -25,34 +25,31 @@ import (
 	k8s "k8s.io/client-go/kubernetes"
 
 	"github.com/elastic/beats/libbeat/autodiscover/builder"
-
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/bus"
 	"github.com/elastic/beats/libbeat/common/kubernetes"
+	"github.com/elastic/beats/libbeat/common/kubernetes/metadata"
 	"github.com/elastic/beats/libbeat/common/safemapstr"
 	"github.com/elastic/beats/libbeat/logp"
 )
 
 type pod struct {
-	uuid    uuid.UUID
-	config  *Config
-	metagen kubernetes.MetaGenerator
-	logger  *logp.Logger
-	publish func(bus.Event)
-	watcher kubernetes.Watcher
+	uuid             uuid.UUID
+	config           *Config
+	metagen          metadata.MetaGen
+	logger           *logp.Logger
+	publish          func(bus.Event)
+	watcher          kubernetes.Watcher
+	nodeWatcher      kubernetes.Watcher
+	namespaceWatcher kubernetes.Watcher
 }
 
 // NewPodEventer creates an eventer that can discover and process pod objects
 func NewPodEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface, publish func(event bus.Event)) (Eventer, error) {
-	metagen, err := kubernetes.NewMetaGenerator(cfg)
-	if err != nil {
-		return nil, err
-	}
-
 	logger := logp.NewLogger("autodiscover.pod")
 
 	config := defaultConfig()
-	err = cfg.Unpack(&config)
+	err := cfg.Unpack(&config)
 	if err != nil {
 		return nil, err
 	}
@@ -71,18 +68,52 @@ func NewPodEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface, pub
 		SyncTimeout: config.SyncPeriod,
 		Node:        config.Node,
 		Namespace:   config.Namespace,
-	})
+	}, nil)
 	if err != nil {
 		return nil, fmt.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Pod{}, err)
 	}
 
+	var nodeMeta, namespaceMeta metadata.MetaGen
+	var nodeWatcher, namespaceWatcher kubernetes.Watcher
+	metaConf := config.AddResourceMetadata
+	if metaConf != nil {
+		if metaConf.Node != nil && metaConf.Node.Enabled() {
+			options := kubernetes.WatchOptions{
+				SyncTimeout: config.SyncPeriod,
+				Node:        config.Node,
+			}
+			if config.Namespace != "" {
+				options.Namespace = config.Namespace
+			}
+			nodeWatcher, err = kubernetes.NewWatcher(client, &kubernetes.Node{}, options, nil)
+			if err != nil {
+				return nil, fmt.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Node{}, err)
+			}
+
+			nodeMeta = metadata.NewNodeMetadataGenerator(metaConf.Node, nodeWatcher.Store())
+		}
+
+		if metaConf.Namespace != nil && metaConf.Namespace.Enabled() {
+			namespaceWatcher, err = kubernetes.NewWatcher(client, &kubernetes.Namespace{}, kubernetes.WatchOptions{
+				SyncTimeout: config.SyncPeriod,
+			}, nil)
+			if err != nil {
+				return nil, fmt.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
+			}
+
+			namespaceMeta = metadata.NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store())
+		}
+	}
+
 	p := &pod{
-		config:  config,
-		uuid:    uuid,
-		publish: publish,
-		metagen: metagen,
-		logger:  logger,
-		watcher: watcher,
+		config:           config,
+		uuid:             uuid,
+		publish:          publish,
+		metagen:          metadata.NewPodMetadataGenerator(cfg, watcher.Store(), nodeMeta, namespaceMeta),
+		logger:           logger,
+		watcher:          watcher,
+		nodeWatcher:      nodeWatcher,
+		namespaceWatcher: namespaceWatcher,
 	}
 
 	watcher.AddEventHandler(p)
@@ -168,12 +199,33 @@ func (p *pod) GenerateHints(event bus.Event) bus.Event {
 
 // Start starts the eventer
 func (p *pod) Start() error {
+	if p.nodeWatcher != nil {
+		err := p.nodeWatcher.Start()
+		if err != nil {
+			return err
+		}
+	}
+
+	if p.namespaceWatcher != nil {
+		if err := p.namespaceWatcher.Start(); err != nil {
+			return err
+		}
+	}
+
 	return p.watcher.Start()
 }
 
 // Stop stops the eventer
 func (p *pod) Stop() {
 	p.watcher.Stop()
+
+	if p.namespaceWatcher != nil {
+		p.namespaceWatcher.Stop()
+	}
+
+	if p.nodeWatcher != nil {
+		p.nodeWatcher.Stop()
+	}
 }
 
 func (p *pod) emit(pod *kubernetes.Pod, flag string) {
@@ -231,7 +283,8 @@ func (p *pod) emitEvents(pod *kubernetes.Pod, flag string, containers []kubernet
 			"image":   c.Image,
 			"runtime": runtimes[c.Name],
 		}
-		meta := p.metagen.ContainerMetadata(pod, c.Name, c.Image)
+		meta := p.metagen.Generate(pod, metadata.WithFields("container.name", c.Name),
+			metadata.WithFields("container.image", c.Image))
 
 		// Information that can be used in discovering a workload
 		kubemeta := meta.Clone()

libbeat/autodiscover/providers/kubernetes/pod_test.go

@@ -21,6 +21,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/elastic/beats/libbeat/common/kubernetes/metadata"
+
 	"github.com/gofrs/uuid"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
@@ -175,6 +177,11 @@ func TestEmitEvent(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	typeMeta := metav1.TypeMeta{
+		Kind:       "Pod",
+		APIVersion: "v1",
+	}
+
 	tests := []struct {
 		Message  string
 		Flag     string
@@ -192,6 +199,7 @@ func TestEmitEvent(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
+				TypeMeta: typeMeta,
 				Status: v1.PodStatus{
 					PodIP: podIP,
 					ContainerStatuses: []kubernetes.PodContainerStatus{
@@ -264,6 +272,7 @@ func TestEmitEvent(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
+				TypeMeta: typeMeta,
 				Status: v1.PodStatus{
 					ContainerStatuses: []kubernetes.PodContainerStatus{
 						{
@@ -295,6 +304,7 @@ func TestEmitEvent(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
+				TypeMeta: typeMeta,
 				Status: v1.PodStatus{
 					PodIP: podIP,
 					ContainerStatuses: []kubernetes.PodContainerStatus{
@@ -326,6 +336,7 @@ func TestEmitEvent(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
+				TypeMeta: typeMeta,
 				Status: v1.PodStatus{
 					ContainerStatuses: []kubernetes.PodContainerStatus{
 						{
@@ -393,6 +404,7 @@ func TestEmitEvent(t *testing.T) {
 					Labels:      map[string]string{},
 					Annotations: map[string]string{},
 				},
+				TypeMeta: typeMeta,
 				Status: v1.PodStatus{
 					PodIP: podIP,
 					ContainerStatuses: []kubernetes.PodContainerStatus{
@@ -459,11 +471,7 @@ func TestEmitEvent(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			metaGen, err := kubernetes.NewMetaGenerator(common.NewConfig())
-			if err != nil {
-				t.Fatal(err)
-			}
-
+			metaGen := metadata.NewPodMetadataGenerator(common.NewConfig(), nil, nil, nil)
 			p := &Provider{
 				config:    defaultConfig(),
 				bus:       bus.New("test"),