Update allow_older_versions when running under Elastic Agent

CHANGELOG.next.asciidoc

@@ -60,6 +60,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - Fix concurrent map writes when system/process code called from reporter code {pull}32491[32491]
 - Fix panics when a processor is closed twice {pull}34647[34647]
 - Update elastic-agent-system-metrics to v0.4.6 to allow builds on mips platforms. {pull}34674[34674]
+- Fix Beats started by agent do not respect the allow_older_versions: true configuration flag {issue}34227[34227] {pull}34964[34964]
 
 *Auditbeat*
 

libbeat/cmd/instance/beat.go

@@ -102,6 +102,12 @@ type Beat struct {
 
 	// shouldReexec is a flag to indicate the Beat should restart
 	shouldReexec bool
+
+	// allowOlderESVersions is used in a global Elasticsearch connection callback
+	// this is a pre-cached value for the callback, so we don't have to parse the
+	// config on each callback.
+	// This value must be updated every time we update/reload the elasticsearch output configuration.
+	allowOlderESVersions bool
 }
 
 type beatConfig struct {
@@ -336,7 +342,12 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 	logSystemInfo(b.Info)
 	logp.Info("Setup Beat: %s; Version: %s", b.Info.Beat, b.Info.Version)
 
-	b.checkElasticsearchVersion()
+	b.allowOlderESVersions = b.isConnectionToOlderVersionAllowed()
+
+	err = b.registerESVersionCheckCallback()
+	if err != nil {
+		return nil, err
+	}
 
 	err = b.registerESIndexManagement()
 	if err != nil {
@@ -972,15 +983,15 @@ func (b *Beat) loadDashboards(ctx context.Context, force bool) error {
 	return nil
 }
 
-// checkElasticsearchVersion registers a global callback to make sure ES instance we are connecting
+// registerESVersionCheckCallback registers a global callback to make sure ES instance we are connecting
 // to is at least on the same version as the Beat.
 // If the check is disabled or the output is not Elasticsearch, nothing happens.
-func (b *Beat) checkElasticsearchVersion() {
-	if b.isConnectionToOlderVersionAllowed() {
-		return
-	}
+func (b *Beat) registerESVersionCheckCallback() error {
+	_, err := elasticsearch.RegisterGlobalCallback(func(conn *eslegclient.Connection) error {
+		if b.allowOlderESVersions {
+			return nil
+		}
 
-	_, _ = elasticsearch.RegisterGlobalCallback(func(conn *eslegclient.Connection) error {
 		esVersion := conn.GetVersion()
 		beatVersion, err := libversion.New(b.Info.Version)
 		if err != nil {
@@ -991,6 +1002,8 @@ func (b *Beat) checkElasticsearchVersion() {
 		}
 		return nil
 	})
+
+	return err
 }
 
 func (b *Beat) isConnectionToOlderVersionAllowed() bool {
@@ -1026,13 +1039,32 @@ func (b *Beat) indexSetupCallback() elasticsearch.ConnectCallback {
 }
 
 func (b *Beat) makeOutputReloader(outReloader pipeline.OutputReloader) reload.Reloadable {
-	return reload.ReloadableFunc(func(config *reload.ConfigWithMeta) error {
+	return reload.ReloadableFunc(func(update *reload.ConfigWithMeta) error {
+		if update == nil {
+			return nil
+		}
+
 		if b.OutputConfigReloader != nil {
-			if err := b.OutputConfigReloader.Reload(config); err != nil {
+			if err := b.OutputConfigReloader.Reload(update); err != nil {
+				return err
+			}
+		}
+
+		if update.Config != nil {
+			err := b.Config.Output.Unpack(update.Config)
+			if err != nil {
 				return err
 			}
+
+			// after the output configuration change
+			if isElasticsearchOutput(b.Config.Output.Name()) {
+				// if the output is Elasticsearch, we need to update some pre-saved flags
+				// for its global connection callbacks
+				b.allowOlderESVersions = b.isConnectionToOlderVersionAllowed()
+			}
 		}
-		return outReloader.Reload(config, b.createOutput)
+
+		return outReloader.Reload(update, b.createOutput)
 	})
 }
 

libbeat/cmd/instance/beat_test.go

@@ -26,6 +26,9 @@ import (
 	"testing"
 
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
+	"github.com/elastic/beats/v7/libbeat/common/reload"
+	"github.com/elastic/beats/v7/libbeat/outputs"
+	"github.com/elastic/elastic-agent-libs/config"
 
 	"github.com/gofrs/uuid"
 	"github.com/stretchr/testify/assert"
@@ -224,3 +227,79 @@ func TestSanitizeIPs(t *testing.T) {
 		})
 	}
 }
+
+func TestReloader(t *testing.T) {
+	t.Run("updates the output configuration on the beat", func(t *testing.T) {
+		b, err := NewBeat("testbeat", "testidx", "0.9", false)
+		require.NoError(t, err)
+
+		cfg := `
+elasticsearch:
+  hosts: ["https://127.0.0.1:9200"]
+  username: "elastic"
+  allow_older_versions: true
+`
+		c, err := config.NewConfigWithYAML([]byte(cfg), cfg)
+		require.NoError(t, err)
+		outCfg, err := c.Child("elasticsearch", -1)
+		require.NoError(t, err)
+
+		update := &reload.ConfigWithMeta{Config: c}
+		m := &outputReloaderMock{}
+		reloader := b.makeOutputReloader(m)
+
+		require.False(t, b.allowOlderESVersions)
+		require.False(t, b.Config.Output.IsSet(), "the output should not be set yet")
+
+		err = reloader.Reload(update)
+		require.NoError(t, err)
+
+		require.True(t, b.Config.Output.IsSet(), "now the output should be set")
+		require.True(t, b.allowOlderESVersions)
+		require.Equal(t, outCfg, b.Config.Output.Config())
+		require.Same(t, c, m.cfg.Config)
+	})
+
+	t.Run("does not update `allowOlderESVersions` if output is not Elasticsearch", func(t *testing.T) {
+		b, err := NewBeat("testbeat", "testidx", "0.9", false)
+		require.NoError(t, err)
+
+		cfg := `
+logstash:
+  hosts: ["https://127.0.0.1:9200"]
+  username: "elastic"
+  allow_older_versions: true
+`
+		c, err := config.NewConfigWithYAML([]byte(cfg), cfg)
+		require.NoError(t, err)
+		outCfg, err := c.Child("logstash", -1)
+		require.NoError(t, err)
+
+		update := &reload.ConfigWithMeta{Config: c}
+		m := &outputReloaderMock{}
+		reloader := b.makeOutputReloader(m)
+
+		require.False(t, b.allowOlderESVersions)
+		require.False(t, b.Config.Output.IsSet(), "the output should not be set yet")
+
+		err = reloader.Reload(update)
+		require.NoError(t, err)
+
+		require.True(t, b.Config.Output.IsSet(), "now the output should be set")
+		require.False(t, b.allowOlderESVersions, "the flag should not be updated")
+		require.Equal(t, outCfg, b.Config.Output.Config())
+		require.Same(t, c, m.cfg.Config)
+	})
+}
+
+type outputReloaderMock struct {
+	cfg *reload.ConfigWithMeta
+}
+
+func (r *outputReloaderMock) Reload(
+	cfg *reload.ConfigWithMeta,
+	factory func(o outputs.Observer, cfg config.Namespace) (outputs.Group, error),
+) error {
+	r.cfg = cfg
+	return nil
+}