Improve config validation for prometheus receiver and target allocator (

#1729)

* Improve validation for prometheus receiver config

* Improve config validation for target allocator

* add changelog entry

* review fixes

* log empty Prometheus configs passed to discoverer

* add tests for ValidateConfig

.chloggen/fix_validate-ta-config.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. operator, target allocator, github action)
+component: operator
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Improve config validation for prometheus receiver and target allocator
+
+# One or more tracking issues related to the change
+issues: [1581]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

apis/v1alpha1/opentelemetrycollector_webhook.go

@@ -26,6 +26,7 @@ import (
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
+	"github.com/open-telemetry/opentelemetry-operator/pkg/featuregate"
 	ta "github.com/open-telemetry/opentelemetry-operator/pkg/targetallocator/adapters"
 )
 
@@ -160,7 +161,11 @@ func (r *OpenTelemetryCollector) validateCRDSpec() error {
 
 	// validate Prometheus config for target allocation
 	if r.Spec.TargetAllocator.Enabled {
-		_, err := ta.ConfigToPromConfig(r.Spec.Config)
+		promCfg, err := ta.ConfigToPromConfig(r.Spec.Config)
+		if err != nil {
+			return fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
+		}
+		err = ta.ValidatePromConfig(promCfg, r.Spec.TargetAllocator.Enabled, featuregate.EnableTargetAllocatorRewrite.IsEnabled())
 		if err != nil {
 			return fmt.Errorf("the OpenTelemetry Spec Prometheus configuration is incorrect, %w", err)
 		}

cmd/otel-allocator/config/config.go

@@ -129,3 +129,12 @@ func ParseCLI() (CLIConfig, error) {
 	cLIConf.ClusterConfig = clusterConfig
 	return cLIConf, nil
 }
+
+// ValidateConfig validates the cli and file configs together.
+func ValidateConfig(config *Config, cliConfig *CLIConfig) error {
+	scrapeConfigsPresent := (config.Config != nil && len(config.Config.ScrapeConfigs) > 0)
+	if !(*cliConfig.PromCRWatcherConf.Enabled || scrapeConfigsPresent) {
+		return fmt.Errorf("at least one scrape config must be defined, or Prometheus CR watching must be enabled")
+	}
+	return nil
+}

cmd/otel-allocator/config/config_test.go

@@ -91,6 +91,14 @@ func TestLoad(t *testing.T) {
 			},
 			wantErr: assert.NoError,
 		},
+		{
+			name: "no config",
+			args: args{
+				file: "./testdata/no_config.yaml",
+			},
+			want:    Config{},
+			wantErr: assert.NoError,
+		},
 		{
 			name: "service monitor pod monitor selector",
 			args: args{
@@ -157,3 +165,56 @@ func TestLoad(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateConfig(t *testing.T) {
+	enabled := true
+	disabled := false
+	testCases := []struct {
+		name        string
+		cliConfig   CLIConfig
+		fileConfig  Config
+		expectedErr error
+	}{
+		{
+			name:        "promCR enabled, no Prometheus config",
+			cliConfig:   CLIConfig{PromCRWatcherConf: PrometheusCRWatcherConfig{Enabled: &enabled}},
+			fileConfig:  Config{Config: nil},
+			expectedErr: nil,
+		},
+		{
+			name:        "promCR disabled, no Prometheus config",
+			cliConfig:   CLIConfig{PromCRWatcherConf: PrometheusCRWatcherConfig{Enabled: &disabled}},
+			fileConfig:  Config{Config: nil},
+			expectedErr: fmt.Errorf("at least one scrape config must be defined, or Prometheus CR watching must be enabled"),
+		},
+		{
+			name:        "promCR disabled, Prometheus config present, no scrapeConfigs",
+			cliConfig:   CLIConfig{PromCRWatcherConf: PrometheusCRWatcherConfig{Enabled: &disabled}},
+			fileConfig:  Config{Config: &promconfig.Config{}},
+			expectedErr: fmt.Errorf("at least one scrape config must be defined, or Prometheus CR watching must be enabled"),
+		},
+		{
+			name:      "promCR disabled, Prometheus config present, scrapeConfigs present",
+			cliConfig: CLIConfig{PromCRWatcherConf: PrometheusCRWatcherConfig{Enabled: &disabled}},
+			fileConfig: Config{
+				Config: &promconfig.Config{ScrapeConfigs: []*promconfig.ScrapeConfig{{}}},
+			},
+			expectedErr: nil,
+		},
+		{
+			name:      "promCR enabled, Prometheus config present, scrapeConfigs present",
+			cliConfig: CLIConfig{PromCRWatcherConf: PrometheusCRWatcherConfig{Enabled: &enabled}},
+			fileConfig: Config{
+				Config: &promconfig.Config{ScrapeConfigs: []*promconfig.ScrapeConfig{{}}},
+			},
+			expectedErr: nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := ValidateConfig(&tc.fileConfig, &tc.cliConfig)
+			assert.Equal(t, tc.expectedErr, err)
+		})
+	}
+}

cmd/otel-allocator/main.go

@@ -74,6 +74,10 @@ func main() {
 		setupLog.Error(configLoadErr, "Unable to load configuration")
 	}
 
+	if validationErr := config.ValidateConfig(&cfg, &cliConf); validationErr != nil {
+		setupLog.Error(validationErr, "Invalid configuration")
+	}
+
 	cliConf.RootLogger.Info("Starting the Target Allocator")
 	ctx := context.Background()
 	log := ctrl.Log.WithName("allocator")

cmd/otel-allocator/target/discovery.go

@@ -64,6 +64,10 @@ func NewDiscoverer(log logr.Logger, manager *discovery.Manager, hook discoveryHo
 }
 
 func (m *Discoverer) ApplyConfig(source allocatorWatcher.EventSource, cfg *config.Config) error {
+	if cfg == nil {
+		m.log.Info("Service Discovery got empty Prometheus config", "source", source.String())
+		return nil
+	}
 	m.configsMap[source] = cfg
 	jobToScrapeConfig := make(map[string]*config.ScrapeConfig)
 

cmd/otel-allocator/target/discovery_test.go

@@ -334,6 +334,23 @@ func TestDiscovery_ScrapeConfigHashing(t *testing.T) {
 	}
 }
 
+func TestDiscovery_NoConfig(t *testing.T) {
+	scu := &mockScrapeConfigUpdater{mockCfg: map[string]*promconfig.ScrapeConfig{}}
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	d := discovery.NewManager(ctx, gokitlog.NewNopLogger())
+	manager := NewDiscoverer(ctrl.Log.WithName("test"), d, nil, scu)
+	defer close(manager.close)
+	defer cancelFunc()
+
+	go func() {
+		err := d.Run()
+		assert.NoError(t, err)
+	}()
+	// check the updated scrape configs
+	expectedScrapeConfigs := map[string]*promconfig.ScrapeConfig{}
+	assert.Equal(t, expectedScrapeConfigs, scu.mockCfg)
+}
+
 func BenchmarkApplyScrapeConfig(b *testing.B) {
 	numConfigs := 1000
 	scrapeConfig := promconfig.ScrapeConfig{

pkg/collector/reconcile/config_replace.go

@@ -60,6 +60,11 @@ func ReplaceConfig(instance v1alpha1.OpenTelemetryCollector) (string, error) {
 		return "", getCfgPromErr
 	}
 
+	validateCfgPromErr := ta.ValidatePromConfig(promCfgMap, instance.Spec.TargetAllocator.Enabled, featuregate.EnableTargetAllocatorRewrite.IsEnabled())
+	if validateCfgPromErr != nil {
+		return "", validateCfgPromErr
+	}
+
 	// yaml marshaling/unsmarshaling is preferred because of the problems associated with the conversion of map to a struct using mapstructure
 	promCfg, marshalErr := yaml.Marshal(promCfgMap)
 	if marshalErr != nil {

pkg/collector/reconcile/configmap.go

@@ -103,7 +103,7 @@ func desiredTAConfigMap(params Params) (corev1.ConfigMap, error) {
 		labels["app.kubernetes.io/version"] = "latest"
 	}
 
-	promConfig, err := ta.ConfigToPromConfig(params.Instance.Spec.Config)
+	prometheusReceiverConfig, err := ta.ConfigToPromConfig(params.Instance.Spec.Config)
 	if err != nil {
 		return corev1.ConfigMap{}, err
 	}
@@ -114,8 +114,11 @@ func desiredTAConfigMap(params Params) (corev1.ConfigMap, error) {
 		"app.kubernetes.io/managed-by": "opentelemetry-operator",
 		"app.kubernetes.io/component":  "opentelemetry-collector",
 	}
-	// We only take the "config" from the returned object, we don't need the "target_allocator" configuration here.
-	taConfig["config"] = promConfig["config"]
+	// We only take the "config" from the returned object, if it's present
+	if prometheusConfig, ok := prometheusReceiverConfig["config"]; ok {
+		taConfig["config"] = prometheusConfig
+	}
+
 	if len(params.Instance.Spec.TargetAllocator.AllocationStrategy) > 0 {
 		taConfig["allocation_strategy"] = params.Instance.Spec.TargetAllocator.AllocationStrategy
 	} else {

pkg/targetallocator/adapters/config_to_prom_config.go

@@ -15,6 +15,7 @@
 package adapters
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/open-telemetry/opentelemetry-operator/pkg/collector/adapters"
@@ -57,3 +58,28 @@ func ConfigToPromConfig(cfg string) (map[interface{}]interface{}, error) {
 
 	return prometheus, nil
 }
+
+// ValidatePromConfig checks if the prometheus receiver config is valid given other collector-level settings.
+func ValidatePromConfig(config map[interface{}]interface{}, targetAllocatorEnabled bool, targetAllocatorRewriteEnabled bool) error {
+	_, promConfigExists := config["config"]
+
+	if targetAllocatorEnabled {
+		if targetAllocatorRewriteEnabled { // if rewrite is enabled, we will add a target_allocator section during rewrite
+			return nil
+		}
+		_, targetAllocatorExists := config["target_allocator"]
+
+		// otherwise, either the target_allocator or config section needs to be here
+		if !(promConfigExists || targetAllocatorExists) {
+			return errors.New("either target allocator or prometheus config needs to be present")
+		}
+
+		return nil
+	}
+	// if target allocator isn't enabled, we need a config section
+	if !promConfigExists {
+		return errorNoComponent("prometheusConfig")
+	}
+
+	return nil
+}

pkg/targetallocator/adapters/config_to_prom_config_test.go

@@ -15,6 +15,7 @@
 package adapters_test
 
 import (
+	"errors"
 	"fmt"
 	"reflect"
 	"testing"
@@ -114,3 +115,70 @@ func TestExtractPromConfigFromNullConfig(t *testing.T) {
 	// verify
 	assert.True(t, reflect.ValueOf(promConfig).IsNil())
 }
+
+func TestValidatePromConfig(t *testing.T) {
+	testCases := []struct {
+		description                   string
+		config                        map[interface{}]interface{}
+		targetAllocatorEnabled        bool
+		targetAllocatorRewriteEnabled bool
+		expectedError                 error
+	}{
+		{
+			description:                   "target_allocator and rewrite enabled",
+			config:                        map[interface{}]interface{}{},
+			targetAllocatorEnabled:        true,
+			targetAllocatorRewriteEnabled: true,
+			expectedError:                 nil,
+		},
+		{
+			description: "target_allocator enabled, target_allocator section present",
+			config: map[interface{}]interface{}{
+				"target_allocator": map[interface{}]interface{}{},
+			},
+			targetAllocatorEnabled:        true,
+			targetAllocatorRewriteEnabled: false,
+			expectedError:                 nil,
+		},
+		{
+			description: "target_allocator enabled, config section present",
+			config: map[interface{}]interface{}{
+				"config": map[interface{}]interface{}{},
+			},
+			targetAllocatorEnabled:        true,
+			targetAllocatorRewriteEnabled: false,
+			expectedError:                 nil,
+		},
+		{
+			description:                   "target_allocator enabled, neither section present",
+			config:                        map[interface{}]interface{}{},
+			targetAllocatorEnabled:        true,
+			targetAllocatorRewriteEnabled: false,
+			expectedError:                 errors.New("either target allocator or prometheus config needs to be present"),
+		},
+		{
+			description: "target_allocator disabled, config section present",
+			config: map[interface{}]interface{}{
+				"config": map[interface{}]interface{}{},
+			},
+			targetAllocatorEnabled:        false,
+			targetAllocatorRewriteEnabled: false,
+			expectedError:                 nil,
+		},
+		{
+			description:                   "target_allocator disabled, config section not present",
+			config:                        map[interface{}]interface{}{},
+			targetAllocatorEnabled:        false,
+			targetAllocatorRewriteEnabled: false,
+			expectedError:                 fmt.Errorf("no %s available as part of the configuration", "prometheusConfig"),
+		},
+	}
+
+	for _, testCase := range testCases {
+		testCase := testCase
+		t.Run(testCase.description, func(t *testing.T) {
+			err := ta.ValidatePromConfig(testCase.config, testCase.targetAllocatorEnabled, testCase.targetAllocatorRewriteEnabled)
+			assert.Equal(t, testCase.expectedError, err)
+		})
+	}
+}