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"
@@ -67,7 +68,7 @@ func main() {
 
 	// 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 {
@@ -79,16 +80,17 @@ func main() {
 		Version: version.Version,
 	}
 
-	parserProvider := configprovider.NewConfigSourceParserProvider(
+	cfgSourcePP := configprovider.NewConfigSourceParserProvider(
 		newBaseParserProvider(),
 		zap.NewNop(), // The service logger is not available yet, setting it to NoP.
 		info,
 		configsources.Get()...,
 	)
+	configConverterPP := configconverter.ParserProvider(cfgSourcePP, configconverter.RemoveBallastKey)
 	serviceParams := service.CollectorSettings{
 		BuildInfo:      info,
 		Factories:      factories,
-		ParserProvider: parserProvider,
+		ParserProvider: configConverterPP,
 	}
 
 	if err := run(serviceParams); err != nil {
@@ -256,7 +258,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 +270,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 +284,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 +302,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)
 		}
 	}
 }

internal/configconverter/ballast_remover.go

@@ -0,0 +1,45 @@
+// 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"
+)
+
+var ballastKeyRegexp *regexp.Regexp
+
+func init() {
+	const expr = "processors::memory_limiter(/\\w+)?::ballast_size_mib"
+	ballastKeyRegexp, _ = regexp.Compile(expr)
+}
+
+// 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 {
+	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