feat(ConfigProvider): Modify config merge to do replace instead of update for Specific tags for Remotely managed collectors

.changelog/1604.changed.txt

@@ -0,0 +1 @@
+feat(ConfigProvider)!: Modify config merge to do replace instead of update for Specific tags for Remotely managed collectors

otelcolbuilder/.otelcol-builder.yaml

@@ -245,3 +245,4 @@ replaces:
   - github.com/open-telemetry/opentelemetry-collector-contrib/extension/sumologicextension v0.108.0 => github.com/SumoLogic/opentelemetry-collector-contrib/extension/sumologicextension 6afd5814da3e09014683ab0e4a08f8f781bc904f
   - github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/globprovider => ../../pkg/configprovider/globprovider
   - github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/opampprovider => ../../pkg/configprovider/opampprovider
+  - github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil => ../../pkg/configprovider/providerutil

pkg/configprovider/globprovider/go.mod

@@ -5,6 +5,7 @@ go 1.22.0
 toolchain go1.22.8
 
 require (
+	github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil v0.0.0-00010101000000-000000000000
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/confmap v0.107.0
 	gopkg.in/yaml.v3 v3.0.1
@@ -25,3 +26,5 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 )
+
+replace github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil => ../providerutil

pkg/configprovider/globprovider/provider.go

@@ -22,27 +22,33 @@ import (
 	"sort"
 	"strings"
 
-	"gopkg.in/yaml.v3"
-
+	"github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil"
 	"go.opentelemetry.io/collector/confmap"
+	"gopkg.in/yaml.v3"
 )
 
 const (
 	schemeName   = "glob"
 	schemePrefix = schemeName + ":"
 )
 
-type provider struct{}
+type Provider struct{
+	remotelyManagedMergeFlow bool
+}
 
 func NewWithSettings(_ confmap.ProviderSettings) confmap.Provider {
-	return &provider{}
+	return &Provider{}
 }
 
 func NewFactory() confmap.ProviderFactory {
 	return confmap.NewProviderFactory(NewWithSettings)
 }
 
-func (fmp *provider) Retrieve(ctx context.Context, uri string, _ confmap.WatcherFunc) (*confmap.Retrieved, error) {
+func (fmp *Provider) SetRemotelyManagedMergeFlow(enable bool) {
+    fmp.remotelyManagedMergeFlow = enable
+}
+
+func (fmp *Provider) Retrieve(ctx context.Context, uri string, _ confmap.WatcherFunc) (*confmap.Retrieved, error) {
 	var rawConf map[string]interface{}
 	if !strings.HasPrefix(uri, schemePrefix) {
 		return &confmap.Retrieved{}, fmt.Errorf("%q uri is not supported by %q provider", uri, schemeName)
@@ -68,18 +74,21 @@ func (fmp *provider) Retrieve(ctx context.Context, uri string, _ confmap.Watcher
 			return &confmap.Retrieved{}, err
 		}
 		pathConf := confmap.NewFromStringMap(rawConf)
+		if fmp.remotelyManagedMergeFlow {
+			providerutil.PrepareForReplaceBehavior(conf, pathConf)
+		}
 		if err := conf.Merge(pathConf); err != nil {
 			return &confmap.Retrieved{}, err
 		}
-	}
 
+	}
 	return confmap.NewRetrieved(conf.ToStringMap())
 }
 
