Add k8s cluster identifier

deploy/kubernetes/auditbeat-kubernetes.yaml

@@ -237,6 +237,7 @@ rules:
   - nodes
   - namespaces
   - pods
+  - configmaps
   verbs: ["get", "list", "watch"]
 - apiGroups: ["apps"]
   resources:

deploy/kubernetes/auditbeat/auditbeat-role.yaml

@@ -10,6 +10,7 @@ rules:
   - nodes
   - namespaces
   - pods
+  - configmaps
   verbs: ["get", "list", "watch"]
 - apiGroups: ["apps"]
   resources:

deploy/kubernetes/elastic-agent-standalone-kubernetes.yaml

@@ -562,6 +562,7 @@ rules:
       - events
       - pods
       - services
+      - configmaps
     verbs: ["get", "list", "watch"]
   # Enable this rule only if planing to use kubernetes_secrets provider
   #- apiGroups: [""]

deploy/kubernetes/elastic-agent-standalone/elastic-agent-standalone-role.yaml

@@ -12,6 +12,7 @@ rules:
       - events
       - pods
       - services
+      - configmaps
     verbs: ["get", "list", "watch"]
   # Enable this rule only if planing to use kubernetes_secrets provider
   #- apiGroups: [""]

deploy/kubernetes/filebeat-kubernetes.yaml

@@ -152,6 +152,7 @@ rules:
   - namespaces
   - pods
   - nodes
+  - configmaps
   verbs:
   - get
   - watch

deploy/kubernetes/filebeat/filebeat-role.yaml

@@ -10,6 +10,7 @@ rules:
   - namespaces
   - pods
   - nodes
+  - configmaps
   verbs:
   - get
   - watch

deploy/kubernetes/heartbeat-kubernetes.yaml

@@ -148,6 +148,7 @@ rules:
   - namespaces
   - pods
   - services
+  - configmaps
   verbs: ["get", "list", "watch"]
 - apiGroups: ["apps"]
   resources:

deploy/kubernetes/heartbeat/heartbeat-role.yaml

@@ -11,6 +11,7 @@ rules:
   - namespaces
   - pods
   - services
+  - configmaps
   verbs: ["get", "list", "watch"]
 - apiGroups: ["apps"]
   resources:

deploy/kubernetes/metricbeat-kubernetes.yaml

@@ -258,6 +258,7 @@ rules:
   - events
   - pods
   - services
+  - configmaps
   verbs: ["get", "list", "watch"]
 # Enable this rule only if planing to use Kubernetes keystore
 #- apiGroups: [""]

deploy/kubernetes/metricbeat/metricbeat-role.yaml

@@ -12,6 +12,7 @@ rules:
   - events
   - pods
   - services
+  - configmaps
   verbs: ["get", "list", "watch"]
 # Enable this rule only if planing to use Kubernetes keystore
 #- apiGroups: [""]

libbeat/autodiscover/providers/kubernetes/node.go

@@ -79,7 +79,7 @@ func NewNodeEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface, pu
 		config:  config,
 		uuid:    uuid,
 		publish: publish,
-		metagen: metadata.NewNodeMetadataGenerator(cfg, watcher.Store()),
+		metagen: metadata.NewNodeMetadataGenerator(cfg, watcher.Store(), client),
 		logger:  logger,
 		watcher: watcher,
 	}
