From 13764510b87ac3a5894ed534c8d28d5bd982dc6b Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 09:21:01 +0000 Subject: [PATCH 01/10] Register enable-prometheus-cr-watcher flag and fix PrometheusCR watcher startup The target-allocator declared the enable-prometheus-cr-watcher flag name as a constant but never registered it on the flag set, while the operator passes --enable-prometheus-cr-watcher whenever PrometheusCR.enabled is true. Because args are parsed with pflag.ExitOnError, the unregistered flag caused the binary to print 'unknown flag' and exit(2), putting the target-allocator pod into CrashLoopBackOff. This change registers the flag and ORs it with the YAML prometheus_cr.enabled setting, then fixes three latent defects that were previously unreachable because the binary crashed first: - promOperator: set a non-empty Namespace on the synthetic Prometheus object so the prometheus-operator config generator no longer panics with 'namespace can't be empty' in store.ForNamespace. - promOperator: set EvaluationInterval so the generated config does not render an empty global.evaluation_interval, which the prometheus config parser rejects with 'empty duration string'. - main: create and register service-discovery metrics and pass them to discovery.NewManager; passing a nil sdMetrics map makes every SD provider fail to register, yielding zero discovered targets. RELEASE_NOTES updated. --- RELEASE_NOTES | 6 ++++++ .../config/config.go | 9 +++++++++ .../config/flags.go | 5 +++++ .../config/flags_test.go | 13 ++++++++++++ .../main.go | 13 +++++++++++- .../watcher/promOperator.go | 20 +++++++++++++++++++ 6 files changed, 65 insertions(+), 1 deletion(-) diff --git a/RELEASE_NOTES b/RELEASE_NOTES index 6128a2210..ff73e4d26 100644 --- a/RELEASE_NOTES +++ b/RELEASE_NOTES @@ -1,3 +1,9 @@ +======================================================================== +Amazon CloudWatch Agent Operator (Unreleased) +======================================================================== +Bug Fixes: +* [TargetAllocator] Register the `--enable-prometheus-cr-watcher` CLI flag so the target-allocator no longer exits with "unknown flag" (CrashLoopBackOff) when the operator passes it for PrometheusCR.enabled. The flag is OR'd with the YAML `prometheus_cr.enabled` setting. + ======================================================================== Amazon CloudWatch Agent Operator v3.5.0 (2026-06-05) ======================================================================== diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 2272eedbe..6b2a3d81d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -119,6 +119,15 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } + // Enable the Prometheus CR watcher when requested via the CLI flag. The YAML + // `prometheus_cr.enabled` value is loaded before this point, so OR the flag in + // rather than overwriting it: the watcher is enabled if either source sets it. + 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 diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go index caf639ed9..294f9bfe6 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go +++ b/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) } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go index 8beed29d0..66c71a268 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go +++ b/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"}, diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/main.go b/cmd/amazon-cloudwatch-agent-target-allocator/main.go index dbc956ebf..57185184d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/main.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/main.go @@ -88,7 +88,18 @@ func main() { srv := server.NewServer(log, allocator, cfg.ListenAddr, httpOptions...) discoveryCtx, discoveryCancel := context.WithCancel(ctx) - discoveryManager = discovery.NewManager(discoveryCtx, nil, prometheus.NewRegistry(), nil) + // Service Discovery metrics MUST be created and passed to the discovery + // manager. Each SD provider's NewDiscoverer fails when its DiscovererMetrics + // is nil, so passing a nil sdMetrics map makes every provider (including + // kubernetes_sd) fail to register, yielding zero discovered targets. Mirror + // the upstream opentelemetry-operator target-allocator. + 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) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 2651e8d29..a15db9d78 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -21,6 +21,7 @@ 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" @@ -49,11 +50,30 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable + // The synthetic Prometheus object must carry a non-empty namespace: the + // prometheus-operator config generator calls store.ForNamespace(prom.Namespace) + // when generating the server configuration, which panics on an empty namespace + // ("namespace can't be empty"). Mirror the upstream opentelemetry-operator + // target-allocator by setting it to the collector/TA namespace, derived from the + // operator-injected OTELCOL_NAMESPACE downward-API env (fallback: amazon-cloudwatch). + collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") + if collectorNamespace == "" { + collectorNamespace = "amazon-cloudwatch" + } prom := &monitoringv1.Prometheus{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: collectorNamespace, + }, Spec: monitoringv1.PrometheusSpec{ CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, + // EvaluationInterval must be non-empty: the prometheus-operator config + // generator renders it verbatim as global.evaluation_interval, and the + // downstream prometheus config parser rejects an empty value with + // "empty duration string". The CWA TA does not evaluate rules, so mirror + // the scrape interval as a sane non-empty default. + EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } From 0405f4decb019dd2ee173333a3985ea517b4ffb5 Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 15:57:02 +0000 Subject: [PATCH 02/10] test(target-allocator): guard scrape_protocols defaulting on config load Add a regression test asserting that loading a Target Allocator config whose static scrape job omits scrape_protocols still yields a non-empty ScrapeProtocols on every loaded scrape config. This is defaulted by the pinned Prometheus library during yaml.UnmarshalStrict into the prometheus Config type, so the distributed /scrape_configs payload is never empty and the agent's prometheus-receiver validation passes. The test fails fast if a future dependency or load-path change drops this defaulting. --- .../scrape_protocols_regression_test.go | 36 +++++++++++++++++++ .../scrape_protocols_omitted_test.yaml | 14 ++++++++ 2 files changed, 50 insertions(+) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go new file mode 100644 index 000000000..9c432d7ee --- /dev/null +++ b/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") +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml new file mode 100644 index 000000000..9405f61e8 --- /dev/null +++ b/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 From d61d693fa6f0536786951e8e4d08f378a5e77c82 Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 16:01:54 +0000 Subject: [PATCH 03/10] fix(collector): roll pods when Prometheus config changes The pod-template restart-trigger sha256 was computed from Spec.Config only, so a change to Spec.Prometheus (rendered into a separate ConfigMap) left the pod template byte-identical and the workload controller did not roll the pods. Fold the serialized Spec.Prometheus (PrometheusConfig.Yaml()) into the hash input when it is non-empty, so a Prometheus-only change bumps the pod-template annotation and triggers a rolling restart, matching agent-config behavior. When no Prometheus config is set the hash input is byte-identical to the agent config alone, leaving non-Prometheus agents unaffected. --- RELEASE_NOTES | 6 -- .../config/config.go | 4 +- .../main.go | 6 +- .../watcher/promOperator.go | 22 +++---- internal/manifests/collector/annotations.go | 21 ++++++- .../manifests/collector/annotations_test.go | 61 +++++++++++++++++++ 6 files changed, 92 insertions(+), 28 deletions(-) diff --git a/RELEASE_NOTES b/RELEASE_NOTES index ff73e4d26..6128a2210 100644 --- a/RELEASE_NOTES +++ b/RELEASE_NOTES @@ -1,9 +1,3 @@ -======================================================================== -Amazon CloudWatch Agent Operator (Unreleased) -======================================================================== -Bug Fixes: -* [TargetAllocator] Register the `--enable-prometheus-cr-watcher` CLI flag so the target-allocator no longer exits with "unknown flag" (CrashLoopBackOff) when the operator passes it for PrometheusCR.enabled. The flag is OR'd with the YAML `prometheus_cr.enabled` setting. - ======================================================================== Amazon CloudWatch Agent Operator v3.5.0 (2026-06-05) ======================================================================== diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 6b2a3d81d..081e646e9 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -119,9 +119,7 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } - // Enable the Prometheus CR watcher when requested via the CLI flag. The YAML - // `prometheus_cr.enabled` value is loaded before this point, so OR the flag in - // rather than overwriting it: the watcher is enabled if either source sets it. + // OR the CLI flag into the YAML value so either source can enable the watcher. prometheusCREnabled, err := getPrometheusCREnabled(flagSet) if err != nil { return err diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/main.go b/cmd/amazon-cloudwatch-agent-target-allocator/main.go index 57185184d..2b553dce6 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/main.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/main.go @@ -88,11 +88,7 @@ func main() { srv := server.NewServer(log, allocator, cfg.ListenAddr, httpOptions...) discoveryCtx, discoveryCancel := context.WithCancel(ctx) - // Service Discovery metrics MUST be created and passed to the discovery - // manager. Each SD provider's NewDiscoverer fails when its DiscovererMetrics - // is nil, so passing a nil sdMetrics map makes every provider (including - // kubernetes_sd) fail to register, yielding zero discovered targets. Mirror - // the upstream opentelemetry-operator target-allocator. + // SD metrics must be non-nil; providers fail to register without them. sdRegistry := prometheus.NewRegistry() sdMetrics, sdMetricsErr := discovery.CreateAndRegisterSDMetrics(sdRegistry) if sdMetricsErr != nil { diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index a15db9d78..5dfc314dd 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -29,6 +29,8 @@ import ( 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) { @@ -50,15 +52,15 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable - // The synthetic Prometheus object must carry a non-empty namespace: the - // prometheus-operator config generator calls store.ForNamespace(prom.Namespace) - // when generating the server configuration, which panics on an empty namespace - // ("namespace can't be empty"). Mirror the upstream opentelemetry-operator - // target-allocator by setting it to the collector/TA namespace, derived from the - // operator-injected OTELCOL_NAMESPACE downward-API env (fallback: amazon-cloudwatch). + // Namespace must be non-empty; the config generator panics otherwise. collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") if collectorNamespace == "" { - collectorNamespace = "amazon-cloudwatch" + 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{ @@ -68,11 +70,7 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, - // EvaluationInterval must be non-empty: the prometheus-operator config - // generator renders it verbatim as global.evaluation_interval, and the - // downstream prometheus config parser rejects an empty value with - // "empty duration string". The CWA TA does not evaluate rules, so mirror - // the scrape interval as a sane non-empty default. + // Must be non-empty; default to scrape interval. EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } diff --git a/internal/manifests/collector/annotations.go b/internal/manifests/collector/annotations.go index a665e465e..364072330 100644 --- a/internal/manifests/collector/annotations.go +++ b/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) diff --git a/internal/manifests/collector/annotations_test.go b/internal/manifests/collector/annotations_test.go index e02ef1678..1ab8e3bd7 100644 --- a/internal/manifests/collector/annotations_test.go +++ b/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"]) +} From 634bcefad0911110c3fdef1b3c3f1a1db119ce63 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Thu, 25 Jun 2026 16:59:45 +0100 Subject: [PATCH 04/10] feat(target-allocator): tolerate missing ServiceMonitor/PodMonitor CRDs Start each monitor informer lazily, driven by a watch on the ServiceMonitor/PodMonitor CustomResourceDefinitions, instead of building both informers eagerly and blocking on cache sync at startup. The Target Allocator now starts and stays healthy whether or not the CRDs are installed, and begins (or stops) watching each type automatically as its CRD is created or deleted -- no restart required. - NewPrometheusCRWatcher no longer builds informers; it adds an apiextensions client and empty per-type informer/stop-channel maps. - startMonitorInformer/stopMonitorInformer manage each type independently (fresh factory per start so a deleted-then-recreated CRD restarts cleanly). - watchCRDs starts/stops informers on CRD add/delete; Watch never fails just because a CRD is absent. - LoadConfig guards against absent informers under a read lock. - Close stops all per-type informers. Implicitly gated on prometheusCR.enabled (the watcher is only constructed when that is set), so behavior is unchanged when CR watching is off. Promotes k8s.io/apiextensions-apiserver to a direct dependency. --- .../watcher/promOperator.go | 317 ++++++++++++++---- .../watcher/promOperator_test.go | 212 ++++++++++-- go.mod | 2 +- 3 files changed, 440 insertions(+), 91 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 5dfc314dd..47be05a3d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -8,6 +8,7 @@ import ( "fmt" "log/slog" "os" + "sync" "time" "github.com/go-logr/logr" @@ -20,9 +21,14 @@ import ( promconfig "github.com/prometheus/prometheus/config" kubeDiscovery "github.com/prometheus/prometheus/discovery/kubernetes" "gopkg.in/yaml.v2" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + apiextensionsinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions" v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" @@ -33,6 +39,22 @@ const defaultCollectorNamespace = "amazon-cloudwatch" const minEventInterval = time.Second * 5 +// monitoringResources maps the prometheus-operator resource name to the +// GroupVersionResource used to build its informer. +var monitoringResources = map[string]schema.GroupVersionResource{ + monitoringv1.ServiceMonitorName: monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.ServiceMonitorName), + monitoringv1.PodMonitorName: monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.PodMonitorName), +} + +// crdNameToResource maps a CustomResourceDefinition object name to the +// prometheus-operator resource key whose informer it backs. The TA watches +// these CRDs so it can start/stop each informer independently as the CRD +// appears or disappears, rather than requiring both CRDs to exist at startup. +var crdNameToResource = map[string]string{ + monitoringv1.ServiceMonitorName + "." + monitoringv1.SchemeGroupVersion.Group: monitoringv1.ServiceMonitorName, + monitoringv1.PodMonitorName + "." + monitoringv1.SchemeGroupVersion.Group: monitoringv1.PodMonitorName, +} + func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*PrometheusCRWatcher, error) { mClient, err := monitoringclient.NewForConfig(cfg.ClusterConfig) if err != nil { @@ -44,9 +66,7 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr return nil, err } - factory := informers.NewMonitoringInformerFactories(map[string]struct{}{v1.NamespaceAll: {}}, map[string]struct{}{}, mClient, allocatorconfig.DefaultResyncTime, nil) //TODO decide what strategy to use regarding namespaces - - monitoringInformers, err := getInformers(factory) + crdClient, err := apiextensionsclient.NewForConfig(cfg.ClusterConfig) if err != nil { return nil, err } @@ -86,11 +106,17 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr podMonSelector := getSelector(cfg.PodMonitorSelector) + // Informers are NOT built here. Each ServiceMonitor/PodMonitor informer is + // started lazily, only once its CRD is observed to exist (see Watch). This + // lets the TA start and run healthily whether or not the CRDs are present, + // and pick up each type independently when its CRD appears. return &PrometheusCRWatcher{ logger: logger, kubeMonitoringClient: mClient, k8sClient: clientset, - informers: monitoringInformers, + crdClient: crdClient, + informers: map[string]*informers.ForResource{}, + informerStopChannels: map[string]chan struct{}{}, stopChannel: make(chan struct{}), eventInterval: minEventInterval, configGenerator: generator, @@ -105,13 +131,19 @@ type PrometheusCRWatcher struct { logger logr.Logger kubeMonitoringClient monitoringclient.Interface k8sClient kubernetes.Interface - informers map[string]*informers.ForResource + crdClient apiextensionsclient.Interface eventInterval time.Duration stopChannel chan struct{} configGenerator *prometheus.ConfigGenerator prom *monitoringv1.Prometheus kubeConfigPath string + // informersMtx guards informers and informerStopChannels, which are mutated + // from CRD-watch event handlers (separate goroutines) and read by LoadConfig. + informersMtx sync.RWMutex + informers map[string]*informers.ForResource + informerStopChannels map[string]chan struct{} + serviceMonitorSelector labels.Selector podMonitorSelector labels.Selector } @@ -123,65 +155,198 @@ func getSelector(s map[string]string) labels.Selector { return labels.SelectorFromSet(s) } -// getInformers returns a map of informers for the given resources. -func getInformers(factory informers.FactoriesForNamespaces) (map[string]*informers.ForResource, error) { - serviceMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.ServiceMonitorName)) +// newMonitoringFactory builds a fresh monitoring informer factory. A new factory +// is created per informer start so that a stopped informer (its CRD was deleted) +// can be cleanly restarted later if the CRD is recreated — shared informers +// cannot be restarted once their stop channel is closed. +func (w *PrometheusCRWatcher) newMonitoringFactory() informers.FactoriesForNamespaces { + return informers.NewMonitoringInformerFactories( + map[string]struct{}{v1.NamespaceAll: {}}, + map[string]struct{}{}, + w.kubeMonitoringClient, + allocatorconfig.DefaultResyncTime, + nil, + ) //TODO decide what strategy to use regarding namespaces +} + +// crdExists reports whether the named CustomResourceDefinition is currently +// registered in the cluster. +func (w *PrometheusCRWatcher) crdExists(ctx context.Context, crdName string) (bool, error) { + _, err := w.crdClient.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, crdName, metav1.GetOptions{}) if err != nil { - return nil, err + if apierrors.IsNotFound(err) { + return false, nil + } + return false, err + } + return true, nil +} + +// startMonitorInformer builds and starts the informer for a single monitoring +// resource (ServiceMonitor or PodMonitor), waits for its cache to sync, wires up +// the notification handler, and records it for use by LoadConfig. It is +// idempotent: calling it for an already-running informer is a no-op. +func (w *PrometheusCRWatcher) startMonitorInformer(resourceName string, notifyEvents chan struct{}) error { + w.informersMtx.Lock() + defer w.informersMtx.Unlock() + + if _, running := w.informers[resourceName]; running { + return nil + } + + gvr, ok := monitoringResources[resourceName] + if !ok { + return fmt.Errorf("unknown monitoring resource %q", resourceName) } - podMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.PodMonitorName)) + informer, err := informers.NewInformersForResource(w.newMonitoringFactory(), gvr) if err != nil { - return nil, err + return err } - return map[string]*informers.ForResource{ - monitoringv1.ServiceMonitorName: serviceMonitorInformers, - monitoringv1.PodMonitorName: podMonitorInformers, - }, nil + stopCh := make(chan struct{}) + informer.Start(stopCh) + if ok := cache.WaitForNamedCacheSync(resourceName, stopCh, informer.HasSynced); !ok { + close(stopCh) + return fmt.Errorf("failed to sync %s informer cache", resourceName) + } + + informer.AddEventHandler(notifyHandler(notifyEvents)) + + w.informers[resourceName] = informer + w.informerStopChannels[resourceName] = stopCh + + // A new resource type just became available; trigger a config reload. + notify(notifyEvents) + return nil } -// Watch wrapped informers and wait for an initial sync. -func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors chan error) error { - success := true - // this channel needs to be buffered because notifications are asynchronous and neither producers nor consumers wait - notifyEvents := make(chan struct{}, 1) +// stopMonitorInformer stops the informer for a single monitoring resource (its +// CRD was removed) and drops it from the active set so its targets are no longer +// generated. It is idempotent. +func (w *PrometheusCRWatcher) stopMonitorInformer(resourceName string, notifyEvents chan struct{}) { + w.informersMtx.Lock() + defer w.informersMtx.Unlock() - for name, resource := range w.informers { - resource.Start(w.stopChannel) + stopCh, running := w.informerStopChannels[resourceName] + if !running { + return + } + close(stopCh) + delete(w.informerStopChannels, resourceName) + delete(w.informers, resourceName) + + // The type went away; trigger a config reload so its targets are dropped. + notify(notifyEvents) +} - if ok := cache.WaitForNamedCacheSync(name, w.stopChannel, resource.HasSynced); !ok { - success = false +// crdObjectName extracts the CustomResourceDefinition name from an informer +// event object, tolerating the tombstone wrapper delivered on some deletes. +func crdObjectName(obj interface{}) (string, bool) { + switch t := obj.(type) { + case *apiextensionsv1.CustomResourceDefinition: + return t.Name, true + case cache.DeletedFinalStateUnknown: + if crd, ok := t.Obj.(*apiextensionsv1.CustomResourceDefinition); ok { + return crd.Name, true } + } + return "", false +} - // only send an event notification if there isn't one already - resource.AddEventHandler(cache.ResourceEventHandlerFuncs{ - // these functions only write to the notification channel if it's empty to avoid blocking - // if scrape config updates are being rate-limited - AddFunc: func(obj interface{}) { - select { - case notifyEvents <- struct{}{}: - default: - } - }, - UpdateFunc: func(oldObj, newObj interface{}) { - select { - case notifyEvents <- struct{}{}: - default: - } - }, - DeleteFunc: func(obj interface{}) { - select { - case notifyEvents <- struct{}{}: - default: - } - }, - }) +// watchCRDs starts a CustomResourceDefinition informer that starts/stops the +// per-type monitoring informers as the ServiceMonitor/PodMonitor CRDs are +// created or deleted at runtime — no TA restart required. The informer replays +// existing CRDs on its initial sync, so CRDs present at startup are handled here +// too (startMonitorInformer is idempotent). +func (w *PrometheusCRWatcher) watchCRDs(notifyEvents chan struct{}) { + factory := apiextensionsinformers.NewSharedInformerFactory(w.crdClient, allocatorconfig.DefaultResyncTime) + crdInformer := factory.Apiextensions().V1().CustomResourceDefinitions().Informer() + _, _ = crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + name, ok := crdObjectName(obj) + if !ok { + return + } + resourceName, tracked := crdNameToResource[name] + if !tracked { + return + } + if err := w.startMonitorInformer(resourceName, notifyEvents); err != nil { + w.logger.Error(err, "prometheus-cr: failed to start informer after CRD became available", "crd", name) + return + } + w.logger.Info("prometheus-cr: CRD available, started informer", "crd", name, "resource", resourceName) + }, + DeleteFunc: func(obj interface{}) { + name, ok := crdObjectName(obj) + if !ok { + return + } + resourceName, tracked := crdNameToResource[name] + if !tracked { + return + } + w.stopMonitorInformer(resourceName, notifyEvents) + w.logger.Info("prometheus-cr: CRD removed, stopped informer and dropped targets", "crd", name, "resource", resourceName) + }, + }) + factory.Start(w.stopChannel) +} + +// notify performs a non-blocking send on the buffered notification channel. +func notify(notifyEvents chan struct{}) { + select { + case notifyEvents <- struct{}{}: + default: + } +} + +// notifyHandler returns event handlers that coalesce ServiceMonitor/PodMonitor +// changes into the notification channel (non-blocking so rate-limiting upstream +// is never starved). +func notifyHandler(notifyEvents chan struct{}) cache.ResourceEventHandlerFuncs { + return cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { notify(notifyEvents) }, + UpdateFunc: func(oldObj, newObj interface{}) { notify(notifyEvents) }, + DeleteFunc: func(obj interface{}) { notify(notifyEvents) }, } - if !success { - return fmt.Errorf("failed to sync cache") +} + +// Watch starts watching for ServiceMonitor/PodMonitor changes. It is resilient +// to either CRD being absent: it starts the informer for each CRD that exists +// now, defers the rest until their CRDs appear (via a CustomResourceDefinition +// watch), and never fails startup just because a CRD is missing. This means the +// TA starts and stays healthy regardless of CRD install ordering, and begins +// (or stops) watching each type automatically as its CRD is created or deleted. +func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors chan error) error { + // this channel needs to be buffered because notifications are asynchronous and neither producers nor consumers wait + notifyEvents := make(chan struct{}, 1) + + // Start informers for CRDs that already exist. Absent CRDs are simply + // skipped here; the CRD watch below starts them if/when they appear. + for crdName, resourceName := range crdNameToResource { + exists, err := w.crdExists(context.Background(), crdName) + if err != nil { + // Surface the error but keep going — a transient API error must not + // take down the allocator. The CRD watch will recover the informer. + w.logger.Error(err, "prometheus-cr: failed to check for CRD, deferring to CRD watch", "crd", crdName) + continue + } + if !exists { + w.logger.Info("prometheus-cr: CRD not present at startup, deferring informer until it is created", "crd", crdName) + continue + } + if startErr := w.startMonitorInformer(resourceName, notifyEvents); startErr != nil { + w.logger.Error(startErr, "prometheus-cr: failed to start informer for present CRD, deferring to CRD watch", "crd", crdName) + continue + } + w.logger.Info("prometheus-cr: started informer for present CRD", "crd", crdName, "resource", resourceName) } + // React to CRDs being created/deleted at runtime with no restart. + w.watchCRDs(notifyEvents) + // limit the rate of outgoing events w.rateLimitedEventSender(upstreamEvents, notifyEvents) @@ -222,32 +387,54 @@ func (w *PrometheusCRWatcher) rateLimitedEventSender(upstreamEvents chan Event, } func (w *PrometheusCRWatcher) Close() error { + // Stop any per-type monitoring informers (each has its own stop channel). + w.informersMtx.Lock() + for name, stopCh := range w.informerStopChannels { + close(stopCh) + delete(w.informerStopChannels, name) + delete(w.informers, name) + } + w.informersMtx.Unlock() + + // Stop the CRD watch and release Watch's blocking read. close(w.stopChannel) return nil } func (w *PrometheusCRWatcher) LoadConfig(ctx context.Context) (*promconfig.Config, error) { + // Snapshot the currently-running informers. Either may be absent if its CRD + // is not (yet) installed; in that case its monitors are simply skipped and + // no scrape jobs are generated for that type. + w.informersMtx.RLock() + smInformer := w.informers[monitoringv1.ServiceMonitorName] + pmInformer := w.informers[monitoringv1.PodMonitorName] + w.informersMtx.RUnlock() + store := assets.NewStoreBuilder(w.k8sClient.CoreV1(), w.k8sClient.CoreV1()) serviceMonitorInstances := make(map[string]*monitoringv1.ServiceMonitor) - smRetrieveErr := w.informers[monitoringv1.ServiceMonitorName].ListAll(w.serviceMonitorSelector, func(sm interface{}) { - monitor := sm.(*monitoringv1.ServiceMonitor) - key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor) - w.addStoreAssetsForServiceMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.Endpoints, store) - serviceMonitorInstances[key] = monitor - }) - if smRetrieveErr != nil { - return nil, smRetrieveErr + if smInformer != nil { + smRetrieveErr := smInformer.ListAll(w.serviceMonitorSelector, func(sm interface{}) { + monitor := sm.(*monitoringv1.ServiceMonitor) + key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor) + w.addStoreAssetsForServiceMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.Endpoints, store) + serviceMonitorInstances[key] = monitor + }) + if smRetrieveErr != nil { + return nil, smRetrieveErr + } } podMonitorInstances := make(map[string]*monitoringv1.PodMonitor) - pmRetrieveErr := w.informers[monitoringv1.PodMonitorName].ListAll(w.podMonitorSelector, func(pm interface{}) { - monitor := pm.(*monitoringv1.PodMonitor) - key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor) - w.addStoreAssetsForPodMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.PodMetricsEndpoints, store) - podMonitorInstances[key] = monitor - }) - if pmRetrieveErr != nil { - return nil, pmRetrieveErr + if pmInformer != nil { + pmRetrieveErr := pmInformer.ListAll(w.podMonitorSelector, func(pm interface{}) { + monitor := pm.(*monitoringv1.PodMonitor) + key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor) + w.addStoreAssetsForPodMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.PodMetricsEndpoints, store) + podMonitorInstances[key] = monitor + }) + if pmRetrieveErr != nil { + return nil, pmRetrieveErr + } } generatedConfig, err := w.configGenerator.GenerateServerConfiguration( diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go index 7ad07f717..662800508 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/go-logr/logr" monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1" fakemonitoringclient "github.com/prometheus-operator/prometheus-operator/pkg/client/versioned/fake" "github.com/prometheus-operator/prometheus-operator/pkg/informers" @@ -21,7 +22,10 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" v1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + apiextensionsfake "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/kubernetes/fake" "k8s.io/client-go/tools/cache" "k8s.io/utils/ptr" @@ -252,17 +256,12 @@ func TestLoadConfig(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { w := getTestPrometheusCRWatcher(t, tt.serviceMonitor, tt.podMonitor) - for _, informer := range w.informers { - // Start informers in order to populate cache. - informer.Start(w.stopChannel) - } - - // Wait for informers to sync. - for _, informer := range w.informers { - for !informer.HasSynced() { - time.Sleep(50 * time.Millisecond) - } - } + defer func() { _ = w.Close() }() + + // Start both informers via the per-type lifecycle (both CRDs present). + notifyEvents := make(chan struct{}, 1) + require.NoError(t, w.startMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents)) + require.NoError(t, w.startMonitorInformer(monitoringv1.PodMonitorName, notifyEvents)) got, err := w.LoadConfig(context.Background()) require.NoError(t, err) @@ -305,11 +304,15 @@ func TestRateLimit(t *testing.T) { _, err = w.kubeMonitoringClient.MonitoringV1().ServiceMonitors("test").Create(context.Background(), serviceMonitor, metav1.CreateOptions{}) require.NoError(t, err) - // wait for cache sync first - for _, informer := range w.informers { - success := cache.WaitForCacheSync(w.stopChannel, informer.HasSynced) - require.True(t, success) - } + // wait for the watcher to start the ServiceMonitor informer and finish its + // startup loop (both CRDs present => both informers running) so the + // rate-limited event sender is active before we measure event timing. + require.Eventually(t, func() bool { + w.informersMtx.RLock() + defer w.informersMtx.RUnlock() + sm, ok := w.informers[monitoringv1.ServiceMonitorName] + return ok && sm.HasSynced() && len(w.informers) == 2 + }, 5*time.Second, 10*time.Millisecond) require.Eventually(t, func() bool { _, createErr := w.kubeMonitoringClient.MonitoringV1().ServiceMonitors("test").Update(context.Background(), serviceMonitor, metav1.UpdateOptions{}) @@ -353,9 +356,26 @@ func TestRateLimit(t *testing.T) { } -// getTestPrometheuCRWatcher creates a test instance of PrometheusCRWatcher with fake clients -// and test secrets. +// getTestPrometheusCRWatcher creates a test instance of PrometheusCRWatcher with +// fake clients and test secrets. Both the ServiceMonitor and PodMonitor CRDs are +// registered in the fake apiextensions client, so the watcher behaves as if both +// CRDs are present. Use getTestPrometheusCRWatcherWithCRDs to control CRD presence. func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, pm *monitoringv1.PodMonitor) *PrometheusCRWatcher { + return getTestPrometheusCRWatcherWithCRDs(t, sm, pm, true, true) +} + +// crdFor builds a minimal CustomResourceDefinition object for the given +// monitoring resource (e.g. "servicemonitors"), matching the names the watcher +// keys on. +func crdFor(resourceName string) *apiextensionsv1.CustomResourceDefinition { + return &apiextensionsv1.CustomResourceDefinition{ + ObjectMeta: metav1.ObjectMeta{ + Name: resourceName + "." + monitoringv1.SchemeGroupVersion.Group, + }, + } +} + +func getTestPrometheusCRWatcherWithCRDs(t *testing.T, sm *monitoringv1.ServiceMonitor, pm *monitoringv1.PodMonitor, smCRD, pmCRD bool) *PrometheusCRWatcher { mClient := fakemonitoringclient.NewSimpleClientset() //nolint:staticcheck // NewClientset causes structured merge diff schema errors in tests if sm != nil { _, err := mClient.MonitoringV1().ServiceMonitors("test").Create(context.Background(), sm, metav1.CreateOptions{}) @@ -370,6 +390,15 @@ func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, p } } + var crdObjects []runtime.Object + if smCRD { + crdObjects = append(crdObjects, crdFor(monitoringv1.ServiceMonitorName)) + } + if pmCRD { + crdObjects = append(crdObjects, crdFor(monitoringv1.PodMonitorName)) + } + crdClient := apiextensionsfake.NewSimpleClientset(crdObjects...) + k8sClient := fake.NewSimpleClientset() _, err := k8sClient.CoreV1().Secrets("test").Create(context.Background(), &v1.Secret{ ObjectMeta: metav1.ObjectMeta{ @@ -392,12 +421,6 @@ func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, p t.Fatal(t, err) } - factory := informers.NewMonitoringInformerFactories(map[string]struct{}{v1.NamespaceAll: {}}, map[string]struct{}{}, mClient, 0, nil) - informers, err := getInformers(factory) - if err != nil { - t.Fatal(t, err) - } - prom := &monitoringv1.Prometheus{ ObjectMeta: metav1.ObjectMeta{ Name: "test", @@ -417,9 +440,12 @@ func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, p } return &PrometheusCRWatcher{ + logger: logr.Discard(), kubeMonitoringClient: mClient, k8sClient: k8sClient, - informers: informers, + crdClient: crdClient, + informers: map[string]*informers.ForResource{}, + informerStopChannels: map[string]chan struct{}{}, configGenerator: generator, prom: prom, serviceMonitorSelector: getSelector(nil), @@ -445,3 +471,139 @@ func sanitizeScrapeConfigsForTest(scs []*promconfig.ScrapeConfig) { sc.ExtraScrapeMetrics = nil } } + +// runningInformers returns the set of monitoring resource names whose informers +// are currently active. +func runningInformers(w *PrometheusCRWatcher) map[string]bool { + w.informersMtx.RLock() + defer w.informersMtx.RUnlock() + out := map[string]bool{} + for name := range w.informers { + out[name] = true + } + return out +} + +// TestWatchStartsHealthyWithoutCRDs verifies the TA does not error when neither +// the ServiceMonitor nor PodMonitor CRD exists: Watch returns no error, no +// informers are started, and LoadConfig yields a config with no scrape jobs. +func TestWatchStartsHealthyWithoutCRDs(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false) + w.eventInterval = 5 * time.Millisecond + defer func() { _ = w.Close() }() + + watchDone := make(chan error, 1) + go func() { watchDone <- w.Watch(make(chan Event, 1), make(chan error, 1)) }() + + // Give the CRD watch time to sync and (not) start anything. + require.Never(t, func() bool { + return len(runningInformers(w)) > 0 + }, 200*time.Millisecond, 20*time.Millisecond) + + cfg, err := w.LoadConfig(context.Background()) + require.NoError(t, err) + assert.Empty(t, cfg.ScrapeConfigs) + + // Watch must still be running (resilient), not have returned an error. + select { + case err := <-watchDone: + t.Fatalf("Watch exited unexpectedly with: %v", err) + default: + } +} + +// TestWatchPerTypeIndependence verifies that with only the ServiceMonitor CRD +// present, the SM informer starts and the PM informer does not. +func TestWatchPerTypeIndependence(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, false) + w.eventInterval = 5 * time.Millisecond + defer func() { _ = w.Close() }() + + go func() { _ = w.Watch(make(chan Event, 1), make(chan error, 1)) }() + + require.Eventually(t, func() bool { + return runningInformers(w)[monitoringv1.ServiceMonitorName] + }, 5*time.Second, 10*time.Millisecond) + + // PodMonitor CRD is absent, so its informer must never start. + require.Never(t, func() bool { + return runningInformers(w)[monitoringv1.PodMonitorName] + }, 200*time.Millisecond, 20*time.Millisecond) +} + +// TestWatchStartsInformerWhenCRDCreated verifies the TA begins watching a type +// when its CRD is created at runtime — no restart required. +func TestWatchStartsInformerWhenCRDCreated(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false) + w.eventInterval = 5 * time.Millisecond + defer func() { _ = w.Close() }() + + go func() { _ = w.Watch(make(chan Event, 1), make(chan error, 1)) }() + + // Nothing running initially. + require.Never(t, func() bool { + return len(runningInformers(w)) > 0 + }, 200*time.Millisecond, 20*time.Millisecond) + + // Create the ServiceMonitor CRD after startup. + _, err := w.crdClient.ApiextensionsV1().CustomResourceDefinitions().Create( + context.Background(), crdFor(monitoringv1.ServiceMonitorName), metav1.CreateOptions{}) + require.NoError(t, err) + + require.Eventually(t, func() bool { + return runningInformers(w)[monitoringv1.ServiceMonitorName] + }, 5*time.Second, 10*time.Millisecond) +} + +// TestStopMonitorInformerDropsType verifies the stop path used when a CRD is +// removed at runtime: the informer is stopped, dropped from the active set (so +// LoadConfig no longer emits its targets), and a reload is signalled. This is +// the logic invoked by the CRD watch's DeleteFunc; it is exercised directly so +// the assertion does not depend on fake-clientset delete-watch delivery. +func TestStopMonitorInformerDropsType(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, false) + defer func() { _ = w.Close() }() + + notifyEvents := make(chan struct{}, 1) + require.NoError(t, w.startMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents)) + require.True(t, runningInformers(w)[monitoringv1.ServiceMonitorName]) + // drain the start notification + select { + case <-notifyEvents: + default: + } + + w.stopMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents) + + assert.False(t, runningInformers(w)[monitoringv1.ServiceMonitorName], "informer should be stopped and dropped") + // a reload must be signalled so the dropped type's targets are removed + select { + case <-notifyEvents: + default: + t.Fatal("expected a reload notification after stopping the informer") + } + + // LoadConfig must succeed and emit no scrape jobs once the type is dropped. + cfg, err := w.LoadConfig(context.Background()) + require.NoError(t, err) + assert.Empty(t, cfg.ScrapeConfigs) +} + +// TestCRDObjectName verifies the CRD-name extraction used by the CRD watch, +// including the delete tombstone wrapper. +func TestCRDObjectName(t *testing.T) { + crd := crdFor(monitoringv1.ServiceMonitorName) + name, ok := crdObjectName(crd) + require.True(t, ok) + assert.Equal(t, "servicemonitors.monitoring.coreos.com", name) + if _, tracked := crdNameToResource[name]; !tracked { + t.Fatalf("CRD name %q is not mapped to a monitoring resource", name) + } + + name, ok = crdObjectName(cache.DeletedFinalStateUnknown{Key: "k", Obj: crd}) + require.True(t, ok) + assert.Equal(t, "servicemonitors.monitoring.coreos.com", name) + + _, ok = crdObjectName("not-a-crd") + assert.False(t, ok) +} diff --git a/go.mod b/go.mod index ada68fccd..91e7dd2ef 100644 --- a/go.mod +++ b/go.mod @@ -37,6 +37,7 @@ require ( gopkg.in/yaml.v2 v2.4.0 gopkg.in/yaml.v3 v3.0.1 k8s.io/api v0.35.4 + k8s.io/apiextensions-apiserver v0.35.4 k8s.io/apimachinery v0.35.4 k8s.io/client-go v0.35.4 k8s.io/component-base v0.35.4 @@ -245,7 +246,6 @@ require ( gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/ini.v1 v1.67.1 // indirect - k8s.io/apiextensions-apiserver v0.35.4 // indirect k8s.io/kube-openapi v0.0.0-20260317180543-43fb72c5454a // indirect sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect sigs.k8s.io/randfill v1.0.0 // indirect From 249cf73fd71b28a4e30b2563e755d2884e5ea951 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Wed, 15 Jul 2026 14:56:03 +0100 Subject: [PATCH 05/10] test(target-allocator): cover CRD-resilience error and idempotency paths Add unit tests for the missing branches in the TA CRD-resilience logic: startMonitorInformer idempotency and unknown-resource error, stopMonitorInformer no-op when not running, crdExists surfacing non-NotFound errors, Watch tolerating a CRD-check error, the untracked-CRD add path, and the notify coalescing handler on add/update/delete. --- .../watcher/promOperator_resilience_test.go | 158 ++++++++++++++++++ 1 file changed, 158 insertions(+) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go new file mode 100644 index 000000000..2694c5fd8 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go @@ -0,0 +1,158 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package watcher + +import ( + "context" + "fmt" + "testing" + "time" + + monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + apiextensionsfake "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + clienttesting "k8s.io/client-go/testing" +) + +// TestStartMonitorInformerIdempotent verifies that starting an already-running +// informer is a no-op: it stays running and no second reload is signalled. +func TestStartMonitorInformerIdempotent(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, false) + defer func() { _ = w.Close() }() + + notifyEvents := make(chan struct{}, 1) + require.NoError(t, w.startMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents)) + // drain the reload notification emitted by the first (real) start + select { + case <-notifyEvents: + default: + } + + // Second call must be a no-op. + require.NoError(t, w.startMonitorInformer(monitoringv1.ServiceMonitorName, notifyEvents)) + require.True(t, runningInformers(w)[monitoringv1.ServiceMonitorName]) + select { + case <-notifyEvents: + t.Fatal("no-op start must not signal a reload") + default: + } +} + +// TestStartMonitorInformerUnknownResource verifies an unrecognised resource name +// is rejected with an error and starts nothing. +func TestStartMonitorInformerUnknownResource(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false) + defer func() { _ = w.Close() }() + + err := w.startMonitorInformer("widgets", make(chan struct{}, 1)) + require.Error(t, err) + assert.Contains(t, err.Error(), "unknown monitoring resource") + assert.Empty(t, runningInformers(w)) +} + +// TestStopMonitorInformerNotRunning verifies stopping an informer that is not +// running is a no-op and signals no reload. +func TestStopMonitorInformerNotRunning(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false) + defer func() { _ = w.Close() }() + + notifyEvents := make(chan struct{}, 1) + w.stopMonitorInformer(monitoringv1.PodMonitorName, notifyEvents) + + assert.Empty(t, runningInformers(w)) + select { + case <-notifyEvents: + t.Fatal("stopping a non-running informer must not signal a reload") + default: + } +} + +// TestCrdExistsSurfacesNonNotFoundError verifies crdExists propagates +// non-NotFound API errors instead of treating them as "absent". +func TestCrdExistsSurfacesNonNotFoundError(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false) + defer func() { _ = w.Close() }() + + fakeCRD := w.crdClient.(*apiextensionsfake.Clientset) + fakeCRD.PrependReactor("get", "customresourcedefinitions", + func(action clienttesting.Action) (bool, runtime.Object, error) { + return true, nil, apierrors.NewInternalError(fmt.Errorf("boom")) + }) + + exists, err := w.crdExists(context.Background(), crdFor(monitoringv1.ServiceMonitorName).Name) + require.Error(t, err) + assert.False(t, exists) +} + +// TestWatchToleratesCRDCheckError verifies a non-NotFound error while probing +// for a CRD at startup does not take the watcher down: Watch keeps running. +func TestWatchToleratesCRDCheckError(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, true) + w.eventInterval = 5 * time.Millisecond + defer func() { _ = w.Close() }() + + fakeCRD := w.crdClient.(*apiextensionsfake.Clientset) + fakeCRD.PrependReactor("get", "customresourcedefinitions", + func(action clienttesting.Action) (bool, runtime.Object, error) { + return true, nil, apierrors.NewInternalError(fmt.Errorf("boom")) + }) + + watchDone := make(chan error, 1) + go func() { watchDone <- w.Watch(make(chan Event, 1), make(chan error, 1)) }() + + select { + case err := <-watchDone: + t.Fatalf("Watch exited unexpectedly with: %v", err) + case <-time.After(200 * time.Millisecond): + } +} + +// TestWatchIgnoresUntrackedCRD verifies that creating a CRD the watcher does not +// track never starts a monitoring informer. +func TestWatchIgnoresUntrackedCRD(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false) + w.eventInterval = 5 * time.Millisecond + defer func() { _ = w.Close() }() + + go func() { _ = w.Watch(make(chan Event, 1), make(chan error, 1)) }() + + _, err := w.crdClient.ApiextensionsV1().CustomResourceDefinitions().Create( + context.Background(), + &apiextensionsv1.CustomResourceDefinition{ObjectMeta: metav1.ObjectMeta{Name: "widgets.example.com"}}, + metav1.CreateOptions{}) + require.NoError(t, err) + + require.Never(t, func() bool { + return len(runningInformers(w)) > 0 + }, 200*time.Millisecond, 20*time.Millisecond) +} + +// TestNotifyHandlerSignalsOnAllEvents verifies the coalescing handler signals a +// reload on add, update and delete events. +func TestNotifyHandlerSignalsOnAllEvents(t *testing.T) { + for _, tc := range []string{"add", "update", "delete"} { + t.Run(tc, func(t *testing.T) { + ch := make(chan struct{}, 1) + h := notifyHandler(ch) + switch tc { + case "add": + h.AddFunc(nil) + case "update": + h.UpdateFunc(nil, nil) + case "delete": + h.DeleteFunc(nil) + } + select { + case <-ch: + default: + t.Fatalf("%s handler did not signal a reload", tc) + } + }) + } +} From 1260c2739a23a3bc0dd3f51ce429635d35000e88 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Tue, 21 Jul 2026 11:09:49 +0100 Subject: [PATCH 06/10] fix(target-allocator): don't hold informer lock across cache sync startMonitorInformer held informersMtx across informer.Start and WaitForNamedCacheSync, with the sync waiting on a local stop channel not reachable by Close(). Under a slow or rejected CRD LIST this blocked LoadConfig for the whole sync and, worse, deadlocked Close() (which needs the same lock) until SIGKILL. Run Start/WaitForNamedCacheSync outside the lock and select on w.stopChannel so Close() always interrupts an in-flight sync; re-check shutdown under the lock before registering so no live informer is left after Close(). Close() now signals w.stopChannel before draining, closing the register-after-close race. Adds TestCloseInterruptsInFlightSync (blocks the ServiceMonitor LIST, asserts Close() returns) which hangs before the fix. Verified with go test -race. --- .../watcher/promOperator.go | 62 +++++++++++++++---- .../watcher/promOperator_resilience_test.go | 46 ++++++++++++++ 2 files changed, 97 insertions(+), 11 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 47be05a3d..3a072fc41 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -187,18 +187,20 @@ func (w *PrometheusCRWatcher) crdExists(ctx context.Context, crdName string) (bo // the notification handler, and records it for use by LoadConfig. It is // idempotent: calling it for an already-running informer is a no-op. func (w *PrometheusCRWatcher) startMonitorInformer(resourceName string, notifyEvents chan struct{}) error { - w.informersMtx.Lock() - defer w.informersMtx.Unlock() - - if _, running := w.informers[resourceName]; running { - return nil - } - gvr, ok := monitoringResources[resourceName] if !ok { return fmt.Errorf("unknown monitoring resource %q", resourceName) } + // Fast path: already running. Checked under a short read lock so we never hold + // informersMtx across the (potentially slow) cache sync below. + w.informersMtx.RLock() + _, running := w.informers[resourceName] + w.informersMtx.RUnlock() + if running { + return nil + } + informer, err := informers.NewInformersForResource(w.newMonitoringFactory(), gvr) if err != nil { return err @@ -206,15 +208,51 @@ func (w *PrometheusCRWatcher) startMonitorInformer(resourceName string, notifyEv stopCh := make(chan struct{}) informer.Start(stopCh) - if ok := cache.WaitForNamedCacheSync(resourceName, stopCh, informer.HasSynced); !ok { + + // Wait for the cache to sync WITHOUT holding informersMtx, and abort the wait + // if the watcher is shutting down. A CRD's initial LIST can be slow or rejected + // (exactly the condition this resilience change tolerates); holding the lock + // here would block LoadConfig and, worse, deadlock Close() — which needs the + // same lock — until SIGKILL. Selecting on w.stopChannel lets Close() always + // interrupt an in-flight sync. + synced := make(chan bool, 1) + go func() { + synced <- cache.WaitForNamedCacheSync(resourceName, stopCh, informer.HasSynced) + }() + select { + case ok := <-synced: + if !ok { + close(stopCh) + return fmt.Errorf("failed to sync %s informer cache", resourceName) + } + case <-w.stopChannel: + // Shutting down: stop the informer and abandon the start (not an error). + // Closing stopCh also unblocks the WaitForNamedCacheSync goroutine. close(stopCh) - return fmt.Errorf("failed to sync %s informer cache", resourceName) + return nil } informer.AddEventHandler(notifyHandler(notifyEvents)) + w.informersMtx.Lock() + // If the watcher closed while we were syncing, don't register a live informer: + // Close() has already drained the stop channels and would never stop this one. + select { + case <-w.stopChannel: + w.informersMtx.Unlock() + close(stopCh) + return nil + default: + } + // If another goroutine already started this resource, discard ours. + if _, running := w.informers[resourceName]; running { + w.informersMtx.Unlock() + close(stopCh) + return nil + } w.informers[resourceName] = informer w.informerStopChannels[resourceName] = stopCh + w.informersMtx.Unlock() // A new resource type just became available; trigger a config reload. notify(notifyEvents) @@ -387,6 +425,10 @@ func (w *PrometheusCRWatcher) rateLimitedEventSender(upstreamEvents chan Event, } func (w *PrometheusCRWatcher) Close() error { + // Signal shutdown first so any in-flight startMonitorInformer aborts its cache + // sync and will not register a new informer after this point. + close(w.stopChannel) + // Stop any per-type monitoring informers (each has its own stop channel). w.informersMtx.Lock() for name, stopCh := range w.informerStopChannels { @@ -396,8 +438,6 @@ func (w *PrometheusCRWatcher) Close() error { } w.informersMtx.Unlock() - // Stop the CRD watch and release Watch's blocking read. - close(w.stopChannel) return nil } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go index 2694c5fd8..3c1e6bcd0 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go @@ -10,6 +10,7 @@ import ( "time" monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1" + fakemonitoringclient "github.com/prometheus-operator/prometheus-operator/pkg/client/versioned/fake" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" @@ -156,3 +157,48 @@ func TestNotifyHandlerSignalsOnAllEvents(t *testing.T) { }) } } + + +// TestCloseInterruptsInFlightSync verifies that Close() returns promptly even +// when an informer is still syncing against a slow/hung apiserver LIST. This is +// the shutdown path the CRD-resilience change must tolerate: a CRD's initial +// LIST is slow or rejected, so WaitForNamedCacheSync is still blocked when the +// process is asked to shut down. Close() must not wait for that sync (which +// could otherwise hang until SIGKILL). +func TestCloseInterruptsInFlightSync(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, false) + + // Make the ServiceMonitor LIST block, simulating a slow/hung apiserver so the + // informer never finishes its initial cache sync. + fakeMon := w.kubeMonitoringClient.(*fakemonitoringclient.Clientset) + release := make(chan struct{}) + defer close(release) + fakeMon.PrependReactor("list", "servicemonitors", + func(action clienttesting.Action) (bool, runtime.Object, error) { + <-release + return false, nil, nil + }) + + // Start the informer; it will block waiting for the (never-completing) sync. + startReturned := make(chan struct{}) + go func() { + _ = w.startMonitorInformer(monitoringv1.ServiceMonitorName, make(chan struct{}, 1)) + close(startReturned) + }() + + // Let it reach the sync wait. + time.Sleep(300 * time.Millisecond) + + // Close() must return promptly regardless of the in-flight sync. + closed := make(chan struct{}) + go func() { + _ = w.Close() + close(closed) + }() + + select { + case <-closed: + case <-time.After(3 * time.Second): + t.Fatal("Close() hung while an informer was still syncing against a slow apiserver LIST") + } +} From 7c41c205a5db904dbf064d1a992eb5afd4688541 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 10:41:50 +0100 Subject: [PATCH 07/10] perf(target-allocator): watch CRDs via metadata-only informer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit watchCRDs used a full CustomResourceDefinition informer that cached every CRD object including its OpenAPI v3 schema, though only the two names in crdNameToResource matter — memory could balloon on clusters with many large CRDs. Switch to a client-go metadata informer (PartialObjectMetadata), which caches object metadata only. Add a metadata client to the watcher; crdExists keeps using the apiextensions client. Tests drive CRD events via the metadata fake tracker; crdObjectName now handles PartialObjectMetadata. --- .../watcher/promOperator.go | 28 +++++++++---- .../watcher/promOperator_resilience_test.go | 13 ++++-- .../watcher/promOperator_test.go | 41 ++++++++++++++++--- go.mod | 2 +- 4 files changed, 67 insertions(+), 17 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 3a072fc41..a9f4272da 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -23,13 +23,14 @@ import ( "gopkg.in/yaml.v2" apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" - apiextensionsinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions" v1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/metadata" + "k8s.io/client-go/metadata/metadatainformer" "k8s.io/client-go/tools/cache" allocatorconfig "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/config" @@ -71,6 +72,14 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr return nil, err } + // Metadata-only client for the CRD watch: it caches just object metadata + // (name), not each CustomResourceDefinition's full spec/OpenAPI schema, so + // memory stays flat on clusters with many (large) CRDs. + metadataClient, err := metadata.NewForConfig(cfg.ClusterConfig) + if err != nil { + return nil, err + } + // TODO: We should make these durations configurable // Namespace must be non-empty; the config generator panics otherwise. collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") @@ -115,6 +124,7 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr kubeMonitoringClient: mClient, k8sClient: clientset, crdClient: crdClient, + metadataClient: metadataClient, informers: map[string]*informers.ForResource{}, informerStopChannels: map[string]chan struct{}{}, stopChannel: make(chan struct{}), @@ -132,6 +142,7 @@ type PrometheusCRWatcher struct { kubeMonitoringClient monitoringclient.Interface k8sClient kubernetes.Interface crdClient apiextensionsclient.Interface + metadataClient metadata.Interface eventInterval time.Duration stopChannel chan struct{} configGenerator *prometheus.ConfigGenerator @@ -278,14 +289,16 @@ func (w *PrometheusCRWatcher) stopMonitorInformer(resourceName string, notifyEve notify(notifyEvents) } -// crdObjectName extracts the CustomResourceDefinition name from an informer -// event object, tolerating the tombstone wrapper delivered on some deletes. +// crdObjectName extracts the CustomResourceDefinition name from a metadata +// informer event object, tolerating the tombstone wrapper delivered on some +// deletes. The metadata informer delivers *metav1.PartialObjectMetadata (name +// only), not the full CustomResourceDefinition. func crdObjectName(obj interface{}) (string, bool) { switch t := obj.(type) { - case *apiextensionsv1.CustomResourceDefinition: + case *metav1.PartialObjectMetadata: return t.Name, true case cache.DeletedFinalStateUnknown: - if crd, ok := t.Obj.(*apiextensionsv1.CustomResourceDefinition); ok { + if crd, ok := t.Obj.(*metav1.PartialObjectMetadata); ok { return crd.Name, true } } @@ -298,8 +311,9 @@ func crdObjectName(obj interface{}) (string, bool) { // existing CRDs on its initial sync, so CRDs present at startup are handled here // too (startMonitorInformer is idempotent). func (w *PrometheusCRWatcher) watchCRDs(notifyEvents chan struct{}) { - factory := apiextensionsinformers.NewSharedInformerFactory(w.crdClient, allocatorconfig.DefaultResyncTime) - crdInformer := factory.Apiextensions().V1().CustomResourceDefinitions().Informer() + crdGVR := apiextensionsv1.SchemeGroupVersion.WithResource("customresourcedefinitions") + factory := metadatainformer.NewSharedInformerFactory(w.metadataClient, allocatorconfig.DefaultResyncTime) + crdInformer := factory.ForResource(crdGVR).Informer() _, _ = crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { name, ok := crdObjectName(obj) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go index 3c1e6bcd0..2f1cbc9e4 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go @@ -18,6 +18,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + metadatafake "k8s.io/client-go/metadata/fake" clienttesting "k8s.io/client-go/testing" ) @@ -123,10 +124,14 @@ func TestWatchIgnoresUntrackedCRD(t *testing.T) { go func() { _ = w.Watch(make(chan Event, 1), make(chan error, 1)) }() - _, err := w.crdClient.ApiextensionsV1().CustomResourceDefinitions().Create( - context.Background(), - &apiextensionsv1.CustomResourceDefinition{ObjectMeta: metav1.ObjectMeta{Name: "widgets.example.com"}}, - metav1.CreateOptions{}) + // The CRD watch is metadata-only; add an untracked CRD to the metadata tracker + // so the informer delivers it and we can assert it starts no monitoring informer. + crdGVR := apiextensionsv1.SchemeGroupVersion.WithResource("customresourcedefinitions") + err := w.metadataClient.(*metadatafake.FakeMetadataClient).Tracker().Create(crdGVR, + &metav1.PartialObjectMetadata{ + TypeMeta: metav1.TypeMeta{APIVersion: apiextensionsv1.SchemeGroupVersion.String(), Kind: "CustomResourceDefinition"}, + ObjectMeta: metav1.ObjectMeta{Name: "widgets.example.com"}, + }, "") require.NoError(t, err) require.Never(t, func() bool { diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go index 662800508..9e00ee20e 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_test.go @@ -27,6 +27,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/kubernetes/fake" + metadatafake "k8s.io/client-go/metadata/fake" "k8s.io/client-go/tools/cache" "k8s.io/utils/ptr" ) @@ -375,6 +376,21 @@ func crdFor(resourceName string) *apiextensionsv1.CustomResourceDefinition { } } +// crdMetaFor builds the metadata-only view of a CRD for the fake metadata client +// backing the CRD watch. TypeMeta must carry the CRD GVK so the fake tracker maps +// it to the customresourcedefinitions resource the informer lists on. +func crdMetaFor(resourceName string) *metav1.PartialObjectMetadata { + return &metav1.PartialObjectMetadata{ + TypeMeta: metav1.TypeMeta{ + APIVersion: apiextensionsv1.SchemeGroupVersion.String(), + Kind: "CustomResourceDefinition", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: resourceName + "." + monitoringv1.SchemeGroupVersion.Group, + }, + } +} + func getTestPrometheusCRWatcherWithCRDs(t *testing.T, sm *monitoringv1.ServiceMonitor, pm *monitoringv1.PodMonitor, smCRD, pmCRD bool) *PrometheusCRWatcher { mClient := fakemonitoringclient.NewSimpleClientset() //nolint:staticcheck // NewClientset causes structured merge diff schema errors in tests if sm != nil { @@ -399,6 +415,20 @@ func getTestPrometheusCRWatcherWithCRDs(t *testing.T, sm *monitoringv1.ServiceMo } crdClient := apiextensionsfake.NewSimpleClientset(crdObjects...) + // Metadata-only fake backing the CRD watch, populated with the same present CRDs. + metaScheme := metadatafake.NewTestScheme() + if err := metav1.AddMetaToScheme(metaScheme); err != nil { + t.Fatal(t, err) + } + var crdMetaObjects []runtime.Object + if smCRD { + crdMetaObjects = append(crdMetaObjects, crdMetaFor(monitoringv1.ServiceMonitorName)) + } + if pmCRD { + crdMetaObjects = append(crdMetaObjects, crdMetaFor(monitoringv1.PodMonitorName)) + } + metadataClient := metadatafake.NewSimpleMetadataClient(metaScheme, crdMetaObjects...) + k8sClient := fake.NewSimpleClientset() _, err := k8sClient.CoreV1().Secrets("test").Create(context.Background(), &v1.Secret{ ObjectMeta: metav1.ObjectMeta{ @@ -444,6 +474,7 @@ func getTestPrometheusCRWatcherWithCRDs(t *testing.T, sm *monitoringv1.ServiceMo kubeMonitoringClient: mClient, k8sClient: k8sClient, crdClient: crdClient, + metadataClient: metadataClient, informers: map[string]*informers.ForResource{}, informerStopChannels: map[string]chan struct{}{}, configGenerator: generator, @@ -545,10 +576,10 @@ func TestWatchStartsInformerWhenCRDCreated(t *testing.T) { return len(runningInformers(w)) > 0 }, 200*time.Millisecond, 20*time.Millisecond) - // Create the ServiceMonitor CRD after startup. - _, err := w.crdClient.ApiextensionsV1().CustomResourceDefinitions().Create( - context.Background(), crdFor(monitoringv1.ServiceMonitorName), metav1.CreateOptions{}) - require.NoError(t, err) + // Create the ServiceMonitor CRD after startup (metadata-only CRD watch). + crdGVR := apiextensionsv1.SchemeGroupVersion.WithResource("customresourcedefinitions") + require.NoError(t, w.metadataClient.(*metadatafake.FakeMetadataClient).Tracker().Create( + crdGVR, crdMetaFor(monitoringv1.ServiceMonitorName), "")) require.Eventually(t, func() bool { return runningInformers(w)[monitoringv1.ServiceMonitorName] @@ -592,7 +623,7 @@ func TestStopMonitorInformerDropsType(t *testing.T) { // TestCRDObjectName verifies the CRD-name extraction used by the CRD watch, // including the delete tombstone wrapper. func TestCRDObjectName(t *testing.T) { - crd := crdFor(monitoringv1.ServiceMonitorName) + crd := crdMetaFor(monitoringv1.ServiceMonitorName) name, ok := crdObjectName(crd) require.True(t, ok) assert.Equal(t, "servicemonitors.monitoring.coreos.com", name) diff --git a/go.mod b/go.mod index 3f3f0bda2..c19d15fe3 100644 --- a/go.mod +++ b/go.mod @@ -36,6 +36,7 @@ require ( gopkg.in/yaml.v2 v2.4.0 gopkg.in/yaml.v3 v3.0.1 k8s.io/api v0.36.2 + k8s.io/apiextensions-apiserver v0.36.2 k8s.io/apimachinery v0.36.2 k8s.io/client-go v0.36.2 k8s.io/component-base v0.36.2 @@ -247,7 +248,6 @@ require ( gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/ini.v1 v1.67.2 // indirect - k8s.io/apiextensions-apiserver v0.36.2 // indirect k8s.io/kube-openapi v0.0.0-20260603220949-865597e52e25 // indirect sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect sigs.k8s.io/randfill v1.0.0 // indirect From 6fe888d8a8c20c75378afc930e9517f41b1b4433 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 10:44:55 +0100 Subject: [PATCH 08/10] fix(target-allocator): bound CRD existence checks with a timeout The Watch startup loop called crdExists with context.Background() and no deadline, so a wedged apiserver could block startup indefinitely. Give each check a crdCheckTimeout-bounded context (cancelled per iteration) so a slow CRD Get fails fast and defers to the CRD watch instead of hanging. --- .../watcher/promOperator.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index a9f4272da..ba13ee542 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -40,6 +40,10 @@ const defaultCollectorNamespace = "amazon-cloudwatch" const minEventInterval = time.Second * 5 +// crdCheckTimeout bounds each startup CustomResourceDefinition existence check so +// a wedged apiserver cannot block the Watch startup loop indefinitely. +const crdCheckTimeout = 10 * time.Second + // monitoringResources maps the prometheus-operator resource name to the // GroupVersionResource used to build its informer. var monitoringResources = map[string]schema.GroupVersionResource{ @@ -378,7 +382,9 @@ func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors ch // Start informers for CRDs that already exist. Absent CRDs are simply // skipped here; the CRD watch below starts them if/when they appear. for crdName, resourceName := range crdNameToResource { - exists, err := w.crdExists(context.Background(), crdName) + ctx, cancel := context.WithTimeout(context.Background(), crdCheckTimeout) + exists, err := w.crdExists(ctx, crdName) + cancel() if err != nil { // Surface the error but keep going — a transient API error must not // take down the allocator. The CRD watch will recover the informer. From cd6e9665a88b8ef209d8ca54bbe8249b0a0ff0cb Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 10:48:34 +0100 Subject: [PATCH 09/10] fix(target-allocator): harden Close and log CRD handler registration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close() now guards shutdown with sync.Once so a second call is a no-op instead of panicking on the already-closed stopChannel. The CRD informer's AddEventHandler error is logged instead of dropped, so a failed registration (which would silently observe no CRD add/remove) is visible. (startMonitorInformer's ForResource.AddEventHandler returns no error — the upstream wrapper swallows it.) --- .../watcher/promOperator.go | 37 ++++++++++++------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index ba13ee542..83b07f41b 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -149,6 +149,7 @@ type PrometheusCRWatcher struct { metadataClient metadata.Interface eventInterval time.Duration stopChannel chan struct{} + closeOnce sync.Once configGenerator *prometheus.ConfigGenerator prom *monitoringv1.Prometheus kubeConfigPath string @@ -318,7 +319,7 @@ func (w *PrometheusCRWatcher) watchCRDs(notifyEvents chan struct{}) { crdGVR := apiextensionsv1.SchemeGroupVersion.WithResource("customresourcedefinitions") factory := metadatainformer.NewSharedInformerFactory(w.metadataClient, allocatorconfig.DefaultResyncTime) crdInformer := factory.ForResource(crdGVR).Informer() - _, _ = crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + if _, err := crdInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { name, ok := crdObjectName(obj) if !ok { @@ -346,7 +347,11 @@ func (w *PrometheusCRWatcher) watchCRDs(notifyEvents chan struct{}) { w.stopMonitorInformer(resourceName, notifyEvents) w.logger.Info("prometheus-cr: CRD removed, stopped informer and dropped targets", "crd", name, "resource", resourceName) }, - }) + }); err != nil { + // A failed registration means CRD create/delete would go unobserved; log it + // so a dead CRD watch is visible rather than silently wiring up no handler. + w.logger.Error(err, "prometheus-cr: failed to register CRD informer event handler; runtime CRD add/remove will not be observed") + } factory.Start(w.stopChannel) } @@ -445,18 +450,22 @@ func (w *PrometheusCRWatcher) rateLimitedEventSender(upstreamEvents chan Event, } func (w *PrometheusCRWatcher) Close() error { - // Signal shutdown first so any in-flight startMonitorInformer aborts its cache - // sync and will not register a new informer after this point. - close(w.stopChannel) - - // Stop any per-type monitoring informers (each has its own stop channel). - w.informersMtx.Lock() - for name, stopCh := range w.informerStopChannels { - close(stopCh) - delete(w.informerStopChannels, name) - delete(w.informers, name) - } - w.informersMtx.Unlock() + // sync.Once so a second Close() is a safe no-op rather than a panic on the + // already-closed stopChannel. + w.closeOnce.Do(func() { + // Signal shutdown first so any in-flight startMonitorInformer aborts its cache + // sync and will not register a new informer after this point. + close(w.stopChannel) + + // Stop any per-type monitoring informers (each has its own stop channel). + w.informersMtx.Lock() + for name, stopCh := range w.informerStopChannels { + close(stopCh) + delete(w.informerStopChannels, name) + delete(w.informers, name) + } + w.informersMtx.Unlock() + }) return nil } From cc2a001062969321a4bddfdee2b082cb85c2b6dc Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 10:48:34 +0100 Subject: [PATCH 10/10] test(target-allocator): cover CRD delete path and idempotent Close Add TestWatchStopsInformerWhenCRDDeleted, driving the CRD watch DeleteFunc via the metadata fake tracker to assert a deleted CRD stops its informer end to end (reliable with the metadata informer; the earlier apiextensions delete-watch was flaky). Add TestCloseIdempotent for the sync.Once guard. --- .../watcher/promOperator_resilience_test.go | 36 +++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go index 2f1cbc9e4..6a38827fc 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go @@ -207,3 +207,39 @@ func TestCloseInterruptsInFlightSync(t *testing.T) { t.Fatal("Close() hung while an informer was still syncing against a slow apiserver LIST") } } + + +// TestWatchStopsInformerWhenCRDDeleted covers the delete path end to end: with a +// tracked CRD present at startup its informer is running; deleting that CRD via +// the metadata client must drive the CRD watch's DeleteFunc, stopping the +// informer and dropping its targets — no restart required. +func TestWatchStopsInformerWhenCRDDeleted(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, true, false) // ServiceMonitor CRD present + w.eventInterval = 5 * time.Millisecond + defer func() { _ = w.Close() }() + + go func() { _ = w.Watch(make(chan Event, 1), make(chan error, 1)) }() + + // Present-at-startup CRD => its informer starts. + require.Eventually(t, func() bool { + return runningInformers(w)[monitoringv1.ServiceMonitorName] + }, 5*time.Second, 10*time.Millisecond) + + // Delete the CRD from the metadata tracker; DeleteFunc must stop the informer. + crdGVR := apiextensionsv1.SchemeGroupVersion.WithResource("customresourcedefinitions") + tracker := w.metadataClient.(*metadatafake.FakeMetadataClient).Tracker() + require.NoError(t, tracker.Delete(crdGVR, "", crdFor(monitoringv1.ServiceMonitorName).Name)) + + require.Eventually(t, func() bool { + return !runningInformers(w)[monitoringv1.ServiceMonitorName] + }, 5*time.Second, 10*time.Millisecond) +} + +// TestCloseIdempotent verifies Close() is safe to call more than once: the +// sync.Once guard makes a second call a no-op instead of panicking on the +// already-closed stop channel. +func TestCloseIdempotent(t *testing.T) { + w := getTestPrometheusCRWatcherWithCRDs(t, nil, nil, false, false) + require.NoError(t, w.Close()) + require.NotPanics(t, func() { _ = w.Close() }) +}