Incorporate obsoleted overwritepropertiesconverter

open-telemetry/opentelemetry-collector#6656 removed
the overwrite properties config converter that this distribution uses.

internal/configconverter/overwrite_properties.go

@@ -0,0 +1,65 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Taken from https://github.com/open-telemetry/opentelemetry-collector/blob/v0.66.0/confmap/converter/overwritepropertiesconverter/properties.go
+// to prevent breaking changes.
+// "Deprecated: [v0.63.0] this converter will not be supported anymore because of dot separation limitation.
+// See https://github.com/open-telemetry/opentelemetry-collector/issues/6294 for more details."
+package configconverter
+
+import (
+	"bytes"
+	"context"
+	"strings"
+
+	"github.com/knadh/koanf/maps"
+	"github.com/magiconair/properties"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+type converter struct {
+	properties []string
+}
+
+func NewOverwritePropertiesConverter(properties []string) confmap.Converter {
+	return &converter{properties: properties}
+}
+
+func (c *converter) Convert(_ context.Context, conf *confmap.Conf) error {
+	if len(c.properties) == 0 {
+		return nil
+	}
+
+	b := &bytes.Buffer{}
+	for _, property := range c.properties {
+		property = strings.TrimSpace(property)
+		b.WriteString(property)
+		b.WriteString("\n")
+	}
+
+	var props *properties.Properties
+	var err error
+	if props, err = properties.Load(b.Bytes(), properties.UTF8); err != nil {
+		return err
+	}
+
+	// Create a map manually instead of using properties.Map() to not expand the env vars.
+	parsed := make(map[string]interface{}, props.Len())
+	for _, key := range props.Keys() {
+		value, _ := props.Get(key)
+		parsed[key] = value
+	}
+	prop := maps.Unflatten(parsed, ".")
+	return conf.Merge(confmap.NewFromStringMap(prop))
+}

internal/configconverter/overwrite_properties_test.go

@@ -0,0 +1,58 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Taken from https://github.com/open-telemetry/opentelemetry-collector/blob/v0.66.0/confmap/converter/overwritepropertiesconverter/properties_test.go
+// to prevent breaking changes.
+package configconverter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func TestOverwritePropertiesConverter_Empty(t *testing.T) {
+	pmp := NewOverwritePropertiesConverter(nil)
+	conf := confmap.NewFromStringMap(map[string]interface{}{"foo": "bar"})
+	assert.NoError(t, pmp.Convert(context.Background(), conf))
+	assert.Equal(t, map[string]interface{}{"foo": "bar"}, conf.ToStringMap())
+}
+
+func TestOverwritePropertiesConverter(t *testing.T) {
+	props := []string{
+		"processors.batch.timeout=2s",
+		"processors.batch/foo.timeout=3s",
+		"receivers.otlp.protocols.grpc.endpoint=localhost:1818",
+		"exporters.kafka.brokers=foo:9200,foo2:9200",
+	}
+
+	pmp := NewOverwritePropertiesConverter(props)
+	conf := confmap.New()
+	require.NoError(t, pmp.Convert(context.Background(), conf))
+	keys := conf.AllKeys()
+	assert.Len(t, keys, 4)
+	assert.Equal(t, "2s", conf.Get("processors::batch::timeout"))
+	assert.Equal(t, "3s", conf.Get("processors::batch/foo::timeout"))
+	assert.Equal(t, "foo:9200,foo2:9200", conf.Get("exporters::kafka::brokers"))
+	assert.Equal(t, "localhost:1818", conf.Get("receivers::otlp::protocols::grpc::endpoint"))
+}
+
+func TestOverwritePropertiesConverter_InvalidProperty(t *testing.T) {
+	pmp := NewOverwritePropertiesConverter([]string{"=2s"})
+	conf := confmap.New()
+	assert.Error(t, pmp.Convert(context.Background(), conf))
+}

internal/settings/settings.go

@@ -138,10 +138,8 @@ func getConfigDir(f *Settings) string {
 // ConfMapConverters returns confmap.Converters for the collector core service.
 func (s *Settings) ConfMapConverters() []confmap.Converter {
 	confMapConverters := []confmap.Converter{
-		// nolint: staticcheck
-		// overwritepropertiesconverter.New(s.setProperties.value), // support until there's an actual replacement
+		configconverter.NewOverwritePropertiesConverter(s.setProperties.value),
 	}
-
 	if !s.noConvertConfig {
 		confMapConverters = append(
 			confMapConverters,

internal/settings/settings_test.go

@@ -89,7 +89,7 @@ func TestNewSettingsNoConvertConfig(t *testing.T) {
 		"--config", anotherConfigPath,
 		"--set", "foo",
 		"--set", "bar",
-		"--set", "baz",
+		"--set=baz",
 		"--feature-gates", "foo",
 		"--feature-gates", "-bar",
 	})
@@ -102,8 +102,7 @@ func TestNewSettingsNoConvertConfig(t *testing.T) {
 
 	require.Equal(t, []string{configPath, anotherConfigPath}, settings.ResolverURIs())
 	require.Equal(t, []confmap.Converter{
-		// nolint: staticcheck
-		// overwritepropertiesconverter.New(settings.setProperties.value), // support until there's an actual replacement
+		configconverter.NewOverwritePropertiesConverter(settings.setProperties.value),
 	}, settings.ConfMapConverters())
 	require.Equal(t, []string{"--feature-gates", "foo", "--feature-gates", "-bar"}, settings.ColCoreArgs())
 }
@@ -114,7 +113,7 @@ func TestNewSettingsConvertConfig(t *testing.T) {
 		"--config", configPath,
 		"--config", anotherConfigPath,
 		"--set", "foo",
-		"--set", "bar",
+		"--set=bar",
 		"--set", "baz",
 		"--feature-gates", "foo",
 		"--feature-gates", "-bar",
@@ -129,8 +128,7 @@ func TestNewSettingsConvertConfig(t *testing.T) {
 
 	require.Equal(t, []string{configPath, anotherConfigPath}, settings.ResolverURIs())
 	require.Equal(t, []confmap.Converter{
-		// nolint: staticcheck
-		// overwritepropertiesconverter.New(settings.setProperties.value), // support until there's an actual replacement
+		configconverter.NewOverwritePropertiesConverter(settings.setProperties.value),
 		configconverter.RemoveBallastKey{},
 		configconverter.MoveOTLPInsecureKey{},
 		configconverter.MoveHecTLS{},

tests/general/set_properties_test.go

@@ -0,0 +1,152 @@
+// Copyright Splunk, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build integration
+
+package tests
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/signalfx/splunk-otel-collector/tests/testutils"
+)
+
+func TestSetProperties(t *testing.T) {
+	tc := testutils.NewTestcase(t)
+	defer tc.PrintLogsOnFailure()
+	defer tc.ShutdownOTLPReceiverSink()
+
+	configServerPort := testutils.GetAvailablePort(t)
+	cc, shutdown := tc.SplunkOtelCollectorContainer(
+		"set_properties_config.yaml",
+		func(collector testutils.Collector) testutils.Collector {
+			return collector.
+				WithArgs(
+					"--config", "/etc/config.yaml",
+					"--set", "receivers.otlp.protocols.grpc.max_recv_msg_size_mib=1",
+					"--set=receivers.otlp.protocols.http.endpoint=localhost:0",
+					"--set", "processors.filter/one.metrics.include.match_type=regexp",
+					"--set=processors.filter/one.metrics.include.metric_names=[a.name]",
+					"--set", "processors.filter/two.metrics.include.match_type=strict",
+					"--set=processors.filter/two.metrics.include.metric_names=[another.name]",
+				).
+				WithEnv(
+					map[string]string{
+						"SPLUNK_DEBUG_CONFIG_SERVER_PORT": fmt.Sprintf("%d", configServerPort),
+					},
+				)
+		},
+	)
+	defer shutdown()
+
+	expectedInitial := map[string]any{
+		"file": map[string]any{
+			"receivers": map[string]any{
+				"otlp": map[string]any{"protocols": "overwritten"},
+			},
+			"processors": map[string]any{
+				"filter/one": map[string]any{
+					"metrics": map[string]any{
+						"include": map[string]any{
+							"match_type":   "overwritten",
+							"metric_names": "overwritten",
+						},
+					},
+				},
+				"filter/two": map[string]any{
+					"metrics": map[string]any{
+						"include": map[string]any{
+							"match_type":   "overwritten",
+							"metric_names": "overwritten",
+						},
+					},
+				},
+			},
+			"exporters": map[string]any{
+				"otlp": map[string]any{
+					"endpoint": "${OTLP_ENDPOINT}",
+					"tls": map[string]any{
+						"insecure": true,
+					},
+				},
+			},
+			"service": map[string]any{
+				"pipelines": map[string]any{
+					"metrics": map[string]any{
+						"receivers":  []any{"otlp"},
+						"processors": []any{"filter/one", "filter/two"},
+						"exporters":  []any{"otlp"},
+					},
+				},
+			},
+		},
+	}
+
+	require.Equal(t, expectedInitial, cc.InitialConfig(t, configServerPort))
+
+	expectedEffective := map[string]any{
+		"receivers": map[string]any{
+			"otlp": map[string]any{
+				"protocols": map[string]any{
+					"grpc": map[string]any{
+						"max_recv_msg_size_mib": "1",
+					},
+					"http": map[string]any{
+						"endpoint": "localhost:0",
+					},
+				},
+			},
+		},
+		"processors": map[string]any{
+			"filter/one": map[string]any{
+				"metrics": map[string]any{
+					"include": map[string]any{
+						"match_type":   "regexp",
+						"metric_names": "[a.name]",
+					},
+				},
+			},
+			"filter/two": map[string]any{
+				"metrics": map[string]any{
+					"include": map[string]any{
+						"match_type":   "strict",
+						"metric_names": "[another.name]",
+					},
+				},
+			},
+		},
+		"exporters": map[string]any{
+			"otlp": map[string]any{
+				"endpoint": tc.OTLPEndpoint,
+				"tls": map[string]any{
+					"insecure": true,
+				},
+			},
+		},
+		"service": map[string]any{
+			"pipelines": map[string]any{
+				"metrics": map[string]any{
+					"receivers":  []any{"otlp"},
+					"processors": []any{"filter/one", "filter/two"},
+					"exporters":  []any{"otlp"},
+				},
+			},
+		},
+	}
+
+	require.Equal(t, expectedEffective, cc.EffectiveConfig(t, configServerPort))
+}

tests/general/testdata/set_properties_config.yaml

@@ -0,0 +1,28 @@
+receivers:
+  otlp:
+    protocols: overwritten
+
+processors:
+  filter/one:
+    metrics:
+      include:
+        match_type: overwritten
+        metric_names: overwritten
+  filter/two:
+    metrics:
+      include:
+        match_type: overwritten
+        metric_names: overwritten
+
+exporters:
+  otlp:
+    endpoint: "${OTLP_ENDPOINT}"
+    tls:
+      insecure: true
+
+service:
+  pipelines:
+    metrics:
+      receivers: [otlp]
+      processors: [filter/one, filter/two]
+      exporters: [otlp]

tests/testutils/collector_container.go

@@ -263,9 +263,18 @@ func (collector *CollectorContainer) EffectiveConfig(t testing.TB, port uint16)
 }
 
 func (collector *CollectorContainer) execConfigRequest(t testing.TB, uri string) map[string]any {
-	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-	defer cancel()
-	n, r, err := collector.Container.Exec(ctx, []string{"curl", "-s", uri})
+	var n int
+	var r io.Reader
+	var err error
+	for i := 0; i < 3; i++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		n, r, err = collector.Container.Exec(ctx, []string{"curl", "-s", uri})
+		cancel()
+		if err == nil && n == 0 {
+			break
+		}
+		time.Sleep(time.Second)
+	}
 	require.NoError(t, err)
 	require.Zero(t, n)
 	out, err := io.ReadAll(r)