Handle backwards compatibility of internal ballast removal

cmd/otelcol/main.go

@@ -31,6 +31,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/signalfx/splunk-otel-collector/internal/components"
+	"github.com/signalfx/splunk-otel-collector/internal/configconverter"
 	"github.com/signalfx/splunk-otel-collector/internal/configprovider"
 	"github.com/signalfx/splunk-otel-collector/internal/configsources"
 	"github.com/signalfx/splunk-otel-collector/internal/version"
@@ -60,14 +61,14 @@ func main() {
 	// TODO: Use same format as the collector
 	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
 
-	if !contains(os.Args[1:], "-h") && !contains(os.Args[1:], "--help") {
+	if !hasFlag("-h") && !hasFlag("--help") {
 		checkRuntimeParams()
 		setDefaultEnvVars()
 	}
 
 	// Allow dumping configuration locally by default
 	// Used by support bundle script
-	os.Setenv(configServerEnabledEnvVar, "true")
+	_ = os.Setenv(configServerEnabledEnvVar, "true")
 
 	factories, err := components.Get()
 	if err != nil {
@@ -85,6 +86,13 @@ func main() {
 		info,
 		configsources.Get()...,
 	)
+	const noConvertConfigFlag = "--no-convert-config"
+	if hasFlag(noConvertConfigFlag) {
+		// the collector complains about this flag if we don't remove it
+		removeFlag(&os.Args, noConvertConfigFlag)
+	} else {
+		parserProvider = configconverter.ParserProvider(parserProvider, configconverter.RemoveBallastKey)
+	}
 	serviceParams := service.CollectorSettings{
 		BuildInfo:      info,
 		Factories:      factories,
@@ -96,6 +104,10 @@ func main() {
 	}
 }
 
+func hasFlag(flag string) bool {
+	return contains(os.Args[1:], flag)
+}
+
 // Check whether a string exists in an array of CLI arguments
 // Support key/value with and without an equal sign
 func contains(arr []string, str string) bool {
@@ -111,6 +123,16 @@ func contains(arr []string, str string) bool {
 	return false
 }
 
+func removeFlag(flags *[]string, flag string) {
+	var out []string
+	for _, s := range *flags {
+		if s != flag {
+			out = append(out, s)
+		}
+	}
+	*flags = out
+}
+
 // Get the value of a key in an array
 // Support key/value with and with an equal sign
 func getKeyValue(args []string, arg string) (exists bool, value string) {
@@ -256,7 +278,7 @@ func setMemoryBallast(memTotalSizeMiB int) int {
 		if os.Getenv(ballastEnvVarName) != "" {
 			log.Fatalf("Both %v and '--mem-ballast-size-mib' were specified, but only one is allowed", ballastEnvVarName)
 		}
-		os.Setenv(ballastEnvVarName, ballastSizeFlag)
+		_ = os.Setenv(ballastEnvVarName, ballastSizeFlag)
 	}
 
 	ballastSize := memTotalSizeMiB * defaultMemoryBallastPercentage / 100
@@ -268,7 +290,7 @@ func setMemoryBallast(memTotalSizeMiB int) int {
 		}
 	}
 
-	os.Setenv(ballastEnvVarName, strconv.Itoa(ballastSize))
+	_ = os.Setenv(ballastEnvVarName, strconv.Itoa(ballastSize))
 	log.Printf("Set ballast to %d MiB", ballastSize)
 	return ballastSize
 }
@@ -282,7 +304,7 @@ func setMemoryLimit(memTotalSizeMiB int) int {
 		memLimit = envVarAsInt(memLimitMiBEnvVarName)
 	}
 
-	os.Setenv(memLimitMiBEnvVarName, strconv.Itoa(memLimit))
+	_ = os.Setenv(memLimitMiBEnvVarName, strconv.Itoa(memLimit))
 	log.Printf("Set memory limit to %d MiB", memLimit)
 	return memLimit
 }
@@ -300,15 +322,15 @@ func setDefaultEnvVars() {
 		for _, v := range testArgs {
 			_, ok := os.LookupEnv(v[0])
 			if !ok {
-				os.Setenv(v[0], v[1])
+				_ = os.Setenv(v[0], v[1])
 			}
 		}
 	}
 	token, tokenOk := os.LookupEnv("SPLUNK_ACCESS_TOKEN")
 	if tokenOk {
 		_, ok := os.LookupEnv("SPLUNK_HEC_TOKEN")
 		if !ok {
-			os.Setenv("SPLUNK_HEC_TOKEN", token)
+			_ = os.Setenv("SPLUNK_HEC_TOKEN", token)
 		}
 	}
 }

cmd/otelcol/main_test.go

@@ -311,3 +311,13 @@ service:
 		})
 	}
 }
+
+func TestRemoveFlag(t *testing.T) {
+	args := []string{"--aaa", "--bbb", "--ccc"}
+	removeFlag(&args, "--bbb")
+	assert.Equal(t, []string{"--aaa", "--ccc"}, args)
+	removeFlag(&args, "--ccc")
+	assert.Equal(t, []string{"--aaa"}, args)
+	removeFlag(&args, "--aaa")
+	assert.Nil(t, args)
+}

internal/configconverter/ballast_remover.go

