Add k8s metadata autodetection

CHANGELOG.next.asciidoc

@@ -248,6 +248,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add support for RFC3339 time zone offsets in JSON output. {pull}13227[13227]
 - Add autodetection mode for add_docker_metadata and enable it by default in included configuration files{pull}13374[13374]
 - Added `monitoring.cluster_uuid` setting to associate Beat data with specified ES cluster in Stack Monitoring UI. {pull}13182[13182]
+- Add autodetection mode for add_kubernetes_metadata and enable it by default in included configuration files{pull}13473[13473]
+
 
 *Auditbeat*
 

auditbeat/auditbeat.yml

@@ -149,6 +149,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

filebeat/filebeat.yml

@@ -177,6 +177,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

journalbeat/journalbeat.yml

@@ -146,6 +146,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

libbeat/_meta/config.yml.tmpl

@@ -94,6 +94,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 {{else}}
 processors:
   - add_observer_metadata:

libbeat/common/kubernetes/util.go

@@ -36,6 +36,7 @@ const defaultNode = "localhost"
 // in cluster configuration based on the secrets mounted in the Pod. If kubeConfig is passed,
 // it parses the config file to get the config required to build a client.
 func GetKubernetesClient(kubeconfig string) (kubernetes.Interface, error) {
+
 	cfg, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
 	if err != nil {
 		return nil, fmt.Errorf("unable to build kube config due to error: %+v", err)

libbeat/processors/add_kubernetes_metadata/kubernetes.go

@@ -26,17 +26,20 @@ import (
 	"github.com/elastic/beats/libbeat/common/kubernetes"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/processors"
+
+	k8s "k8s.io/client-go/kubernetes"
 )
 
 const (
 	timeout = time.Second * 5
 )
 
 type kubernetesAnnotator struct {
-	watcher  kubernetes.Watcher
-	indexers *Indexers
-	matchers *Matchers
-	cache    *cache
+	watcher             kubernetes.Watcher
+	indexers            *Indexers
+	matchers            *Matchers
+	cache               *cache
+	kubernetesAvailable bool
 }
 
 func init() {
@@ -51,6 +54,16 @@ func init() {
 	Indexing.AddMatcher(FieldFormatMatcherName, NewFieldFormatMatcher)
 }
 
+func isKubernetesAvailable(client k8s.Interface) bool {
+	server, err := client.Discovery().ServerVersion()
+	if err != nil {
+		logp.Info("%v: could not detect kubernetes env: %v", "add_kubernetes_metadata", err)
+		return false
+	}
+	logp.Info("%v: kubernetes env detected, with version: %v", "add_kubernetes_metadata", server)
+	return true
+}
+
 // New constructs a new add_kubernetes_metadata processor.
 func New(cfg *common.Config) (processors.Processor, error) {
 	config := defaultKubernetesAnnotatorConfig()
@@ -91,9 +104,26 @@ func New(cfg *common.Config) (processors.Processor, error) {
 		return nil, fmt.Errorf("Can not initialize kubernetes plugin with zero matcher plugins")
 	}
 
+	processor := &kubernetesAnnotator{
+		indexers:            indexers,
+		matchers:            matchers,
+		cache:               newCache(config.CleanupTimeout),
+		kubernetesAvailable: false,
+	}
+
 	client, err := kubernetes.GetKubernetesClient(config.KubeConfig)
 	if err != nil {
-		return nil, err
+		if kubernetes.IsInCluster(config.KubeConfig) {
+			logp.Err("%v: could not create kubernetes client using in_cluster config", "add_kubernetes_metadata")
+		} else {
+			logp.Err("%v: could not create kubernetes client using config: %v", "add_kubernetes_metadata", config.KubeConfig)
+		}
+		processor.kubernetesAvailable = false
+		return processor, nil
+	}
+
+	if !isKubernetesAvailable(client) {
+		return processor, nil
 	}
 
 	config.Host = kubernetes.DiscoverKubernetesNode(config.Host, kubernetes.IsInCluster(config.KubeConfig), client)
@@ -111,12 +141,8 @@ func New(cfg *common.Config) (processors.Processor, error) {
 		return nil, err
 	}
 
-	processor := &kubernetesAnnotator{
-		watcher:  watcher,
-		indexers: indexers,
-		matchers: matchers,
-		cache:    newCache(config.CleanupTimeout),
-	}
+	processor.watcher = watcher
+	processor.kubernetesAvailable = true
 
 	watcher.AddEventHandler(kubernetes.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
@@ -139,6 +165,9 @@ func New(cfg *common.Config) (processors.Processor, error) {
 }
 
 func (k *kubernetesAnnotator) Run(event *beat.Event) (*beat.Event, error) {
+	if !k.kubernetesAvailable {
+		return event, nil
+	}
 	index := k.matchers.MetadataIndex(event.Fields)
 	if index == "" {
 		return event, nil

libbeat/processors/add_kubernetes_metadata/kubernetes_test.go

@@ -42,6 +42,7 @@ func TestAnnotatorDeepUpdate(t *testing.T) {
 		matchers: &Matchers{
 			matchers: []Matcher{matcher},
 		},
+		kubernetesAvailable: true,
 	}
 
 	processor.cache.set("foo", common.MapStr{
@@ -86,3 +87,42 @@ func TestAnnotatorDeepUpdate(t *testing.T) {
 		},
 	}, event.Fields)
 }
+
+// Test metadata are not included in the event
+func TestAnnotatorWithNoKubernetesAvailable(t *testing.T) {
+	cfg := common.MustNewConfigFrom(map[string]interface{}{
+		"lookup_fields": []string{"kubernetes.pod.name"},
+	})
+	matcher, err := NewFieldMatcher(*cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	processor := kubernetesAnnotator{
+		cache: newCache(10 * time.Second),
+		matchers: &Matchers{
+			matchers: []Matcher{matcher},
+		},
+		kubernetesAvailable: false,
+	}
+
+	intialEventMap := common.MapStr{
+		"kubernetes": common.MapStr{
+			"pod": common.MapStr{
+				"name": "foo",
+				"id":   "pod_id",
+				"metrics": common.MapStr{
+					"a": 1,
+					"b": 2,
+				},
+			},
+		},
+	}
+
+	event, err := processor.Run(&beat.Event{
+		Fields: intialEventMap.Clone(),
+	})
+	assert.NoError(t, err)
+
+	assert.Equal(t, intialEventMap, event.Fields)
+}

metricbeat/metricbeat.yml

@@ -121,6 +121,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

packetbeat/packetbeat.yml

@@ -203,6 +203,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

winlogbeat/winlogbeat.yml

@@ -128,6 +128,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

x-pack/auditbeat/auditbeat.yml

@@ -171,6 +171,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

x-pack/filebeat/filebeat.yml

@@ -177,6 +177,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

x-pack/functionbeat/functionbeat.yml

@@ -345,6 +345,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

x-pack/metricbeat/metricbeat.yml

@@ -121,6 +121,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================
 

x-pack/winlogbeat/winlogbeat.yml

@@ -140,6 +140,7 @@ processors:
   - add_host_metadata: ~
   - add_cloud_metadata: ~
   - add_docker_metadata: ~
+  - add_kubernetes_metadata: ~
 
 #================================ Logging =====================================