filebeat: make deep copy before notifying of config change

[orestisfl]

Proposed commit message

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

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

How to test this PR locally

This is a blind spot: We have reproduced this on agentless deployments using abusech (see issue). However, I couldn't find a way to reproduce in an integration or unit test.

The closest is this hacky test:

diff --git a/libbeat/cmd/instance/beat_test.go b/libbeat/cmd/instance/beat_test.go
index ebfecf191c..8efea8eadd 100644
--- a/libbeat/cmd/instance/beat_test.go
+++ b/libbeat/cmd/instance/beat_test.go
@@ -15,14 +15,14 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//go:build !integration
-
 package instance
 
 import (
 	"bytes"
+	es "github.com/elastic/beats/v7/libbeat/statestore/backend/es"
 	"io/ioutil"
 	"os"
+	"sync"
 	"testing"
 
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
@@ -262,6 +262,72 @@ elasticsearch:
 		require.Same(t, c, m.cfg.Config)
 		require.False(t, b.isConnectionToOlderVersionAllowed(), "allow_older_versions flag should now be set to false")
 	})
+
+	t.Run("test tmp", func(t *testing.T) {
+		b, err := NewBeat("testbeat", "testidx", "0.9", false, nil)
+		require.NoError(t, err)
+
+		cfg := `
+elasticsearch:
+  hosts: ["https://127.0.0.1:9200"]
+  username: "elastic"
+  allow_older_versions: false
+`
+		c, err := config.NewConfigWithYAML([]byte(cfg), cfg)
+		require.NoError(t, err)
+		outCfg, err := c.Child("elasticsearch", -1)
+		require.NoError(t, err)
+
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		const size = 100
+		wg.Add(size)
+		notifiers := make([]*es.Notifier, size)
+		for i := 0; i < size; i++ {
+			n := es.NewNotifier()
+			notifiers[i] = n
+			if i%2 == 0 {
+				n.Subscribe(func(c *config.C) {
+					defer wg.Done()
+					t.Log(c.FlattenedKeys())
+				})
+			} else {
+				n.Subscribe(func(c *config.C) {
+					defer wg.Done()
+					c.Merge(map[string]string{"test": "test"})
+				})
+			}
+		}
+
+		update := &reload.ConfigWithMeta{Config: c}
+		b.OutputConfigReloader = reload.ReloadableFunc(func(r *reload.ConfigWithMeta) error {
+			for i := 0; i < size; i++ {
+				outCfg := config.Namespace{}
+				if err := r.Config.Unpack(&outCfg); err != nil || outCfg.Name() != "elasticsearch" {
+					logp.Err("Failed to unpack the output config: %v", err)
+					return nil
+				}
+				c, err := config.NewConfigFrom(outCfg.Config())
+				require.NoError(t, err)
+
+				notifiers[i].Notify(c)
+			}
+			return nil
+		})
+		m := &outputReloaderMock{}
+		reloader := b.makeOutputReloader(m)
+
+		require.False(t, b.Config.Output.IsSet(), "the output should not be set yet")
+		require.True(t, b.isConnectionToOlderVersionAllowed(), "allow_older_versions flag should be true from 8.11")
+
+		err = reloader.Reload(update)
+		require.NoError(t, err)
+		require.True(t, b.Config.Output.IsSet(), "now the output should be set")
+		require.Equal(t, outCfg, b.Config.Output.Config())
+		require.Same(t, c, m.cfg.Config)
+		require.False(t, b.isConnectionToOlderVersionAllowed(), "allow_older_versions flag should now be set to false")
+	})
 }
 
 type outputReloaderMock struct {

however, that doesn't test any code to be commited

Related issues

• Fixes 42815
• Related to [filebeat] Elasticsearch state storage for httpjson and cel inputs #41446

Screenshots

Errors stop once the image with the fix is deployed

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[AndersonQ]

LGTM. There is at least a test failing, but it seems some flakyness, you might retry it. If you want, you can also open a flaky test issue, that would be the best so we can track the flakyness

Btw, the test you added in the PR description, should it trigger the issue? I tried here and it passes with and without the change. So I don't really get the idea of that test.

[orestisfl]

@AndersonQ yes, you are right, I accidentally pasted the "fixed" test. Try the one below a few times.

Just in case it's not clear, that test doesn't actually test the filebeat code, I copied the code from filebeat in a separate test to check the hypothesis that references to r.Config's values can be modified in different goroutines if the config isn't copied first.

I think it's worth it to discuss better testing approaches for this case but I'd prefer to merge this bugfix separately.

diff --git a/libbeat/cmd/instance/beat_test.go b/libbeat/cmd/instance/beat_test.go
index ebfecf191c..f9c290693e 100644
--- a/libbeat/cmd/instance/beat_test.go
+++ b/libbeat/cmd/instance/beat_test.go
@@ -15,14 +15,14 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//go:build !integration
-
 package instance
 
 import (
 	"bytes"
+	es "github.com/elastic/beats/v7/libbeat/statestore/backend/es"
 	"io/ioutil"
 	"os"
+	"sync"
 	"testing"
 
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
@@ -262,6 +262,69 @@ elasticsearch:
 		require.Same(t, c, m.cfg.Config)
 		require.False(t, b.isConnectionToOlderVersionAllowed(), "allow_older_versions flag should now be set to false")
 	})