-func (*provider) Scheme() string {
+func (*Provider) Scheme() string {
 	return schemeName
 }
 
-func (fmp *provider) Shutdown(context.Context) error {
+func (fmp *Provider) Shutdown(context.Context) error {
 	return nil
 }

pkg/configprovider/globprovider/provider_test.go

@@ -159,3 +159,49 @@ func ValidateProviderScheme(p confmap.Provider) error {
 
 	return nil
 }
+
+func TestRemotelyManagedMergeFlow(t *testing.T) {
+	fp := NewWithSettings(confmap.ProviderSettings{})
+	if globProvider, ok := fp.(*Provider); ok {
+    		globProvider.SetRemotelyManagedMergeFlow(true)
+	}
+	ret, err := fp.Retrieve(context.Background(), schemePrefix+filepath.Join("testdata", "mergefunc", "*.yaml"), nil)
+	require.NoError(t, err)
+	retMap, err := ret.AsConf()
+	assert.NoError(t, err)
+	expectedMap := confmap.NewFromStringMap(map[string]interface{}{
+				"extensions": map[string]interface{}{
+					"sumologic": map[string]interface{}{
+						"childKey": "value",
+						"collector_fields": map[string]interface{}{
+							"zone": "eu",
+						},
+					},
+				},
+				"processor": "someprocessor",
+			})
+	assert.Equal(t, expectedMap, retMap)
+	assert.NoError(t, fp.Shutdown(context.Background()))
+}
+
+func TestLocallyManagedMergeFlow(t *testing.T) {
+	fp := NewWithSettings(confmap.ProviderSettings{})
+	ret, err := fp.Retrieve(context.Background(), schemePrefix+filepath.Join("testdata", "mergefunc", "*.yaml"), nil)
+	require.NoError(t, err)
+	retMap, err := ret.AsConf()
+	assert.NoError(t, err)
+	expectedMap := confmap.NewFromStringMap(map[string]interface{}{
+				"extensions": map[string]interface{}{
+					"sumologic": map[string]interface{}{
+						"childKey": "value",
+						"collector_fields": map[string]interface{}{
+							"cluster": "cluster-1",
+							"zone": "eu",
+						},
+					},
+				},
+				"processor": "someprocessor",
+			})
+	assert.Equal(t, expectedMap, retMap)
+	assert.NoError(t, fp.Shutdown(context.Background()))
+}

pkg/configprovider/globprovider/testdata/mergefunc/a_srcconf.yaml

@@ -0,0 +1,6 @@
+extensions:
+  sumologic:
+    childKey: "value"
+    collector_fields:
+      cluster: "cluster-1"
+processor: "someprocessor"

pkg/configprovider/globprovider/testdata/mergefunc/mergeconf.yaml

@@ -0,0 +1,5 @@
+extensions:
+  sumologic:
+    childKey: "value"
+    collector_fields:
+      zone: "eu"

pkg/configprovider/opampprovider/go.mod

@@ -6,6 +6,7 @@ toolchain go1.22.8
 
 require (
 	github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/globprovider v0.0.0-00010101000000-000000000000
+	github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil v0.0.0-00010101000000-000000000000
 	github.com/google/go-cmp v0.5.9
 	go.opentelemetry.io/collector/confmap v1.14.0
 	go.opentelemetry.io/collector/confmap/provider/fileprovider v0.108.0
@@ -25,3 +26,5 @@ require (
 )
 
 replace github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/globprovider => ../globprovider
+
+replace github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil => ../providerutil

pkg/configprovider/opampprovider/provider.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/globprovider"
+	"github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil"
 	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/confmap/provider/fileprovider"
 	"gopkg.in/yaml.v2"
@@ -38,6 +39,7 @@ type ConfigFragment struct {
 	Extensions struct {
 		OpAmp struct {
 			RemoteConfigurationDirectory string `yaml:"remote_configuration_directory"`
+			DisableTagReplacement bool `yaml:"disable_tag_replacement"`
 		} `yaml:"opamp"`
 	} `yaml:"extensions"`
 }
@@ -46,6 +48,10 @@ func (c ConfigFragment) ConfigDir() string {
 	return c.Extensions.OpAmp.RemoteConfigurationDirectory
 }
 
+func (c ConfigFragment) IsRemotelyManagedMergeFlow() bool {
+	return !c.Extensions.OpAmp.DisableTagReplacement
+}
+
 func (c ConfigFragment) Validate() error {
 	if c.ConfigDir() == "" {
 		return errors.New("remote_configuration_directory missing from opamp extension")
@@ -84,6 +90,11 @@ func (p *Provider) Retrieve(ctx context.Context, configPath string, fn confmap.W
 	}
 	conf := confmap.New()
 	glob := p.GlobProvider
+
+	if globProvider, ok := glob.(*globprovider.Provider); ok {
+        	globProvider.SetRemotelyManagedMergeFlow(cfg.IsRemotelyManagedMergeFlow())
+	}
+
 	retrieved, err := glob.Retrieve(ctx, glob.Scheme()+":"+filepath.Join(cfg.ConfigDir(), "*.yaml"), fn)
 	if err != nil {
 		return nil, err
@@ -100,6 +111,12 @@ func (p *Provider) Retrieve(ctx context.Context, configPath string, fn confmap.W
 	if err != nil {
 		return nil, err
 	}
+
+	if cfg.IsRemotelyManagedMergeFlow() {
+		// Order of conf parameters is important, see method comments
+		providerutil.PrepareForReplaceBehavior(addlConf, retConf)
+	}
+
 	// merge the file config in
 	if err := conf.Merge(addlConf); err != nil {
 		return nil, err

pkg/configprovider/opampprovider/provider_test.go

@@ -121,6 +121,75 @@ func TestValid(t *testing.T) {
 	}
 }
 
+func TestRemotelyManagedFlowDisabled(t *testing.T) {
+	p := NewWithSettings(confmap.ProviderSettings{})
+	defer func() {
+		if err := p.Shutdown(context.Background()); err != nil {
+			t.Error(err)
+		}
+	}()
+
+	configPath := "opamp:" + absolutePath(t, filepath.Join("testdata", "config_new_merge_disabled.yaml"))
+	t.Logf("loading opamp config file: %s", configPath)
+
+	ret, err := p.Retrieve(context.Background(), configPath, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	conf, err := ret.AsConf()
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := conf.ToStringMap()
+	exp := confmap.NewFromStringMap(map[string]any{
+		"extensions::sumologic::childKey":                   "value",
+		"extensions::sumologic::collector_fields::cluster":   "cluster-1",
+                "extensions::sumologic::collector_fields::zone":      "eu",
+		"processor":                                         "someprocessor",
+		"extensions::opamp::remote_configuration_directory": "../globprovider/testdata/mergefunc",
+		"extensions::opamp::endpoint":                       "wss://example.com/v1/opamp",
+		"extensions::opamp::disable_tag_replacement":   true,
+	})
+	want := exp.ToStringMap()
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Errorf("Retrieve() mismatch (-want +got):\n%s", diff)
+	}
+}
+
+func TestRemotelyManagedFlowEnabled(t *testing.T) {
+	p := NewWithSettings(confmap.ProviderSettings{})
+	defer func() {
+		if err := p.Shutdown(context.Background()); err != nil {
+			t.Error(err)
+		}
+	}()
+
+	configPath := "opamp:" + absolutePath(t, filepath.Join("testdata", "config_new_merge_enabled.yaml"))
+	t.Logf("loading opamp config file: %s", configPath)
+
+	ret, err := p.Retrieve(context.Background(), configPath, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	conf, err := ret.AsConf()
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := conf.ToStringMap()
+	exp := confmap.NewFromStringMap(map[string]any{
+		"extensions::sumologic::childKey":                   "value",
+		"extensions::sumologic::collector_fields::zone":      "eu",
+		"processor":                                         "someprocessor",
+		"extensions::opamp::remote_configuration_directory": "../globprovider/testdata/mergefunc",
+		"extensions::opamp::endpoint":                       "wss://example.com/v1/opamp",
+	})
+	want := exp.ToStringMap()
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Errorf("Retrieve() mismatch (-want +got):\n%s", diff)
+	}
+}
+
+
 func absolutePath(t *testing.T, relativePath string) string {
 	t.Helper()
 	pth, err := filepath.Abs(relativePath)

pkg/configprovider/opampprovider/testdata/config_new_merge_disabled.yaml

@@ -0,0 +1,5 @@
+extensions:
+  opamp:
+    endpoint: "wss://example.com/v1/opamp"
+    remote_configuration_directory: ../globprovider/testdata/mergefunc
+    disable_tag_replacement: true

pkg/configprovider/opampprovider/testdata/config_new_merge_enabled.yaml

@@ -0,0 +1,4 @@
+extensions:
+  opamp:
+    endpoint: "wss://example.com/v1/opamp"
+    remote_configuration_directory: ../globprovider/testdata/mergefunc

pkg/configprovider/providerutil/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

pkg/configprovider/providerutil/go.mod

@@ -0,0 +1,24 @@
+module github.com/SumoLogic/sumologic-otel-collector/pkg/configprovider/providerutil
+
+go 1.21.0
+
+toolchain go1.22.3
+
+require (
+	github.com/stretchr/testify v1.9.0
+	go.opentelemetry.io/collector/confmap v0.102.1
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect
+	github.com/knadh/koanf/maps v0.1.1 // indirect
+	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
+	github.com/knadh/koanf/v2 v2.1.1 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

pkg/configprovider/providerutil/go.sum

@@ -0,0 +1,30 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 h1:TQcrn6Wq+sKGkpyPvppOz99zsMBaUOKXq6HSv655U1c=
+github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
+github.com/knadh/koanf/maps v0.1.1 h1:G5TjmUh2D7G2YWf5SQQqSiHRJEjaicvU0KpypqB3NIs=
+github.com/knadh/koanf/maps v0.1.1/go.mod h1:npD/QZY3V6ghQDdcQzl1W4ICNVTkohC8E73eI2xW4yI=
+github.com/knadh/koanf/providers/confmap v0.1.0 h1:gOkxhHkemwG4LezxxN8DMOFopOPghxRVp7JbIvdvqzU=
+github.com/knadh/koanf/providers/confmap v0.1.0/go.mod h1:2uLhxQzJnyHKfxG927awZC7+fyHFdQkd697K4MdLnIU=
+github.com/knadh/koanf/v2 v2.1.1 h1:/R8eXqasSTsmDCsAyYj+81Wteg8AqrV9CP6gvsTsOmM=
+github.com/knadh/koanf/v2 v2.1.1/go.mod h1:4mnTRbZCK+ALuBXHZMjDfG9y714L7TykVnZkXbMU3Es=
+github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa15WveJJGw=
+github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
+github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ=
+github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+go.opentelemetry.io/collector/confmap v0.102.1 h1:wZuH+d/P11Suz8wbp+xQCJ0BPE9m5pybtUe74c+rU7E=
+go.opentelemetry.io/collector/confmap v0.102.1/go.mod h1:KgpS7UxH5rkd69CzAzlY2I1heH8Z7eNCZlHmwQBMxNg=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=