@@ -0,0 +1,41 @@
+// 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.
+
+package configconverter
+
+import (
+	"log"
+	"regexp"
+
+	"go.opentelemetry.io/collector/config/configparser"
+)
+
+// RemoveBallastKey is a CfgMapFunc that removes a ballast_size_mib on a
+// memory_limiter processor config if it exists. This config key will go away at
+// some point (or already has) at which point its presence in a config will
+// prevent the Collector from starting.
+func RemoveBallastKey(cfgMap *configparser.ConfigMap) *configparser.ConfigMap {
+	const expr = "processors::memory_limiter(/\\w+)?::ballast_size_mib"
+	ballastKeyRegexp, _ := regexp.Compile(expr)
+
+	out := configparser.NewParser()
+	for _, k := range cfgMap.AllKeys() {
+		if ballastKeyRegexp.MatchString(k) {
+			log.Println("Deprecated memory_limiter processor `ballast_size_mib` key found. Removing from config.")
+		} else {
+			out.Set(k, cfgMap.Get(k))
+		}
+	}
+	return out
+}

internal/configconverter/parser_provider.go

@@ -0,0 +1,55 @@
+// 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.
+
+package configconverter
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/config/configparser"
+	"go.opentelemetry.io/collector/service/parserprovider"
+)
+
+// converterProvider wraps a ParserProvider and accepts a list of functions that
+// convert ConfigMaps. The idea is for this type to conform to the open-closed
+// principle.
+type converterProvider struct {
+	wrapped     parserprovider.ParserProvider
+	cfgMapFuncs []CfgMapFunc
+}
+
+type CfgMapFunc func(*configparser.ConfigMap) *configparser.ConfigMap
+
+var _ parserprovider.ParserProvider = (*converterProvider)(nil)
+
+func ParserProvider(wrapped parserprovider.ParserProvider, funcs ...CfgMapFunc) parserprovider.ParserProvider {
+	return &converterProvider{wrapped: wrapped, cfgMapFuncs: funcs}
+}
+
+func (p converterProvider) Get() (*configparser.ConfigMap, error) {
+	cfgMap, err := p.wrapped.Get()
+	if err != nil {
+		return nil, fmt.Errorf("converterProvider.Get(): %w", err)
+	}
+
+	for _, cfgMapFunc := range p.cfgMapFuncs {
+		cfgMap = cfgMapFunc(cfgMap)
+	}
+
+	out := configparser.NewParser()
+	for _, k := range cfgMap.AllKeys() {
+		out.Set(k, cfgMap.Get(k))
+	}
+	return out, nil
+}

internal/configconverter/parser_provider_test.go

@@ -0,0 +1,62 @@
+// 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.
+
+package configconverter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/config/configparser"
+)
+
+func TestMemLimitBallastRemoverPP(t *testing.T) {
+	tests := []struct {
+		name  string
+		fname string
+		key   string
+	}{
+		{
+			name:  "default",
+			fname: "testdata/ballast_mem_limiter.yaml",
+			key:   "processors::memory_limiter::ballast_size_mib",
+		},
+		{
+			name:  "custom",
+			fname: "testdata/ballast_mem_limiter_custom.yaml",
+			key:   "processors::memory_limiter/foo::ballast_size_mib",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			pp := &converterProvider{
+				wrapped:     &fileParserProvider{fileName: test.fname},
+				cfgMapFuncs: []CfgMapFunc{RemoveBallastKey},
+			}
+			cfgMap, err := pp.Get()
+			require.NoError(t, err)
+			b := cfgMap.IsSet(test.key)
+			assert.False(t, b)
+		})
+	}
+}
+
+type fileParserProvider struct {
+	fileName string
+}
+
+func (fpp fileParserProvider) Get() (*configparser.ConfigMap, error) {
+	return configparser.NewParserFromFile(fpp.fileName)
+}

internal/configconverter/testdata/ballast_mem_limiter.yaml

@@ -0,0 +1,30 @@
+extensions:
+  memory_ballast:
+    size_mib: 64
+receivers:
+  hostmetrics:
+    collection_interval: 1s
+    scrapers:
+      cpu:
+processors:
+  memory_limiter:
+    check_interval: 1s
+    limit_mib: 4000
+    spike_limit_mib: 800
+    ballast_size_mib: 64
+exporters:
+  logging:
+    loglevel: info
+    sampling_initial: 2
+    sampling_thereafter: 500
+service:
+  pipelines:
+    metrics:
+      receivers:
+        - hostmetrics
+      processors:
+        - memory_limiter
+      exporters:
+        - logging
+  extensions:
+    - memory_ballast

internal/configconverter/testdata/ballast_mem_limiter_custom.yaml

@@ -0,0 +1,30 @@
+extensions:
+  memory_ballast:
+    size_mib: 64
+receivers:
+  hostmetrics:
+    collection_interval: 1s
+    scrapers:
+      cpu:
+processors:
+  memory_limiter/foo:
+    check_interval: 1s
+    limit_mib: 4000
+    spike_limit_mib: 800
+    ballast_size_mib: 64
+exporters:
+  logging:
+    loglevel: info
+    sampling_initial: 2
+    sampling_thereafter: 500
+service:
+  pipelines:
+    metrics:
+      receivers:
+        - hostmetrics
+      processors:
+        - memory_limiter/foo
+      exporters:
+        - logging
+  extensions:
+    - memory_ballast