Cherry-pick #18564 to 7.8: [Autodiscover] Check if runner is already running before starting again

CHANGELOG.next.asciidoc

@@ -112,6 +112,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix panic when assigning a key to a `nil` value in an event. {pull}18143[18143]
 - Change `decode_json_fields` processor, to merge parsed json objects with existing objects in the event instead of fully replacing them. {pull}17958[17958]
 - Gives monitoring reporter hosts, if configured, total precedence over corresponding output hosts. {issue}17937[17937] {pull}17991[17991]
+- [Autodiscover] Check if runner is already running before starting again. {pull}18564[18564]
 
 *Auditbeat*
 

libbeat/cfgfile/list.go

@@ -70,7 +70,7 @@ func (r *RunnerList) Reload(configs []*reload.ConfigWithMeta) error {
 			continue
 		}
 
-		if _, ok := stopList[hash]; ok {
+		if _, ok := r.runners[hash]; ok {
 			delete(stopList, hash)
 		} else {
 			startList[hash] = config

libbeat/cfgfile/list_test.go

@@ -103,6 +103,28 @@ func TestReloadSameConfigs(t *testing.T) {
 	assert.Equal(t, state, list.copyRunnerList())
 }
 
+func TestReloadDuplicateConfig(t *testing.T) {
+	factory := &runnerFactory{}
+	list := NewRunnerList("", factory, nil)
+
+	list.Reload([]*reload.ConfigWithMeta{
+		createConfig(1),
+	})
+
+	state := list.copyRunnerList()
+	assert.Equal(t, len(state), 1)
+
+	// This can happen in Autodiscover when a container if getting restarted
+	// but the previous one is not cleaned yet.
+	list.Reload([]*reload.ConfigWithMeta{
+		createConfig(1),
+		createConfig(1),
+	})
+
+	// nothing changed
+	assert.Equal(t, state, list.copyRunnerList())
+}
+
 func TestReloadStopConfigs(t *testing.T) {
 	factory := &runnerFactory{}
 	list := NewRunnerList("", factory, nil)