Automatically enrich Kubernetes module events

auditbeat/docs/fields.asciidoc

@@ -3312,7 +3312,7 @@ Kubernetes pod name
 --
 type: keyword
 
-Kubernetes pod uid
+Kubernetes Pod UID
 
 
 --

deploy/kubernetes/metricbeat-kubernetes.yaml

@@ -262,15 +262,22 @@ metadata:
   labels:
     k8s-app: metricbeat
 rules:
-- apiGroups: [""] # "" indicates the core API group
+- apiGroups: [""]
   resources:
+  - nodes
   - namespaces
   - events
   - pods
-  verbs:
-  - get
-  - watch
-  - list
+  verbs: ["get", "list", "watch"]
+- apiGroups: ["extensions"]
+  resources:
+  - deployments
+  - replicasets
+  verbs: ["get", "list", "watch"]
+- apiGroups: ["apps"]
+  resources:
+  - statefulsets
+  verbs: ["get", "list", "watch"]
 ---
 apiVersion: v1
 kind: ServiceAccount

deploy/kubernetes/metricbeat/metricbeat-role.yaml

@@ -5,12 +5,19 @@ metadata:
   labels:
     k8s-app: metricbeat
 rules:
-- apiGroups: [""] # "" indicates the core API group
+- apiGroups: [""]
   resources:
+  - nodes
   - namespaces
   - events
   - pods
-  verbs:
-  - get
-  - watch
-  - list
+  verbs: ["get", "list", "watch"]
+- apiGroups: ["extensions"]
+  resources:
+  - deployments
+  - replicasets
+  verbs: ["get", "list", "watch"]
+- apiGroups: ["apps"]
+  resources:
+  - statefulsets
+  verbs: ["get", "list", "watch"]

filebeat/docs/fields.asciidoc

@@ -2343,7 +2343,7 @@ Kubernetes pod name
 --
 type: keyword
 
-Kubernetes pod uid
+Kubernetes Pod UID
 
 
 --

heartbeat/docs/fields.asciidoc

@@ -664,7 +664,7 @@ Kubernetes pod name
 --
 type: keyword
 
-Kubernetes pod uid
+Kubernetes Pod UID
 
 
 --

libbeat/common/kubernetes/metadata.go

