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

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.")
 	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/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/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/main.go

@@ -88,7 +88,14 @@ func main() {
 	srv := server.NewServer(log, allocator, cfg.ListenAddr, httpOptions...)
 
 	discoveryCtx, discoveryCancel := context.WithCancel(ctx)
-	discoveryManager = discovery.NewManager(discoveryCtx, nil, prometheus.NewRegistry(), nil)
+	// SD metrics must be non-nil; providers fail to register without them.
+	sdRegistry := prometheus.NewRegistry()
+	sdMetrics, sdMetricsErr := discovery.CreateAndRegisterSDMetrics(sdRegistry)
+	if sdMetricsErr != nil {
+		setupLog.Error(sdMetricsErr, "Unable to register service discovery metrics")
+		os.Exit(1)
+	}
+	discoveryManager = discovery.NewManager(discoveryCtx, nil, sdRegistry, sdMetrics)
 
 	targetDiscoverer = target.NewDiscoverer(log, discoveryManager, allocatorPrehook, srv)
 	collectorWatcher, collectorWatcherErr := collector.NewClient(log, cfg.ClusterConfig)

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

@@ -21,13 +21,16 @@ 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
 
 func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*PrometheusCRWatcher, error) {
@@ -49,11 +52,26 @@ 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 := os.Getenv("OTELCOL_NAMESPACE")
+	if collectorNamespace == "" {
+		if ns, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil && len(ns) > 0 {
+			collectorNamespace = string(ns)
+		} else {
+			collectorNamespace = defaultCollectorNamespace
+		}
+		logger.Info("OTELCOL_NAMESPACE not set, resolved namespace", "namespace", collectorNamespace)
+	}
 	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

@@ -6,6 +6,7 @@ package collector
 import (
 	"crypto/sha256"
 	"fmt"
+	"log/slog"
 
 	"github.com/aws/amazon-cloudwatch-agent-operator/apis/v1alpha1"
 )
@@ -28,7 +29,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 +52,27 @@ 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 {
+			// Static sentinel; Yaml() over map[string]interface{} rarely fails in practice.
+			slog.Warn("failed to serialize Spec.Prometheus for config hash", "error", err)
+			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)

internal/manifests/collector/annotations_test.go

@@ -79,3 +79,64 @@ func TestAnnotationsPropagateDown(t *testing.T) {
 	assert.Equal(t, "mycomponent", podAnnotations["myapp"])
 	assert.Equal(t, "pod_annotation_value", podAnnotations["pod_annotation"])
 }
+
+func promConfig(t *testing.T, replacement string) v1alpha1.PrometheusConfig {
+	t.Helper()
+	cfg := map[string]interface{}{
+		"scrape_configs": []interface{}{
+			map[string]interface{}{
+				"job_name": "kubernetes-pods-annotated",
+				"relabel_configs": []interface{}{
+					map[string]interface{}{
+						"target_label": "bug2probe",
+						"replacement":  replacement,
+					},
+				},
+			},
+		},
+	}
+	return v1alpha1.PrometheusConfig{
+		Config: &v1alpha1.AnyConfig{Object: cfg},
+	}
+}
+
+// TestPrometheusConfigChangeBumpsHash verifies a Prometheus-only change bumps the pod-template hash.
+func TestPrometheusConfigChangeBumpsHash(t *testing.T) {
+	base := v1alpha1.AmazonCloudWatchAgent{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"},
+		Spec: v1alpha1.AmazonCloudWatchAgentSpec{
+			Config:     "agent-config",
+			Prometheus: promConfig(t, "value2"),
+		},
+	}
+
+	// same spec twice -> stable hash
+	h1 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"]
+	h2 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"]
+	assert.Equal(t, h1, h2, "hash must be stable when nothing changes")
+
+	// change ONLY the prometheus config -> hash must change
+	changed := base
+	changed.Spec.Prometheus = promConfig(t, "value3")
+	h3 := PodAnnotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"]
+	assert.NotEqual(t, h1, h3, "pod annotation hash must change when only Spec.Prometheus changes")
+
+	// metadata annotations hash must also reflect the prometheus change
+	a1 := Annotations(base)["amazon-cloudwatch-agent-operator-config/sha256"]
+	a3 := Annotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"]
+	assert.NotEqual(t, a1, a3, "metadata annotation hash must change when only Spec.Prometheus changes")
+}
+
+// TestEmptyPrometheusHashUnchanged verifies non-Prometheus agents keep a stable hash.
+func TestEmptyPrometheusHashUnchanged(t *testing.T) {
+	otelcol := v1alpha1.AmazonCloudWatchAgent{
+		ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"},
+		Spec:       v1alpha1.AmazonCloudWatchAgentSpec{Config: "test"},
+	}
+
+	// sha256("test") == 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08
+	assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
+		Annotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"])
+	assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
+		PodAnnotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"])
+}