From 13764510b87ac3a5894ed534c8d28d5bd982dc6b Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 09:21:01 +0000 Subject: [PATCH 1/4] 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 2/4] 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 3/4] 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 4/4] 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