@@ -26,21 +26,29 @@ import (
 
 // MetaGenerator builds metadata objects for pods and containers
 type MetaGenerator interface {
+	// ResourceMetadata generates metadata for the given kubernetes object taking to account certain filters
+	ResourceMetadata(obj Resource) common.MapStr
+
 	// PodMetadata generates metadata for the given pod taking to account certain filters
 	PodMetadata(pod *Pod) common.MapStr
 
 	// Containermetadata generates metadata for the given container of a pod
 	ContainerMetadata(pod *Pod, container string) common.MapStr
 }
 
-type metaGenerator struct {
-	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"`
+// MetaGeneratorConfig settings
+type MetaGeneratorConfig struct {
+	IncludeLabels      []string `config:"include_labels"`
+	ExcludeLabels      []string `config:"exclude_labels"`
+	IncludeAnnotations []string `config:"include_annotations"`
+
+	// Undocumented settings, to be deprecated in favor of `drop_fields` processor:
+	IncludePodUID          bool `config:"include_pod_uid"`
+	IncludeCreatorMetadata bool `config:"include_creator_metadata"`
 }
 
+type metaGenerator = MetaGeneratorConfig
+
 // NewMetaGenerator initializes and returns a new kubernetes metadata generator
 func NewMetaGenerator(cfg *common.Config) (MetaGenerator, error) {
 	// default settings:
@@ -52,41 +60,37 @@ func NewMetaGenerator(cfg *common.Config) (MetaGenerator, error) {
 	return &generator, err
 }
 
-// PodMetadata generates metadata for the given pod taking to account certain filters
-func (g *metaGenerator) PodMetadata(pod *Pod) common.MapStr {
+// NewMetaGeneratorFromConfig initializes and returns a new kubernetes metadata generator
+func NewMetaGeneratorFromConfig(cfg *MetaGeneratorConfig) MetaGenerator {
+	return cfg
+}
+
+// ResourceMetadata generates metadata for the given kubernetes object taking to account certain filters
+func (g *metaGenerator) ResourceMetadata(obj Resource) common.MapStr {
+	objMeta := obj.GetMetadata()
 	labelMap := common.MapStr{}
 	if len(g.IncludeLabels) == 0 {
-		for k, v := range pod.Metadata.Labels {
+		for k, v := range obj.GetMetadata().Labels {
 			safemapstr.Put(labelMap, k, v)
 		}
 	} else {
-		labelMap = generateMapSubset(pod.Metadata.Labels, g.IncludeLabels)
+		labelMap = generateMapSubset(objMeta.Labels, g.IncludeLabels)
 	}
 
 	// Exclude any labels that are present in the exclude_labels config
 	for _, label := range g.ExcludeLabels {
 		delete(labelMap, label)
 	}
 
-	annotationsMap := generateMapSubset(pod.Metadata.Annotations, g.IncludeAnnotations)
-	meta := common.MapStr{
-		"pod": common.MapStr{
-			"name": pod.Metadata.GetName(),
-		},
-		"node": common.MapStr{
-			"name": pod.Spec.GetNodeName(),
-		},
-		"namespace": pod.Metadata.GetNamespace(),
-	}
-
-	// Add Pod UID metadata if enabled
-	if g.IncludePodUID {
-		safemapstr.Put(meta, "pod.uid", pod.Metadata.GetUid())
+	annotationsMap := generateMapSubset(objMeta.Annotations, g.IncludeAnnotations)
+	meta := common.MapStr{}
+	if objMeta.GetNamespace() != "" {
+		meta["namespace"] = objMeta.GetNamespace()
 	}
 
 	// Add controller metadata if present
 	if g.IncludeCreatorMetadata {
-		for _, ref := range pod.Metadata.OwnerReferences {
+		for _, ref := range objMeta.OwnerReferences {
 			if ref.GetController() {
 				switch ref.GetKind() {
 				// TODO grow this list as we keep adding more `state_*` metricsets
@@ -110,6 +114,21 @@ func (g *metaGenerator) PodMetadata(pod *Pod) common.MapStr {
 	return meta
 }
 
+// PodMetadata generates metadata for the given pod taking to account certain filters
+func (g *metaGenerator) PodMetadata(pod *Pod) common.MapStr {
+	podMeta := g.ResourceMetadata(pod)
+
+	// Add UID metadata if enabled
+	if g.IncludePodUID {
+		safemapstr.Put(podMeta, "pod.uid", pod.GetMetadata().GetUid())
+	}
+
+	safemapstr.Put(podMeta, "pod.name", pod.GetMetadata().GetName())
+	safemapstr.Put(podMeta, "node.name", pod.Spec.GetNodeName())
+
+	return podMeta
+}
+
 // Containermetadata generates metadata for the given container of a pod
 func (g *metaGenerator) ContainerMetadata(pod *Pod, container string) common.MapStr {
 	podMeta := g.PodMetadata(pod)

libbeat/common/kubernetes/metadata_test.go

@@ -28,7 +28,7 @@ import (
 )
 
 func TestPodMetadataDeDot(t *testing.T) {
-	withPodUID, _ := common.NewConfigFrom(map[string]interface{}{"include_pod_uid": true})
+	withUID, _ := common.NewConfigFrom(map[string]interface{}{"include_pod_uid": true})
 
 	UID := "005f3b90-4b9d-12f8-acf0-31020a840133"
 	Deployment := "Deployment"
@@ -44,17 +44,18 @@ func TestPodMetadataDeDot(t *testing.T) {
 		{
 			pod: &Pod{
 				Metadata: &metav1.ObjectMeta{
-					Labels: map[string]string{"a.key": "foo", "a": "bar"},
-					Uid:    &UID,
+					Labels:    map[string]string{"a.key": "foo", "a": "bar"},
+					Uid:       &UID,
+					Namespace: &test,
 				},
 				Spec: &v1.PodSpec{
 					NodeName: &test,
 				},
 			},
 			meta: common.MapStr{
 				"pod":       common.MapStr{"name": ""},
-				"namespace": "",
 				"node":      common.MapStr{"name": "test"},
+				"namespace": "test",
 				"labels":    common.MapStr{"a": common.MapStr{"value": "bar", "key": "foo"}},
 			},
 			config: common.NewConfig(),
@@ -70,12 +71,14 @@ func TestPodMetadataDeDot(t *testing.T) {
 				},
 			},
 			meta: common.MapStr{
-				"pod":       common.MapStr{"name": "", "uid": "005f3b90-4b9d-12f8-acf0-31020a840133"},
-				"namespace": "",
-				"node":      common.MapStr{"name": "test"},
-				"labels":    common.MapStr{"a": common.MapStr{"value": "bar", "key": "foo"}},
+				"pod": common.MapStr{
+					"name": "",
+					"uid":  "005f3b90-4b9d-12f8-acf0-31020a840133",
+				},
+				"node":   common.MapStr{"name": "test"},
+				"labels": common.MapStr{"a": common.MapStr{"value": "bar", "key": "foo"}},
 			},
-			config: withPodUID,
+			config: withUID,
 		},
 		{
 			pod: &Pod{
@@ -101,7 +104,6 @@ func TestPodMetadataDeDot(t *testing.T) {
 			},
 			meta: common.MapStr{
 				"pod":        common.MapStr{"name": ""},
-				"namespace":  "",
 				"node":       common.MapStr{"name": "test"},
 				"labels":     common.MapStr{"a": common.MapStr{"value": "bar", "key": "foo"}},
 				"deployment": common.MapStr{"name": "test"},

libbeat/common/kubernetes/types.go

@@ -22,7 +22,9 @@ import (
 	"time"
 
 	"github.com/ericchiang/k8s"
+	appsv1 "github.com/ericchiang/k8s/apis/apps/v1beta1"
 	"github.com/ericchiang/k8s/apis/core/v1"
+	extv1 "github.com/ericchiang/k8s/apis/extensions/v1beta1"
 	metav1 "github.com/ericchiang/k8s/apis/meta/v1"
 )
 
@@ -46,6 +48,9 @@ type PodSpec = v1.PodSpec
 // PodStatus data
 type PodStatus = v1.PodStatus
 
+// Node data
+type Node = v1.Node
+
 // Container data
 type Container = v1.Container
 
@@ -58,6 +63,15 @@ type Event = v1.Event
 // PodContainerStatus data
 type PodContainerStatus = v1.ContainerStatus
 
+// Deployment data
+type Deployment = appsv1.Deployment
+
+// ReplicaSet data
+type ReplicaSet = extv1.ReplicaSet
+
+// StatefulSet data
+type StatefulSet = appsv1.StatefulSet
+
 // Time extracts time from k8s.Time type
 func Time(t *metav1.Time) time.Time {
 	return time.Unix(t.GetSeconds(), int64(t.GetNanos()))

libbeat/common/kubernetes/watcher.go

@@ -24,7 +24,9 @@ import (
 	"time"
 
 	"github.com/ericchiang/k8s"
+	appsv1 "github.com/ericchiang/k8s/apis/apps/v1beta1"
 	"github.com/ericchiang/k8s/apis/core/v1"
+	extv1 "github.com/ericchiang/k8s/apis/extensions/v1beta1"
 
 	"github.com/elastic/beats/libbeat/logp"
 )
@@ -109,6 +111,50 @@ func NewWatcher(client *k8s.Client, resource Resource, options WatchOptions) (Wa
 			}
 			return rs
 		}
+	case *Node:
+		list := &v1.NodeList{}
+		w.resourceList = list
+		w.k8sResourceFactory = func() k8s.Resource { return &v1.Node{} }
+		w.items = func() []k8s.Resource {
+			rs := make([]k8s.Resource, 0, len(list.Items))
+			for _, item := range list.Items {
+				rs = append(rs, item)
+			}
+			return rs
+		}
+	case *Deployment:
+		list := &appsv1.DeploymentList{}
+		w.resourceList = list
+		w.k8sResourceFactory = func() k8s.Resource { return &appsv1.Deployment{} }
+		w.items = func() []k8s.Resource {
+			rs := make([]k8s.Resource, 0, len(list.Items))
+			for _, item := range list.Items {
+				rs = append(rs, item)
+			}
+			return rs
+		}
+	case *ReplicaSet:
+		list := &extv1.ReplicaSetList{}
+		w.resourceList = list
+		w.k8sResourceFactory = func() k8s.Resource { return &extv1.ReplicaSet{} }
+		w.items = func() []k8s.Resource {
+			rs := make([]k8s.Resource, 0, len(list.Items))
+			for _, item := range list.Items {
+				rs = append(rs, item)
+			}
+			return rs
+		}
+	case *StatefulSet:
+		list := &appsv1.StatefulSetList{}
+		w.resourceList = list
+		w.k8sResourceFactory = func() k8s.Resource { return &appsv1.StatefulSet{} }
+		w.items = func() []k8s.Resource {
+			rs := make([]k8s.Resource, 0, len(list.Items))
+			for _, item := range list.Items {
+				rs = append(rs, item)
+			}
+			return rs
+		}
 	default:
 		return nil, fmt.Errorf("unsupported resource type for watching %T", resource)
 	}
@@ -163,7 +209,6 @@ func (w *watcher) onDelete(obj Resource) {
 
 // Start watching pods
 func (w *watcher) Start() error {
-
 	// Make sure that events don't flow into the annotator before informer is fully set up
 	// Sync initial state:
 	err := w.sync()

libbeat/processors/add_kubernetes_metadata/_meta/fields.yml

@@ -16,7 +16,7 @@
         - name: pod.uid
           type: keyword
           description: >
-            Kubernetes pod uid
+            Kubernetes Pod UID
 
         - name: namespace
           type: keyword

metricbeat/docs/fields.asciidoc

@@ -6945,7 +6945,7 @@ Kubernetes pod name
 --
 type: keyword
 
-Kubernetes pod uid
+Kubernetes Pod UID
 
 
 --