[8.0](backport #29048) Skip config check in autodiscover for duplicated configurations

CHANGELOG.next.asciidoc

@@ -153,6 +153,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix discovery of Nomad allocations with multiple events during startup. {pull}28700[28700]
 - Fix `fingerprint` processor to give it access to the `@timestamp` field. {issue}28683[28683]
 - Fix the wrong beat name on monitoring and state endpoint {issue}27755[27755]
+- Skip configuration checks in autodiscover for configurations that are already running {pull}29048[29048]
 
 *Auditbeat*
 

libbeat/autodiscover/autodiscover.go

@@ -64,6 +64,10 @@ type Autodiscover struct {
 	meta            *meta.Map
 	listener        bus.Listener
 	logger          *logp.Logger
+
+	// workDone is a channel used for testing purpouses, to know when the worker has
+	// done some work.
+	workDone chan struct{}
 }
 
 // NewAutodiscover instantiates and returns a new Autodiscover manager
@@ -165,6 +169,11 @@ func (a *Autodiscover) worker() {
 			// reset updated status
 			updated = false
 		}
+
+		// For testing purpouses.
+		if a.workDone != nil {
+			a.workDone <- struct{}{}
+		}
 	}
 }
 
@@ -207,26 +216,30 @@ func (a *Autodiscover) handleStart(event bus.Event) bool {
 			continue
 		}
 
-		err = a.factory.CheckConfig(config)
-		if err != nil {
-			a.logger.Error(errors.Wrap(err, fmt.Sprintf(
-				"Auto discover config check failed for config '%s', won't start runner",
-				common.DebugString(config, true))))
-			continue
-		}
-
 		// Update meta no matter what
 		dynFields := a.meta.Store(hash, meta)
 
+		if _, ok := newCfg[hash]; ok {
+			a.logger.Debugf("Config %v duplicated in start event", common.DebugString(config, true))
+			continue
+		}
+
 		if cfg, ok := a.configs[eventID][hash]; ok {
 			a.logger.Debugf("Config %v is already running", common.DebugString(config, true))
 			newCfg[hash] = cfg
 			continue
-		} else {
-			newCfg[hash] = &reload.ConfigWithMeta{
-				Config: config,
-				Meta:   &dynFields,
-			}
+		}
+
+		err = a.factory.CheckConfig(config)
+		if err != nil {
+			a.logger.Error(errors.Wrap(err, fmt.Sprintf(
+				"Auto discover config check failed for config '%s', won't start runner",
+				common.DebugString(config, true))))
+			continue
+		}
+		newCfg[hash] = &reload.ConfigWithMeta{
+			Config: config,
+			Meta:   &dynFields,
 		}
 
 		updated = true

libbeat/autodiscover/autodiscover_test.go

@@ -75,6 +75,8 @@ type mockAdapter struct {
 	mutex   sync.Mutex
 	configs []*common.Config
 	runners []*mockRunner
+
+	CheckConfigCallCount int
 }
 
 // CreateConfig generates a valid list of configs from the given event, the received event will have all keys defined by `StartFilter`
@@ -87,6 +89,8 @@ func (m *mockAdapter) CreateConfig(event bus.Event) ([]*common.Config, error) {
 
 // CheckConfig tests given config to check if it will work or not, returns errors in case it won't work
 func (m *mockAdapter) CheckConfig(c *common.Config) error {
+	m.CheckConfigCallCount++
+
 	config := struct {
 		Broken bool `config:"broken"`
 	}{}
@@ -324,6 +328,66 @@ func TestAutodiscoverHash(t *testing.T) {
 	assert.False(t, runners[1].stopped)
 }
 
+func TestAutodiscoverDuplicatedConfigConfigCheckCalledOnce(t *testing.T) {
+	goroutines := resources.NewGoroutinesChecker()
+	defer goroutines.Check(t)
+
+	// Register mock autodiscover provider
+	busChan := make(chan bus.Bus, 1)
+
+	Registry = NewRegistry()
+	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
+
+		return &mockProvider{}, nil
+	})
+
+	// Create a mock adapter that returns a duplicated config
+	runnerConfig, _ := common.NewConfigFrom(map[string]string{
+		"id": "foo",
+	})
+	adapter := mockAdapter{
+		configs: []*common.Config{runnerConfig, runnerConfig},
+	}
+
+	// and settings:
+	providerConfig, _ := common.NewConfigFrom(map[string]string{
+		"type": "mock",
+	})
+	config := Config{
+		Providers: []*common.Config{providerConfig},
+	}
+	k, _ := keystore.NewFileKeystore("test")
+	// Create autodiscover manager
+	autodiscover, err := NewAutodiscover("test", nil, &adapter, &adapter, &config, k)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	autodiscover.workDone = make(chan struct{})
+
+	// Start it
+	autodiscover.Start()
+	defer autodiscover.Stop()
+	eventBus := <-busChan
+
+	// Publish a couple of events.
+	for i := 0; i < 2; i++ {
+		eventBus.Publish(bus.Event{
+			"id":       "foo",
+			"provider": "mock",
+			"start":    true,
+			"meta": common.MapStr{
+				"foo": "bar",
+			},
+		})
+		<-autodiscover.workDone
+		assert.Equal(t, 1, len(adapter.Runners()), "Only one runner should be started")
+		assert.Equal(t, 1, adapter.CheckConfigCallCount, "Check config should have been called only once")
+	}
+}
+
 func TestAutodiscoverWithConfigCheckFailures(t *testing.T) {
 	goroutines := resources.NewGoroutinesChecker()
 	defer goroutines.Check(t)