@@ -177,7 +177,9 @@ func (n *node) emit(node *kubernetes.Node, flag string) {
 	eventID := fmt.Sprint(node.GetObjectMeta().GetUID())
 	meta := n.metagen.Generate(node)
 
-	kubemeta := meta.Clone()
+	kubemetaMap, _ := meta.GetValue("kubernetes")
+	kubemeta, _ := kubemetaMap.(common.MapStr)
+	kubemeta = kubemeta.Clone()
 	// Pass annotations to all events so that it can be used in templating and by annotation builders.
 	annotations := common.MapStr{}
 	for k, v := range node.GetObjectMeta().GetAnnotations() {
@@ -190,9 +192,7 @@ func (n *node) emit(node *kubernetes.Node, flag string) {
 		flag:         true,
 		"host":       host,
 		"kubernetes": kubemeta,
-		"meta": common.MapStr{
-			"kubernetes": meta,
-		},
+		"meta":       meta,
 	}
 	n.publish([]bus.Event{event})
 }

libbeat/autodiscover/providers/kubernetes/node_test.go

@@ -26,6 +26,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 
 	"github.com/elastic/beats/v7/libbeat/autodiscover/template"
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -111,6 +112,7 @@ func TestGenerateHints_Node(t *testing.T) {
 }
 
 func TestEmitEvent_Node(t *testing.T) {
+	client := k8sfake.NewSimpleClientset()
 	name := "metricbeat"
 	nodeIP := "192.168.0.1"
 	uid := "005f3b90-4b9d-12f8-acf0-31020a840133"
@@ -310,7 +312,7 @@ func TestEmitEvent_Node(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			metaGen := metadata.NewNodeMetadataGenerator(common.NewConfig(), nil)
+			metaGen := metadata.NewNodeMetadataGenerator(common.NewConfig(), nil, client)
 			config := defaultConfig()
 			p := &Provider{
 				config:    config,

libbeat/autodiscover/providers/kubernetes/pod.go

@@ -339,7 +339,7 @@ func (p *pod) emit(pod *kubernetes.Pod, flag string) {
 // configured port, a single event is created, with the port set to 0.
 // Host and port information is only included if the container is
 // running.
-// If the container ID is unkown, only "stop" events are generated.
+// If the container ID is unknown, only "stop" events are generated.
 // It also returns a map with the named ports.
 func (p *pod) containerPodEvents(flag string, pod *kubernetes.Pod, c *containerInPod, annotations, namespaceAnnotations common.MapStr) ([]bus.Event, common.MapStr) {
 	if c.id == "" && flag != "stop" {
@@ -361,7 +361,9 @@ func (p *pod) containerPodEvents(flag string, pod *kubernetes.Pod, c *containerI
 	}
 
 	// Information that can be used in discovering a workload
-	kubemeta := meta.Clone()
+	kubemetaMap, _ := meta.GetValue("kubernetes")
+	kubemeta, _ := kubemetaMap.(common.MapStr)
+	kubemeta = kubemeta.Clone()
 	kubemeta["annotations"] = annotations
 	kubemeta["container"] = common.MapStr{
 		"id":      c.id,
@@ -383,17 +385,17 @@ func (p *pod) containerPodEvents(flag string, pod *kubernetes.Pod, c *containerI
 
 	var events []bus.Event
 	portsMap := common.MapStr{}
+
+	meta.Put("container", cmeta)
+
 	for _, port := range ports {
 		event := bus.Event{
 			"provider":   p.uuid,
 			"id":         eventID,
 			flag:         true,
 			"kubernetes": kubemeta,
 			// Actual metadata that will enrich the event.
-			"meta": common.MapStr{
-				"kubernetes": meta,
-				"container":  cmeta,
-			},
+			"meta": meta,
 		}
 		// Include network information only if the container is running,
 		// so templates that need network don't generate a config.
@@ -417,7 +419,9 @@ func (p *pod) podEvent(flag string, pod *kubernetes.Pod, ports common.MapStr, in
 	meta := p.metagen.Generate(pod)
 
 	// Information that can be used in discovering a workload
-	kubemeta := meta.Clone()
+	kubemetaMap, _ := meta.GetValue("kubernetes")
+	kubemeta, _ := kubemetaMap.(common.MapStr)
+	kubemeta = kubemeta.Clone()
 	kubemeta["annotations"] = annotations
 	if len(namespaceAnnotations) != 0 {
 		kubemeta["namespace_annotations"] = namespaceAnnotations
@@ -429,9 +433,7 @@ func (p *pod) podEvent(flag string, pod *kubernetes.Pod, ports common.MapStr, in
 		"id":         fmt.Sprint(pod.GetObjectMeta().GetUID()),
 		flag:         true,
 		"kubernetes": kubemeta,
-		"meta": common.MapStr{
-			"kubernetes": meta,
-		},
+		"meta":       meta,
 	}
 
 	// Include network information only if the pod has an IP and there is any

libbeat/autodiscover/providers/kubernetes/pod_test.go

@@ -27,6 +27,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 
 	"github.com/elastic/beats/v7/libbeat/autodiscover/template"
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -331,7 +332,7 @@ func TestGenerateHints(t *testing.T) {
 	}
 }
 
-func TestEmitEvent(t *testing.T) {
+func TestPod_EmitEvent(t *testing.T) {
 	name := "filebeat"
 	namespace := "default"
 	podIP := "127.0.0.1"
@@ -1898,14 +1899,15 @@ func TestEmitEvent(t *testing.T) {
 		},
 	}
 
+	client := k8sfake.NewSimpleClientset()
 	for _, test := range tests {
 		t.Run(test.Message, func(t *testing.T) {
 			mapper, err := template.NewConfigMapper(nil, nil, nil)
 			if err != nil {
 				t.Fatal(err)
 			}
 
-			metaGen := metadata.NewPodMetadataGenerator(common.NewConfig(), nil, nil, nil, nil)
+			metaGen := metadata.NewPodMetadataGenerator(common.NewConfig(), nil, client, nil, nil)
 			p := &Provider{
 				config:    defaultConfig(),
 				bus:       bus.New(logp.NewLogger("bus"), "test"),

libbeat/autodiscover/providers/kubernetes/service.go

@@ -75,13 +75,13 @@ func NewServiceEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface,
 		return nil, fmt.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
 	}
 
-	namespaceMeta = metadata.NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store())
+	namespaceMeta = metadata.NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store(), client)
 
 	p := &service{
 		config:           config,
 		uuid:             uuid,
 		publish:          publish,
-		metagen:          metadata.NewServiceMetadataGenerator(cfg, watcher.Store(), namespaceMeta),
+		metagen:          metadata.NewServiceMetadataGenerator(cfg, watcher.Store(), namespaceMeta, client),
 		namespaceWatcher: namespaceWatcher,
 		logger:           logger,
 		watcher:          watcher,
@@ -193,7 +193,9 @@ func (s *service) emit(svc *kubernetes.Service, flag string) {
 	eventID := fmt.Sprint(svc.GetObjectMeta().GetUID())
 	meta := s.metagen.Generate(svc)
 
-	kubemeta := meta.Clone()
+	kubemetaMap, _ := meta.GetValue("kubernetes")
+	kubemeta, _ := kubemetaMap.(common.MapStr)
+	kubemeta = kubemeta.Clone()
 	// Pass annotations to all events so that it can be used in templating and by annotation builders.
 	annotations := common.MapStr{}
 	for k, v := range svc.GetObjectMeta().GetAnnotations() {
@@ -223,9 +225,7 @@ func (s *service) emit(svc *kubernetes.Service, flag string) {
 			"host":       host,
 			"port":       int(port.Port),
 			"kubernetes": kubemeta,
-			"meta": common.MapStr{
-				"kubernetes": meta,
-			},
+			"meta":       meta,
 		}
 		events = append(events, event)
 	}

libbeat/autodiscover/providers/kubernetes/service_test.go

@@ -26,6 +26,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
 
 	"github.com/elastic/beats/v7/libbeat/autodiscover/template"
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -232,6 +233,7 @@ func TestGenerateHints_Service(t *testing.T) {
 }
 
 func TestEmitEvent_Service(t *testing.T) {
+	client := k8sfake.NewSimpleClientset()
 	name := "metricbeat"
 	namespace := "default"
 	clusterIP := "192.168.0.1"
@@ -395,7 +397,7 @@ func TestEmitEvent_Service(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			metaGen := metadata.NewServiceMetadataGenerator(common.NewConfig(), nil, nil)
+			metaGen := metadata.NewServiceMetadataGenerator(common.NewConfig(), nil, nil, client)
 
 			p := &Provider{
 				config:    defaultConfig(),

libbeat/common/kubernetes/metadata/config.go

@@ -21,6 +21,7 @@ import "github.com/elastic/beats/v7/libbeat/common"
 
 // Config declares supported configuration for metadata generation
 type Config struct {
+	KubeConfig         string   `config:"kube_config"`
 	IncludeLabels      []string `config:"include_labels"`
 	ExcludeLabels      []string `config:"exclude_labels"`
 	IncludeAnnotations []string `config:"include_annotations"`