Fix Target Allocator startup crashes, PrometheusCR watcher, and Prometheus config pod restart

cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go

@@ -6,6 +6,7 @@ package collector
 import (
 	"context"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -26,7 +27,7 @@ const (
 )
 
 var (
-	ns                   = os.Getenv("OTELCOL_NAMESPACE")
+	ns                   = strings.TrimSpace(os.Getenv("OTELCOL_NAMESPACE"))
 	collectorsDiscovered = promauto.NewGauge(prometheus.GaugeOpts{
 		Name: "amazon_cloudwatch_agent_allocator_collectors_discovered",
 		Help: "Number of collectors discovered.",

cmd/amazon-cloudwatch-agent-target-allocator/config/config.go

@@ -119,6 +119,13 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error {
 		return err
 	}
 
+	// OR the CLI flag into the YAML value so either source can enable the watcher.
+	prometheusCREnabled, err := getPrometheusCREnabled(flagSet)
+	if err != nil {
+		return err
+	}
+	target.PrometheusCR.Enabled = target.PrometheusCR.Enabled || prometheusCREnabled
+
 	target.HTTPS.Enabled, err = getHttpsEnabled(flagSet)
 	if err != nil {
 		return err

cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go

@@ -34,6 +34,7 @@ func getFlagSet(errorHandling pflag.ErrorHandling) *pflag.FlagSet {
 	flagSet := pflag.NewFlagSet(targetAllocatorName, errorHandling)
 	flagSet.String(configFilePathFlagName, DefaultConfigFilePath, "The path to the config file.")
 	flagSet.String(kubeConfigPathFlagName, filepath.Join(homedir.HomeDir(), ".kube", "config"), "absolute path to the KubeconfigPath file")
+	flagSet.Bool(prometheusCREnabledFlagName, false, "Enable watching of Prometheus Operator custom resources (ServiceMonitor/PodMonitor) to dynamically generate scrape configs. Enable-only: this flag is OR-ed with prometheus_cr.enabled from the config file, so passing =false cannot disable a watcher the config file enables.")
 	flagSet.Bool(reloadConfigFlagName, false, "Enable automatic configuration reloading. This functionality is deprecated and will be removed in a future release.")
 	flagSet.Bool(httpsEnabledFlagName, true, "Enable HTTPS additional server")
 	flagSet.String(listenAddrHttpsFlagName, ":8443", "The address where this service serves over HTTPS.")
@@ -58,6 +59,10 @@ func getConfigReloadEnabled(flagSet *pflag.FlagSet) (bool, error) {
 	return flagSet.GetBool(reloadConfigFlagName)
 }
 
+func getPrometheusCREnabled(flagSet *pflag.FlagSet) (bool, error) {
+	return flagSet.GetBool(prometheusCREnabledFlagName)
+}
+
 func getHttpsListenAddr(flagSet *pflag.FlagSet) (string, error) {
 	return flagSet.GetString(listenAddrHttpsFlagName)
 }

cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go

@@ -17,6 +17,7 @@ func TestGetFlagSet(t *testing.T) {
 	// Check if each flag exists
 	assert.NotNil(t, fs.Lookup(configFilePathFlagName), "Flag %s not found", configFilePathFlagName)
 	assert.NotNil(t, fs.Lookup(kubeConfigPathFlagName), "Flag %s not found", kubeConfigPathFlagName)
+	assert.NotNil(t, fs.Lookup(prometheusCREnabledFlagName), "Flag %s not found", prometheusCREnabledFlagName)
 }
 
 func TestFlagGetters(t *testing.T) {
@@ -45,6 +46,18 @@ func TestFlagGetters(t *testing.T) {
 			expectedValue: true,
 			getterFunc:    func(fs *pflag.FlagSet) (interface{}, error) { return getConfigReloadEnabled(fs) },
 		},
+		{
+			name:          "GetPrometheusCREnabled",
+			flagArgs:      []string{"--" + prometheusCREnabledFlagName},
+			expectedValue: true,
+			getterFunc:    func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) },
+		},
+		{
+			name:          "GetPrometheusCREnabledDefault",
+			flagArgs:      []string{},
+			expectedValue: false,
+			getterFunc:    func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) },
+		},
 		{
 			name:        "InvalidFlag",
 			flagArgs:    []string{"--invalid-flag", "value"},

cmd/amazon-cloudwatch-agent-target-allocator/config/prometheus_cr_or_semantics_test.go

@@ -0,0 +1,36 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+
+	"github.com/spf13/pflag"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPrometheusCREnabledOrSemantics(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		yaml bool
+		args []string
+		want bool
+	}{
+		{"yamlFalse_cliAbsent", false, nil, false},
+		{"yamlFalse_cliTrue", false, []string{"--enable-prometheus-cr-watcher"}, true},
+		{"yamlTrue_cliAbsent", true, nil, true},
+		{"yamlTrue_cliFalse", true, []string{"--enable-prometheus-cr-watcher=false"}, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			fs := getFlagSet(pflag.ContinueOnError)
+			args := append([]string{"--kubeconfig-path", "./testdata/kubeconfig_test.yaml"}, tc.args...)
+			require.NoError(t, fs.Parse(args))
+			c := CreateDefaultConfig()
+			c.PrometheusCR.Enabled = tc.yaml
+			require.NoError(t, LoadFromCLI(&c, fs))
+			assert.Equal(t, tc.want, c.PrometheusCR.Enabled)
+		})
+	}
+}

cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go

@@ -0,0 +1,36 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestScrapeProtocolsDefaultedOnLoad guards against "scrape_protocols cannot be empty" regression.
+func TestScrapeProtocolsDefaultedOnLoad(t *testing.T) {
+	got := CreateDefaultConfig()
+	err := LoadFromFile("./testdata/scrape_protocols_omitted_test.yaml", &got)
+	require.NoError(t, err)
+
+	require.NotNil(t, got.PromConfig)
+	require.NotEmpty(t, got.PromConfig.ScrapeConfigs)
+
+	for _, sc := range got.PromConfig.ScrapeConfigs {
+		assert.NotEmpty(t, sc.ScrapeProtocols,
+			"scrape_protocols must be defaulted for job %q", sc.JobName)
+	}
+
+	// Spot-check the specific static job from the reproduction by name.
+	found := false
+	for _, sc := range got.PromConfig.ScrapeConfigs {
+		if sc.JobName == "prometheus-sample-app" {
+			found = true
+			assert.Greater(t, len(sc.ScrapeProtocols), 0)
+		}
+	}
+	require.True(t, found, "expected the 'prometheus-sample-app' job to be present")
+}

cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/kubeconfig_test.yaml

@@ -0,0 +1,16 @@
+apiVersion: v1
+kind: Config
+clusters:
+- name: test-cluster
+  cluster:
+    server: https://127.0.0.1:6443
+contexts:
+- name: test-context
+  context:
+    cluster: test-cluster
+    user: test-user
+current-context: test-context
+users:
+- name: test-user
+  user:
+    token: test-token

cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml

@@ -0,0 +1,14 @@
+# Regression fixture: a scrape job that omits scrape_protocols.
+# The TA config-load path must default scrape_protocols so the agent does not
+# reject the config with "scrape_protocols cannot be empty".
+config:
+  scrape_configs:
+    - job_name: prometheus-sample-app
+      kubernetes_sd_configs:
+        - role: pod
+      relabel_configs:
+        - source_labels: [__meta_kubernetes_pod_label_app]
+          regex: prometheus-sample-app
+          action: keep
+        - target_label: label1
+          replacement: value1

cmd/amazon-cloudwatch-agent-target-allocator/watcher/namespace_resolution_test.go

@@ -0,0 +1,69 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package watcher
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/prometheus/common/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/client-go/rest"
+
+	allocatorconfig "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/config"
+)
+
+func TestResolveCollectorNamespace(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		env      string
+		fileBody *string
+		want     string
+	}{
+		{"env var set", "my-env-ns", nil, "my-env-ns"},
+		{"env var set with whitespace", "  my-env-ns\n", nil, "my-env-ns"},
+		{"env whitespace only falls through to file", "   \n", ptrStr("file-ns"), "file-ns"},
+		{"env empty, file set", "", ptrStr("file-ns"), "file-ns"},
+		{"env empty, file trailing newline", "", ptrStr("file-ns\n"), "file-ns"},
+		{"env empty, file whitespace only", "", ptrStr("  \n\t"), defaultCollectorNamespace},
+		{"both missing falls back to default", "", nil, defaultCollectorNamespace},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv("OTELCOL_NAMESPACE", tc.env)
+			orig := serviceAccountNamespacePath
+			t.Cleanup(func() { serviceAccountNamespacePath = orig })
+			if tc.fileBody == nil {
+				serviceAccountNamespacePath = filepath.Join(t.TempDir(), "does-not-exist")
+			} else {
+				p := filepath.Join(t.TempDir(), "namespace")
+				require.NoError(t, os.WriteFile(p, []byte(*tc.fileBody), 0o600))
+				serviceAccountNamespacePath = p
+			}
+			assert.Equal(t, tc.want, resolveCollectorNamespace(logr.Discard()))
+		})
+	}
+}
+
+func ptrStr(s string) *string { return &s }
+
+func TestNewPrometheusCRWatcherDefaultsEvaluationInterval(t *testing.T) {
+	t.Setenv("OTELCOL_NAMESPACE", "eval-ns")
+	cfg := allocatorconfig.Config{
+		ClusterConfig: &rest.Config{Host: "https://127.0.0.1:65535"},
+		PrometheusCR: allocatorconfig.PrometheusCRConfig{
+			Enabled:        true,
+			ScrapeInterval: model.Duration(45 * time.Second),
+		},
+	}
+	w, err := NewPrometheusCRWatcher(logr.Discard(), cfg)
+	require.NoError(t, err)
+	require.NotNil(t, w.prom)
+	assert.Equal(t, "eval-ns", w.prom.Namespace)
+	assert.Equal(t, w.prom.Spec.ScrapeInterval, w.prom.Spec.EvaluationInterval)
+	assert.Equal(t, "45s", string(w.prom.Spec.EvaluationInterval))
+}

cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"log/slog"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -21,15 +22,37 @@ import (
 	kubeDiscovery "github.com/prometheus/prometheus/discovery/kubernetes"
 	"gopkg.in/yaml.v2"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 
 	allocatorconfig "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/config"
 )
 
+const defaultCollectorNamespace = "amazon-cloudwatch"
+
 const minEventInterval = time.Second * 5
 
+const defaultServiceAccountNamespacePath = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
+
+// serviceAccountNamespacePath is a variable so tests can redirect the read.
+var serviceAccountNamespacePath = defaultServiceAccountNamespacePath
+
+func resolveCollectorNamespace(logger logr.Logger) string {
+	if ns := strings.TrimSpace(os.Getenv("OTELCOL_NAMESPACE")); ns != "" {
+		return ns
+	}
+	namespace := defaultCollectorNamespace
+	if data, err := os.ReadFile(serviceAccountNamespacePath); err == nil {
+		if ns := strings.TrimSpace(string(data)); ns != "" {
+			namespace = ns
+		}
+	}
+	logger.Info("OTELCOL_NAMESPACE not set, resolved namespace", "namespace", namespace)
+	return namespace
+}
+
 func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*PrometheusCRWatcher, error) {
 	mClient, err := monitoringclient.NewForConfig(cfg.ClusterConfig)
 	if err != nil {
@@ -49,11 +72,18 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr
 	}
 
 	// TODO: We should make these durations configurable
+	// Namespace must be non-empty; the config generator panics otherwise.
+	collectorNamespace := resolveCollectorNamespace(logger)
 	prom := &monitoringv1.Prometheus{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: collectorNamespace,
+		},
 		Spec: monitoringv1.PrometheusSpec{
 			CommonPrometheusFields: monitoringv1.CommonPrometheusFields{
 				ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()),
 			},
+			// Must be non-empty; default to scrape interval.
+			EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()),
 		},
 	}
 

