Handle backwards compatibility of internal ballast removal (#759)

* Remove memory_limiter ballast_size_mib key from config if it exists.]\
* Generalize config converter
* Add no-convert-config flag

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