Add leader election for autodiscover (#20281) (#20510)

(cherry picked from commit 9ab9b97)

CHANGELOG.next.asciidoc

@@ -434,6 +434,7 @@ field. You can revert this change by configuring tags for the module and omittin
 - Add support to trim captured values in the dissect processor. {pull}19464[19464]
 - Added the `max_cached_sessions` option to the script processor. {pull}19562[19562]
 - Set index.max_docvalue_fields_search in index template to increase value to 200 fields. {issue}20215[20215]
+- Add leader election for Kubernetes autodiscover. {pull}20281[20281]
 - Add capability of enriching process metadata with contianer id also for non-privileged containers in `add_process_metadata` processor. {pull}19767[19767]
 
 *Auditbeat*

deploy/kubernetes/metricbeat-kubernetes.yaml

@@ -366,6 +366,12 @@ rules:
   - "/metrics"
   verbs:
   - get
+- apiGroups:
+    - coordination.k8s.io
+  resources:
+    - leases
+  verbs:
+    - '*'
 ---
 apiVersion: v1
 kind: ServiceAccount

deploy/kubernetes/metricbeat/metricbeat-role.yaml

@@ -32,3 +32,9 @@ rules:
   - "/metrics"
   verbs:
   - get
+- apiGroups:
+    - coordination.k8s.io
+  resources:
+    - leases
+  verbs:
+    - '*'

go.mod

@@ -166,7 +166,7 @@ require (
 	golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae
 	golang.org/x/text v0.3.2
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0
-	golang.org/x/tools v0.0.0-20200701041122-1837592efa10
+	golang.org/x/tools v0.0.0-20200806022845-90696ccdc692
 	google.golang.org/api v0.15.0
 	google.golang.org/genproto v0.0.0-20191230161307-f3c370f40bfb
 	google.golang.org/grpc v1.29.1

libbeat/autodiscover/autodiscover.go

@@ -83,7 +83,7 @@ func NewAutodiscover(
 	// Init providers
 	var providers []Provider
 	for _, providerCfg := range config.Providers {
-		provider, err := Registry.BuildProvider(bus, providerCfg, keystore)
+		provider, err := Registry.BuildProvider(name, bus, providerCfg, keystore)
 		if err != nil {
 			return nil, errors.Wrap(err, "error in autodiscover provider settings")
 		}

libbeat/autodiscover/autodiscover_test.go

@@ -140,7 +140,7 @@ func TestAutodiscover(t *testing.T) {
 	// Register mock autodiscover provider
 	busChan := make(chan bus.Bus, 1)
 	Registry = NewRegistry()
-	Registry.AddProvider("mock", func(b bus.Bus, uuid uuid.UUID, c *common.Config, k keystore.Keystore) (Provider, error) {
+	Registry.AddProvider("mock", func(beatName string, b bus.Bus, uuid uuid.UUID, c *common.Config, k keystore.Keystore) (Provider, error) {
 		// intercept bus to mock events
 		busChan <- b
 
@@ -259,7 +259,7 @@ func TestAutodiscoverHash(t *testing.T) {
 	busChan := make(chan bus.Bus, 1)
 
 	Registry = NewRegistry()
-	Registry.AddProvider("mock", func(b bus.Bus, uuid uuid.UUID, c *common.Config, k keystore.Keystore) (Provider, error) {
+	Registry.AddProvider("mock", func(beatName string, b bus.Bus, uuid uuid.UUID, c *common.Config, k keystore.Keystore) (Provider, error) {
 		// intercept bus to mock events
 		busChan <- b
 
@@ -323,7 +323,7 @@ func TestAutodiscoverWithConfigCheckFailures(t *testing.T) {
 	// Register mock autodiscover provider
 	busChan := make(chan bus.Bus, 1)
 	Registry = NewRegistry()
-	Registry.AddProvider("mock", func(b bus.Bus, uuid uuid.UUID, c *common.Config, k keystore.Keystore) (Provider, error) {
+	Registry.AddProvider("mock", func(beatName string, b bus.Bus, uuid uuid.UUID, c *common.Config, k keystore.Keystore) (Provider, error) {
 		// intercept bus to mock events
 		busChan <- b
 

libbeat/autodiscover/provider.go

@@ -35,7 +35,7 @@ type Provider interface {
 }
 
 // ProviderBuilder creates a new provider based on the given config and returns it
-type ProviderBuilder func(bus.Bus, uuid.UUID, *common.Config, keystore.Keystore) (Provider, error)
+type ProviderBuilder func(string, bus.Bus, uuid.UUID, *common.Config, keystore.Keystore) (Provider, error)
 
 // AddProvider registers a new ProviderBuilder
 func (r *registry) AddProvider(name string, provider ProviderBuilder) error {
@@ -70,7 +70,7 @@ func (r *registry) GetProvider(name string) ProviderBuilder {
 }
 
 // BuildProvider reads provider configuration and instantiate one
-func (r *registry) BuildProvider(bus bus.Bus, c *common.Config, keystore keystore.Keystore) (Provider, error) {
+func (r *registry) BuildProvider(beatName string, bus bus.Bus, c *common.Config, keystore keystore.Keystore) (Provider, error) {
 	var config ProviderConfig
 	err := c.Unpack(&config)
 	if err != nil {
@@ -87,5 +87,5 @@ func (r *registry) BuildProvider(bus bus.Bus, c *common.Config, keystore keystor
 		return nil, err
 	}
 
-	return builder(bus, uuid, c, keystore)
+	return builder(beatName, bus, uuid, c, keystore)
 }

libbeat/autodiscover/providers/docker/docker.go

@@ -59,7 +59,13 @@ type Provider struct {
 }
 
 // AutodiscoverBuilder builds and returns an autodiscover provider
-func AutodiscoverBuilder(bus bus.Bus, uuid uuid.UUID, c *common.Config, keystore keystore.Keystore) (autodiscover.Provider, error) {
+func AutodiscoverBuilder(
+	beatName string,
+	bus bus.Bus,
+	uuid uuid.UUID,
+	c *common.Config,
+	keystore keystore.Keystore,
+) (autodiscover.Provider, error) {
 	logger := logp.NewLogger("docker")
 
 	errWrap := func(err error) error {

libbeat/autodiscover/providers/docker/docker_integration_test.go

@@ -56,7 +56,7 @@ func TestDockerStart(t *testing.T) {
 	s := &template.MapperSettings{nil, nil}
 	config.Templates = *s
 	k, _ := keystore.NewFileKeystore("test")
-	provider, err := AutodiscoverBuilder(bus, UUID, common.MustNewConfigFrom(config), k)
+	provider, err := AutodiscoverBuilder("mockBeat", bus, UUID, common.MustNewConfigFrom(config), k)
 	if err != nil {
 		t.Fatal(err)
 	}

libbeat/autodiscover/providers/jolokia/jolokia.go

@@ -53,7 +53,13 @@ type Provider struct {
 
 // AutodiscoverBuilder builds a Jolokia Discovery autodiscover provider, it fails if
 // there is some problem with the configuration
-func AutodiscoverBuilder(bus bus.Bus, uuid uuid.UUID, c *common.Config, keystore keystore.Keystore) (autodiscover.Provider, error) {
+func AutodiscoverBuilder(
+	beatName string,
+	bus bus.Bus,
+	uuid uuid.UUID,
+	c *common.Config,
+	keystore keystore.Keystore,
+) (autodiscover.Provider, error) {
 	errWrap := func(err error) error {
 		return errors.Wrap(err, "error setting up jolokia autodiscover provider")
 	}

libbeat/autodiscover/providers/kubernetes/config.go

@@ -44,6 +44,9 @@ type Config struct {
 	// Scope can be either node or cluster.
 	Scope    string `config:"scope"`
 	Resource string `config:"resource"`
+	// Unique identifies if this provider enables its templates only when it is elected as leader in a k8s cluster
+	Unique      bool   `config:"unique"`
+	LeaderLease string `config:"leader_lease"`
 
 	Prefix    string                  `config:"prefix"`
 	Hints     *common.Config          `config:"hints"`
@@ -60,6 +63,7 @@ func defaultConfig() *Config {
 		Resource:       "pod",
 		CleanupTimeout: 60 * time.Second,
 		Prefix:         "co.elastic",
+		Unique:         false,
 	}
 }
 
@@ -98,6 +102,9 @@ func (c *Config) Validate() error {
 	if c.Scope != "node" && c.Scope != "cluster" {
 		return fmt.Errorf("invalid `scope` configured. supported values are `node` and `cluster`")
 	}
+	if c.Unique && c.Scope != "cluster" {
+		logp.L().Warnf("can only set `unique` when scope is `cluster`")
+	}
 
 	return nil
 }