internal/manifests/collector/annotations.go

@@ -28,7 +28,7 @@ func Annotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string {
 	}
 
 	// make sure sha256 for configMap is always calculated
-	annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config)
+	annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance))
 
 	return annotations
 }
@@ -51,11 +51,31 @@ func PodAnnotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string {
 	}
 
 	// make sure sha256 for configMap is always calculated
-	podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config)
+	podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance))
 
 	return podAnnotations
 }
 
+// configHashInput returns the combined config string used for the pod-template restart hash.
+func configHashInput(instance v1alpha1.AmazonCloudWatchAgent) string {
+	config := instance.Spec.Config
+	if !instance.Spec.Prometheus.IsEmpty() {
+		promYaml, err := instance.Spec.Prometheus.Yaml()
+		if err != nil {
+			// Unreachable from the CRD path: Config.Object only ever holds JSON-decoded
+			// types, all of which gopkg.in/yaml.v3 encodes. Deliberately a constant: two
+			// different unserializable specs hash equal, so a Prometheus-only change would
+			// not roll pods while the error persists. Hash stability across operator
+			// restarts depends on gopkg.in/yaml.v3 (v3.0.1) sorting map keys
+			// (encode.go:189, and :242 for the inline-map branch).
+			config += "\x00prometheus-serialize-error"
+		} else {
+			config += "\x00" + promYaml // null byte prevents collision between config suffix and promYAML prefix
+		}
+	}
+	return config
+}
+
 func getConfigMapSHA(config string) string {
 	h := sha256.Sum256([]byte(config))
 	return fmt.Sprintf("%x", h)