Refactor kubernetes autodiscover to enable different resource based discovery

CHANGELOG.next.asciidoc

@@ -357,6 +357,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add support for API keys in Elasticsearch outputs. {pull}14324[14324]
 - Ensure that init containers are no longer tailed after they stop {pull}14394[14394]
 - Add consumer_lag in Kafka consumergroup metricset {pull}14822[14822]
+- Refactor kubernetes autodiscover to enable different resource based discovery {pull}14738[14738]
 
 *Auditbeat*
 

deploy/kubernetes/filebeat-kubernetes.yaml

@@ -23,7 +23,7 @@ data:
     #filebeat.autodiscover:
     #  providers:
     #    - type: kubernetes
-    #      host: ${NODE_NAME}
+    #      node: ${NODE_NAME}
     #      hints.enabled: true
     #      hints.default_config:
     #        type: container

deploy/kubernetes/filebeat/filebeat-configmap.yaml

@@ -23,7 +23,7 @@ data:
     #filebeat.autodiscover:
     #  providers:
     #    - type: kubernetes
-    #      host: ${NODE_NAME}
+    #      node: ${NODE_NAME}
     #      hints.enabled: true
     #      hints.default_config:
     #        type: container

deploy/kubernetes/metricbeat-kubernetes.yaml

@@ -18,7 +18,7 @@ data:
     #metricbeat.autodiscover:
     #  providers:
     #    - type: kubernetes
-    #      host: ${NODE_NAME}
+    #      node: ${NODE_NAME}
     #      hints.enabled: true
 
     processors:

deploy/kubernetes/metricbeat/metricbeat-daemonset-configmap.yaml

@@ -18,7 +18,7 @@ data:
     #metricbeat.autodiscover:
     #  providers:
     #    - type: kubernetes
-    #      host: ${NODE_NAME}
+    #      node: ${NODE_NAME}
     #      hints.enabled: true
 
     processors:

libbeat/autodiscover/providers/kubernetes/config.go

@@ -25,16 +25,24 @@ import (
 
 	"github.com/elastic/beats/libbeat/autodiscover/template"
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	"github.com/elastic/beats/libbeat/logp"
 )
 
 // Config for kubernetes autodiscover provider
 type Config struct {
 	KubeConfig     string        `config:"kube_config"`
-	Host           string        `config:"host"`
 	Namespace      string        `config:"namespace"`
 	SyncPeriod     time.Duration `config:"sync_period"`
 	CleanupTimeout time.Duration `config:"cleanup_timeout" validate:"positive"`
 
+	// Needed when resource is a pod
+	HostDeprecated string `config:"host"`
+	Node           string `config:"node"`
+	// Scope can be either node or cluster.
+	Scope    string `config:"scope"`
+	Resource string `config:"resource"`
+
 	Prefix    string                  `config:"prefix"`
 	Hints     *common.Config          `config:"hints"`
 	Builders  []*common.Config        `config:"builders"`
@@ -45,6 +53,7 @@ type Config struct {
 func defaultConfig() *Config {
 	return &Config{
 		SyncPeriod:     10 * time.Minute,
+		Resource:       "pod",
 		CleanupTimeout: 60 * time.Second,
 		Prefix:         "co.elastic",
 	}
@@ -61,5 +70,30 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("no configs or hints defined for autodiscover provider")
 	}
 
+	// Check if host is being defined and change it to node instead.
+	if c.Node == "" && c.HostDeprecated != "" {
+		c.Node = c.HostDeprecated
+		cfgwarn.Deprecate("8.0", "`host` will be deprecated, use `node` instead")
+	}
+
+	// Check if resource is either node or pod. If yes then default the scope to "node" if not provided.
+	// Default the scope to "cluster" for everything else.
+	switch c.Resource {
+	case "node", "pod":
+		if c.Scope == "" {
+			c.Scope = "node"
+		}
+
+	default:
+		if c.Scope == "node" {
+			logp.L().Warnf("can not set scope to `node` when using resource %s. resetting scope to `cluster`", c.Resource)
+		}
+		c.Scope = "cluster"
+	}
+
+	if c.Scope != "node" && c.Scope != "cluster" {
+		return fmt.Errorf("invalid `scope` configured. supported values are `node` and `cluster`")
+	}
+
 	return nil
 }

libbeat/autodiscover/providers/kubernetes/config_test.go

@@ -53,6 +53,22 @@ func TestConfigWithCustomBuilders(t *testing.T) {
 	assert.NotNil(t, err)
 }
 
+func TestConfigWithIncorrectScope(t *testing.T) {
+	cfg := common.MapStr{
+		"scope":         "node",
+		"resource":      "service",
+		"hints.enabled": true,
+	}
+
+	config := common.MustNewConfigFrom(&cfg)
+	c := defaultConfig()
+	err := config.Unpack(&c)
+	assert.Nil(t, err)
+
+	assert.Equal(t, "service", c.Resource)
+	assert.Equal(t, "cluster", c.Scope)
+}
+
 type mockBuilder struct {
 }