+
+	t.Run("test tmp", func(t *testing.T) {
+		b, err := NewBeat("testbeat", "testidx", "0.9", false, nil)
+		require.NoError(t, err)
+
+		cfg := `
+elasticsearch:
+  hosts: ["https://127.0.0.1:9200"]
+  username: "elastic"
+  allow_older_versions: false
+`
+		c, err := config.NewConfigWithYAML([]byte(cfg), cfg)
+		require.NoError(t, err)
+		outCfg, err := c.Child("elasticsearch", -1)
+		require.NoError(t, err)
+
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		const size = 100
+		wg.Add(size)
+		notifiers := make([]*es.Notifier, size)
+		for i := 0; i < size; i++ {
+			n := es.NewNotifier()
+			notifiers[i] = n
+			if i%2 == 0 {
+				n.Subscribe(func(c *config.C) {
+					defer wg.Done()
+					t.Log(c.FlattenedKeys())
+				})
+			} else {
+				n.Subscribe(func(c *config.C) {
+					defer wg.Done()
+					c.Merge(map[string]string{"test": "test"})
+				})
+			}
+		}
+
+		update := &reload.ConfigWithMeta{Config: c}
+		b.OutputConfigReloader = reload.ReloadableFunc(func(r *reload.ConfigWithMeta) error {
+			for i := 0; i < size; i++ {
+				outCfg := config.Namespace{}
+				if err := r.Config.Unpack(&outCfg); err != nil || outCfg.Name() != "elasticsearch" {
+					logp.Err("Failed to unpack the output config: %v", err)
+					return nil
+				}
+				notifiers[i].Notify(outCfg.Config())
+			}
+			return nil
+		})
+		m := &outputReloaderMock{}
+		reloader := b.makeOutputReloader(m)
+
+		require.False(t, b.Config.Output.IsSet(), "the output should not be set yet")
+		require.True(t, b.isConnectionToOlderVersionAllowed(), "allow_older_versions flag should be true from 8.11")
+
+		err = reloader.Reload(update)
+		require.NoError(t, err)
+		require.True(t, b.Config.Output.IsSet(), "now the output should be set")
+		require.Equal(t, outCfg, b.Config.Output.Config())
+		require.Same(t, c, m.cfg.Config)
+		require.False(t, b.isConnectionToOlderVersionAllowed(), "allow_older_versions flag should now be set to false")
+	})
 }
 
 type outputReloaderMock struct {

[orestisfl]

LGTM. There is at least a test failing, but it seems some flakyness, you might retry it. If you want, you can also open a flaky test issue, that would be the best so we can track the flakyness

#42995