From 5a2600a50b04fa659548808667a821175331396c 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. (cherry picked from commit 13764510b87ac3a5894ed534c8d28d5bd982dc6b) --- .../config/config.go | 9 +++++++++ .../config/flags.go | 5 +++++ .../config/flags_test.go | 13 ++++++++++++ .../watcher/promOperator.go | 20 +++++++++++++++++++ 4 files changed, 47 insertions(+) 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/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 e8737efc3f60f8cd6acf09007d97680c1bcfb31d 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. (cherry picked from commit 0405f4decb019dd2ee173333a3985ea517b4ffb5) --- .../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 789d60ad7433ef99685945481b03b88b18227369 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. (cherry picked from commit d61d693fa6f0536786951e8e4d08f378a5e77c82) --- .../config/config.go | 4 +- .../watcher/promOperator.go | 22 +++---- internal/manifests/collector/annotations.go | 21 ++++++- .../manifests/collector/annotations_test.go | 61 +++++++++++++++++++ 4 files changed, 91 insertions(+), 17 deletions(-) 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/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 52cad873f..fddf89be8 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,7 +52,7 @@ 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 } @@ -71,6 +72,22 @@ func PodLabels(instance v1alpha1.AmazonCloudWatchAgent, base map[string]string) return merged } +// 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 13524bbbe66a218edd9ac3d02fd57e0c90674924 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Wed, 23 Sep 2026 13:40:12 +0100 Subject: [PATCH 4/4] fix(target-allocator): tolerate missing ServiceMonitor/PodMonitor CRDs Makes the Target Allocator start and run healthily whether or not the community monitoring.coreos.com ServiceMonitor/PodMonitor CRDs are installed, and start or stop watching each type as its CRD appears or disappears with no restart. Previously Watch() built both informers unconditionally and blocked on WaitForNamedCacheSync, so an absent CRD meant the apiserver rejected LIST/WATCH, the informer never synced, and Watch() returned 'failed to sync cache'. Informers are now built lazily per type once their CRD is observed, and a metadata-only CustomResourceDefinition informer starts or stops each one at runtime. Requires the chart to grant get/list/watch on customresourcedefinitions (aws-observability/helm-charts#327); without it the CRD check fails and no ServiceMonitor/PodMonitor targets are discovered while the Target Allocator still reports healthy. Verified: go build ./... clean; make impi and make checklicense pass; unit tests pass across the target-allocator and manifests packages. (cherry picked from commit 5f3d79de158905590af7e029ffbe934de6986136) --- .../watcher/promOperator.go | 386 +++++++++++++++--- .../watcher/promOperator_resilience_test.go | 243 +++++++++++ .../watcher/promOperator_test.go | 243 +++++++++-- go.mod | 2 +- 4 files changed, 783 insertions(+), 91 deletions(-) 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.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 5dfc314dd..eee4bac98 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" @@ -21,9 +22,15 @@ import ( kubeDiscovery "github.com/prometheus/prometheus/discovery/kubernetes" "gopkg.in/yaml.v2" v1 "k8s.io/api/core/v1" + apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" + apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + 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" @@ -33,6 +40,26 @@ 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{ + 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 +71,15 @@ 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 + crdClient, err := apiextensionsclient.NewForConfig(cfg.ClusterConfig) + if err != nil { + return nil, err + } - monitoringInformers, err := getInformers(factory) + // 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 } @@ -86,11 +119,18 @@ 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, + metadataClient: metadataClient, + informers: map[string]*informers.ForResource{}, + informerStopChannels: map[string]chan struct{}{}, stopChannel: make(chan struct{}), eventInterval: minEventInterval, configGenerator: generator, @@ -105,13 +145,21 @@ type PrometheusCRWatcher struct { logger logr.Logger kubeMonitoringClient monitoringclient.Interface k8sClient kubernetes.Interface - informers map[string]*informers.ForResource + crdClient apiextensionsclient.Interface + metadataClient metadata.Interface eventInterval time.Duration stopChannel chan struct{} + closeOnce sync.Once 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 +171,245 @@ 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 { + 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 } - 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) + + // 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 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) + 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 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 *metav1.PartialObjectMetadata: + return t.Name, true + case cache.DeletedFinalStateUnknown: + if crd, ok := t.Obj.(*metav1.PartialObjectMetadata); 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{}) { + crdGVR := apiextensionsv1.SchemeGroupVersion.WithResource("customresourcedefinitions") + factory := metadatainformer.NewSharedInformerFactory(w.metadataClient, allocatorconfig.DefaultResyncTime) + crdInformer := factory.ForResource(crdGVR).Informer() + if _, err := 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) + }, + }); 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) +} + +// 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 { + 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. + 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 +450,60 @@ func (w *PrometheusCRWatcher) rateLimitedEventSender(upstreamEvents chan Event, } func (w *PrometheusCRWatcher) Close() error { - close(w.stopChannel) + // 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 } 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_resilience_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go new file mode 100644 index 000000000..48708e8f8 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_resilience_test.go @@ -0,0 +1,243 @@ +// 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" + 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" + 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" + metadatafake "k8s.io/client-go/metadata/fake" + 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)) }() + + // 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 { + 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) + } + }) + } +} + +// 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") + } +} + +// 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() }) +} 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..9e00ee20e 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,8 +22,12 @@ 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" + metadatafake "k8s.io/client-go/metadata/fake" "k8s.io/client-go/tools/cache" "k8s.io/utils/ptr" ) @@ -252,17 +257,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 +305,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 +357,41 @@ 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, + }, + } +} + +// 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 { _, err := mClient.MonitoringV1().ServiceMonitors("test").Create(context.Background(), sm, metav1.CreateOptions{}) @@ -370,6 +406,29 @@ 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...) + + // 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{ @@ -392,12 +451,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 +470,13 @@ func getTestPrometheusCRWatcher(t *testing.T, sm *monitoringv1.ServiceMonitor, p } return &PrometheusCRWatcher{ + logger: logr.Discard(), kubeMonitoringClient: mClient, k8sClient: k8sClient, - informers: informers, + crdClient: crdClient, + metadataClient: metadataClient, + informers: map[string]*informers.ForResource{}, + informerStopChannels: map[string]chan struct{}{}, configGenerator: generator, prom: prom, serviceMonitorSelector: getSelector(nil), @@ -445,3 +502,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 (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] + }, 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 := crdMetaFor(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 94d5e747f..e380d72d8 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