filebeat: make deep copy before notifying of config change (#42992)

This prevents concurrent read & write map access.

Unrelated, but I've escalated one log line to Info to allow for easier verifying that ES store is being used from agent logs.

Fixes #42815

(cherry picked from commit f1e42fc)

filebeat/beater/filebeat.go

@@ -328,7 +328,15 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 				return nil
 			}
 
-			stateStore.notifier.Notify(outCfg.Config())
+			// Create a new config with the output configuration. Since r.Config is a pointer, a copy is required to
+			// avoid concurrent map read and write.
+			// See https://github.com/elastic/beats/issues/42815
+			configCopy, err := conf.NewConfigFrom(outCfg.Config())
+			if err != nil {
+				logp.Err("Failed to create a new config from the output config: %v", err)
+				return nil
+			}
+			stateStore.notifier.Notify(configCopy)
 			return nil
 		})
 	}

libbeat/statestore/backend/es/store.go

@@ -296,7 +296,7 @@ func (s *store) Each(fn func(string, backend.ValueDecoder) (bool, error)) error
 }
 
 func (s *store) configure(ctx context.Context, c *conf.C) {
-	s.log.Debugf("Configure ES store")
+	s.log.Info("Configuring ES store")
 	s.mx.Lock()
 	defer s.mx.Unlock()
 

x-pack/filebeat/tests/integration/managerV2_test.go

@@ -930,7 +930,7 @@ func TestHTTPJSONInputReloadUnderElasticAgentWithElasticStateStore(t *testing.T)
 	)
 
 	for _, contains := range []string{
-		"Configure ES store",
+		"Configuring ES store",
 		"input-cursor::openStore: prefix: httpjson inputID: " + inputID,
 		"input-cursor store read 0 keys", // first, no previous data exists
 		"input-cursor store read 1 keys", // after the restart, previous key is read