From 13764510b87ac3a5894ed534c8d28d5bd982dc6b Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 09:21:01 +0000 Subject: [PATCH 01/13] Register enable-prometheus-cr-watcher flag and fix PrometheusCR watcher startup The target-allocator declared the enable-prometheus-cr-watcher flag name as a constant but never registered it on the flag set, while the operator passes --enable-prometheus-cr-watcher whenever PrometheusCR.enabled is true. Because args are parsed with pflag.ExitOnError, the unregistered flag caused the binary to print 'unknown flag' and exit(2), putting the target-allocator pod into CrashLoopBackOff. This change registers the flag and ORs it with the YAML prometheus_cr.enabled setting, then fixes three latent defects that were previously unreachable because the binary crashed first: - promOperator: set a non-empty Namespace on the synthetic Prometheus object so the prometheus-operator config generator no longer panics with 'namespace can't be empty' in store.ForNamespace. - promOperator: set EvaluationInterval so the generated config does not render an empty global.evaluation_interval, which the prometheus config parser rejects with 'empty duration string'. - main: create and register service-discovery metrics and pass them to discovery.NewManager; passing a nil sdMetrics map makes every SD provider fail to register, yielding zero discovered targets. RELEASE_NOTES updated. --- RELEASE_NOTES | 6 ++++++ .../config/config.go | 9 +++++++++ .../config/flags.go | 5 +++++ .../config/flags_test.go | 13 ++++++++++++ .../main.go | 13 +++++++++++- .../watcher/promOperator.go | 20 +++++++++++++++++++ 6 files changed, 65 insertions(+), 1 deletion(-) diff --git a/RELEASE_NOTES b/RELEASE_NOTES index 6128a2210..ff73e4d26 100644 --- a/RELEASE_NOTES +++ b/RELEASE_NOTES @@ -1,3 +1,9 @@ +======================================================================== +Amazon CloudWatch Agent Operator (Unreleased) +======================================================================== +Bug Fixes: +* [TargetAllocator] Register the `--enable-prometheus-cr-watcher` CLI flag so the target-allocator no longer exits with "unknown flag" (CrashLoopBackOff) when the operator passes it for PrometheusCR.enabled. The flag is OR'd with the YAML `prometheus_cr.enabled` setting. + ======================================================================== Amazon CloudWatch Agent Operator v3.5.0 (2026-06-05) ======================================================================== diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 2272eedbe..6b2a3d81d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -119,6 +119,15 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } + // Enable the Prometheus CR watcher when requested via the CLI flag. The YAML + // `prometheus_cr.enabled` value is loaded before this point, so OR the flag in + // rather than overwriting it: the watcher is enabled if either source sets it. + prometheusCREnabled, err := getPrometheusCREnabled(flagSet) + if err != nil { + return err + } + target.PrometheusCR.Enabled = target.PrometheusCR.Enabled || prometheusCREnabled + target.HTTPS.Enabled, err = getHttpsEnabled(flagSet) if err != nil { return err diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go index caf639ed9..294f9bfe6 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go @@ -34,6 +34,7 @@ func getFlagSet(errorHandling pflag.ErrorHandling) *pflag.FlagSet { flagSet := pflag.NewFlagSet(targetAllocatorName, errorHandling) flagSet.String(configFilePathFlagName, DefaultConfigFilePath, "The path to the config file.") flagSet.String(kubeConfigPathFlagName, filepath.Join(homedir.HomeDir(), ".kube", "config"), "absolute path to the KubeconfigPath file") + flagSet.Bool(prometheusCREnabledFlagName, false, "Enable watching of Prometheus Operator custom resources (ServiceMonitor/PodMonitor) to dynamically generate scrape configs.") flagSet.Bool(reloadConfigFlagName, false, "Enable automatic configuration reloading. This functionality is deprecated and will be removed in a future release.") flagSet.Bool(httpsEnabledFlagName, true, "Enable HTTPS additional server") flagSet.String(listenAddrHttpsFlagName, ":8443", "The address where this service serves over HTTPS.") @@ -58,6 +59,10 @@ func getConfigReloadEnabled(flagSet *pflag.FlagSet) (bool, error) { return flagSet.GetBool(reloadConfigFlagName) } +func getPrometheusCREnabled(flagSet *pflag.FlagSet) (bool, error) { + return flagSet.GetBool(prometheusCREnabledFlagName) +} + func getHttpsListenAddr(flagSet *pflag.FlagSet) (string, error) { return flagSet.GetString(listenAddrHttpsFlagName) } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go index 8beed29d0..66c71a268 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go @@ -17,6 +17,7 @@ func TestGetFlagSet(t *testing.T) { // Check if each flag exists assert.NotNil(t, fs.Lookup(configFilePathFlagName), "Flag %s not found", configFilePathFlagName) assert.NotNil(t, fs.Lookup(kubeConfigPathFlagName), "Flag %s not found", kubeConfigPathFlagName) + assert.NotNil(t, fs.Lookup(prometheusCREnabledFlagName), "Flag %s not found", prometheusCREnabledFlagName) } func TestFlagGetters(t *testing.T) { @@ -45,6 +46,18 @@ func TestFlagGetters(t *testing.T) { expectedValue: true, getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getConfigReloadEnabled(fs) }, }, + { + name: "GetPrometheusCREnabled", + flagArgs: []string{"--" + prometheusCREnabledFlagName}, + expectedValue: true, + getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) }, + }, + { + name: "GetPrometheusCREnabledDefault", + flagArgs: []string{}, + expectedValue: false, + getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) }, + }, { name: "InvalidFlag", flagArgs: []string{"--invalid-flag", "value"}, diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/main.go b/cmd/amazon-cloudwatch-agent-target-allocator/main.go index dbc956ebf..57185184d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/main.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/main.go @@ -88,7 +88,18 @@ func main() { srv := server.NewServer(log, allocator, cfg.ListenAddr, httpOptions...) discoveryCtx, discoveryCancel := context.WithCancel(ctx) - discoveryManager = discovery.NewManager(discoveryCtx, nil, prometheus.NewRegistry(), nil) + // Service Discovery metrics MUST be created and passed to the discovery + // manager. Each SD provider's NewDiscoverer fails when its DiscovererMetrics + // is nil, so passing a nil sdMetrics map makes every provider (including + // kubernetes_sd) fail to register, yielding zero discovered targets. Mirror + // the upstream opentelemetry-operator target-allocator. + sdRegistry := prometheus.NewRegistry() + sdMetrics, sdMetricsErr := discovery.CreateAndRegisterSDMetrics(sdRegistry) + if sdMetricsErr != nil { + setupLog.Error(sdMetricsErr, "Unable to register service discovery metrics") + os.Exit(1) + } + discoveryManager = discovery.NewManager(discoveryCtx, nil, sdRegistry, sdMetrics) targetDiscoverer = target.NewDiscoverer(log, discoveryManager, allocatorPrehook, srv) collectorWatcher, collectorWatcherErr := collector.NewClient(log, cfg.ClusterConfig) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 2651e8d29..a15db9d78 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -21,6 +21,7 @@ import ( kubeDiscovery "github.com/prometheus/prometheus/discovery/kubernetes" "gopkg.in/yaml.v2" v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" @@ -49,11 +50,30 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable + // The synthetic Prometheus object must carry a non-empty namespace: the + // prometheus-operator config generator calls store.ForNamespace(prom.Namespace) + // when generating the server configuration, which panics on an empty namespace + // ("namespace can't be empty"). Mirror the upstream opentelemetry-operator + // target-allocator by setting it to the collector/TA namespace, derived from the + // operator-injected OTELCOL_NAMESPACE downward-API env (fallback: amazon-cloudwatch). + collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") + if collectorNamespace == "" { + collectorNamespace = "amazon-cloudwatch" + } prom := &monitoringv1.Prometheus{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: collectorNamespace, + }, Spec: monitoringv1.PrometheusSpec{ CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, + // EvaluationInterval must be non-empty: the prometheus-operator config + // generator renders it verbatim as global.evaluation_interval, and the + // downstream prometheus config parser rejects an empty value with + // "empty duration string". The CWA TA does not evaluate rules, so mirror + // the scrape interval as a sane non-empty default. + EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } From 0405f4decb019dd2ee173333a3985ea517b4ffb5 Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 15:57:02 +0000 Subject: [PATCH 02/13] test(target-allocator): guard scrape_protocols defaulting on config load Add a regression test asserting that loading a Target Allocator config whose static scrape job omits scrape_protocols still yields a non-empty ScrapeProtocols on every loaded scrape config. This is defaulted by the pinned Prometheus library during yaml.UnmarshalStrict into the prometheus Config type, so the distributed /scrape_configs payload is never empty and the agent's prometheus-receiver validation passes. The test fails fast if a future dependency or load-path change drops this defaulting. --- .../scrape_protocols_regression_test.go | 36 +++++++++++++++++++ .../scrape_protocols_omitted_test.yaml | 14 ++++++++ 2 files changed, 50 insertions(+) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go new file mode 100644 index 000000000..9c432d7ee --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go @@ -0,0 +1,36 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestScrapeProtocolsDefaultedOnLoad guards against "scrape_protocols cannot be empty" regression. +func TestScrapeProtocolsDefaultedOnLoad(t *testing.T) { + got := CreateDefaultConfig() + err := LoadFromFile("./testdata/scrape_protocols_omitted_test.yaml", &got) + require.NoError(t, err) + + require.NotNil(t, got.PromConfig) + require.NotEmpty(t, got.PromConfig.ScrapeConfigs) + + for _, sc := range got.PromConfig.ScrapeConfigs { + assert.NotEmpty(t, sc.ScrapeProtocols, + "scrape_protocols must be defaulted for job %q", sc.JobName) + } + + // Spot-check the specific static job from the reproduction by name. + found := false + for _, sc := range got.PromConfig.ScrapeConfigs { + if sc.JobName == "prometheus-sample-app" { + found = true + assert.Greater(t, len(sc.ScrapeProtocols), 0) + } + } + require.True(t, found, "expected the 'prometheus-sample-app' job to be present") +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml new file mode 100644 index 000000000..9405f61e8 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml @@ -0,0 +1,14 @@ +# Regression fixture: a scrape job that omits scrape_protocols. +# The TA config-load path must default scrape_protocols so the agent does not +# reject the config with "scrape_protocols cannot be empty". +config: + scrape_configs: + - job_name: prometheus-sample-app + kubernetes_sd_configs: + - role: pod + relabel_configs: + - source_labels: [__meta_kubernetes_pod_label_app] + regex: prometheus-sample-app + action: keep + - target_label: label1 + replacement: value1 From d61d693fa6f0536786951e8e4d08f378a5e77c82 Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 16:01:54 +0000 Subject: [PATCH 03/13] 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 efd17bbb60f9a76cd71fad8480b537dbf0808e02 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 26 Jun 2026 14:21:08 +0100 Subject: [PATCH 04/13] feat(target-allocator): add per-node allocation strategy with fallback Add a per-node allocation strategy so each CloudWatch agent (DaemonSet, one per node) scrapes only the ServiceMonitor/PodMonitor targets on its own node, eliminating cross-node/cross-AZ scrape traffic. Targets that cannot be matched to a node-local agent (node-less endpoints, nodes without a Ready agent) fall back to consistent-hashing so they are never silently dropped. - allocation/per_node.go: perNodeAllocator (node index, consistent-hashing fallback ring, unassigned tracking, descriptive logging) + registration. - allocation/strategy.go: Collector.NodeName, NewCollector(name, node), WithFallbackStrategy option, targets_unassigned gauge. - collector/collector.go: capture pod.Spec.NodeName, skip empty-NodeName pods, handle watch.Modified so a collector's node is picked up once scheduled (fixes targets being stuck on the fallback after a DaemonSet rollout). - target/target.go: GetNodeName() from __meta_kubernetes_*_node_name labels. - config + main: FallbackAllocationStrategy wiring. - apis/v1alpha1 + CRD: allocationStrategy enum gains "per-node". - internal/manifests/targetallocator/configmap.go: emit per-node strategy and the consistent-hashing fallback from the CR. --- apis/v1alpha1/allocation_strategy.go | 6 +- .../allocation/consistent_hashing.go | 2 +- .../allocation/per_node.go | 396 ++++++++++++++++++ .../allocation/per_node_test.go | 161 +++++++ .../allocation/strategy.go | 39 +- .../allocation/strategy_test.go | 10 +- .../collector/collector.go | 18 +- .../collector/collector_test.go | 15 +- .../config/config.go | 11 + .../main.go | 4 +- .../target/target.go | 32 ++ ...aws.amazon.com_amazoncloudwatchagents.yaml | 3 +- .../manifests/targetallocator/configmap.go | 13 +- .../targetallocator/configmap_test.go | 37 ++ 14 files changed, 727 insertions(+), 20 deletions(-) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go diff --git a/apis/v1alpha1/allocation_strategy.go b/apis/v1alpha1/allocation_strategy.go index f3e3be140..515b03ced 100644 --- a/apis/v1alpha1/allocation_strategy.go +++ b/apis/v1alpha1/allocation_strategy.go @@ -5,11 +5,15 @@ package v1alpha1 type ( // AmazonCloudWatchAgentTargetAllocatorAllocationStrategy represent which strategy to distribute target to each collector - // +kubebuilder:validation:Enum=consistent-hashing + // +kubebuilder:validation:Enum=consistent-hashing;per-node AmazonCloudWatchAgentTargetAllocatorAllocationStrategy string ) const ( // AmazonCloudWatchAgentTargetAllocatorAllocationStrategyConsistentHashing targets will be consistently added to collectors, which allows a high-availability setup. AmazonCloudWatchAgentTargetAllocatorAllocationStrategyConsistentHashing AmazonCloudWatchAgentTargetAllocatorAllocationStrategy = "consistent-hashing" + + // AmazonCloudWatchAgentTargetAllocatorAllocationStrategyPerNode targets will be allocated to the collector running on the same node as the target. + // Targets without a resolvable node fall back to the configured fallback strategy (consistent-hashing). + AmazonCloudWatchAgentTargetAllocatorAllocationStrategyPerNode AmazonCloudWatchAgentTargetAllocatorAllocationStrategy = "per-node" ) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/consistent_hashing.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/consistent_hashing.go index 4bff39eee..171eae67a 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/consistent_hashing.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/consistent_hashing.go @@ -150,7 +150,7 @@ func (c *consistentHashingAllocator) handleCollectors(diff diff.Changes[*Collect } // Insert the new collectors for _, i := range diff.Additions() { - c.collectors[i.Name] = NewCollector(i.Name) + c.collectors[i.Name] = NewCollector(i.Name, i.NodeName) c.consistentHasher.Add(c.collectors[i.Name]) } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go new file mode 100644 index 000000000..4d1a1b6c6 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go @@ -0,0 +1,396 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package allocation + +import ( + "sort" + "strings" + "sync" + + "github.com/buraksezer/consistent" + "github.com/go-logr/logr" + "github.com/prometheus/client_golang/prometheus" + + "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/diff" + "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/target" +) + +var _ Allocator = &perNodeAllocator{} + +const perNodeStrategyName = "per-node" + +// placement is the outcome of allocating a single target, used for per-cycle +// log summaries describing what the per-node strategy did. +type placement int + +const ( + placedByNode placement = iota // assigned to the collector on the target's node + placedByFallback // assigned via the consistent-hashing fallback + unplaced // left unassigned (no node match, no fallback) +) + +// perNodeAllocator assigns each target to the collector running on the same +// Kubernetes node as the target. It mirrors the node-lookup logic of the +// upstream OpenTelemetry target allocator's per-node strategy +// (cmd/otel-allocator/internal/allocation/per_node.go), adapted to this fork's +// fused Allocator model (see consistent_hashing.go where the allocator and the +// placement strategy are a single type). +// +// Placement: a target's node comes from target.Item.GetNodeName() (the +// __meta_kubernetes_*_node_name discovery labels); a collector's node comes from +// Collector.NodeName (pod.Spec.NodeName, captured by the collector watcher). +// +// Unassigned targets: targets that carry no node (GetNodeName == "") or whose +// node has no matching collector are placed via the fallback strategy when one +// is configured (SetFallbackStrategy("consistent-hashing")); otherwise they are +// retained in targetItems but assigned to no collector, re-evaluated on every +// collector change, and surfaced via the +// cloudwatch_agent_allocator_targets_unassigned gauge. +type perNodeAllocator struct { + // m protects collectors, targetItems, targetItemsPerJobPerCollector, + // collectorByNode and fallbackHasher for concurrent use. + m sync.RWMutex + + // collectors is a map from a Collector's name to a Collector instance. + collectors map[string]*Collector + + // targetItems is a map from a target item's hash to the target item. + targetItems map[string]*target.Item + + // collectorKey -> job -> target item hash -> true + targetItemsPerJobPerCollector map[string]map[string]map[string]bool + + // collectorByNode indexes collectors by their NodeName for O(1) placement. + collectorByNode map[string]*Collector + + // fallbackHasher, when non-nil, is a consistent-hashing ring over all + // collectors used to place targets that cannot be matched to a node. + fallbackHasher *consistent.Consistent + + log logr.Logger + + filter Filter +} + +func newPerNodeAllocator(log logr.Logger, opts ...AllocationOption) Allocator { + pnAllocator := &perNodeAllocator{ + collectors: make(map[string]*Collector), + targetItems: make(map[string]*target.Item), + targetItemsPerJobPerCollector: make(map[string]map[string]map[string]bool), + collectorByNode: make(map[string]*Collector), + log: log, + } + for _, opt := range opts { + opt(pnAllocator) + } + return pnAllocator +} + +// SetFilter sets the filtering hook to use. +func (pn *perNodeAllocator) SetFilter(filter Filter) { + pn.filter = filter +} + +// SetFallbackStrategy enables a fallback placement strategy for targets that +// cannot be matched to a node. Only "consistent-hashing" is supported; any other +// (or empty) name leaves the fallback disabled, so unmatched targets stay +// unassigned. Mirrors the upstream per-node strategy's optional fallbackStrategy. +func (pn *perNodeAllocator) SetFallbackStrategy(name string) { + pn.m.Lock() + defer pn.m.Unlock() + if name != consistentHashingStrategyName { + pn.log.Info("Unsupported fallback strategy for per-node, fallback disabled", "fallback", name) + pn.fallbackHasher = nil + return + } + cfg := consistent.Config{ + PartitionCount: 1061, + ReplicationFactor: 5, + Load: 1.1, + Hasher: hasher{}, + } + pn.fallbackHasher = consistent.New(nil, cfg) + // Seed the ring with any collectors already known. + for _, c := range pn.collectors { + pn.fallbackHasher.Add(c) + } + pn.log.Info("Per-node fallback strategy enabled", "fallback", name) +} + +// addCollectorTargetItemMapping tracks which collector has which jobs and targets. +// The caller has to acquire a lock. +func (pn *perNodeAllocator) addCollectorTargetItemMapping(tg *target.Item) { + if pn.targetItemsPerJobPerCollector[tg.CollectorName] == nil { + pn.targetItemsPerJobPerCollector[tg.CollectorName] = make(map[string]map[string]bool) + } + if pn.targetItemsPerJobPerCollector[tg.CollectorName][tg.JobName] == nil { + pn.targetItemsPerJobPerCollector[tg.CollectorName][tg.JobName] = make(map[string]bool) + } + pn.targetItemsPerJobPerCollector[tg.CollectorName][tg.JobName][tg.Hash()] = true +} + +// addTargetToTargetItems assigns a target to the collector on the same node and +// stores it in targetItems. The caller has to acquire a lock. Targets with no +// resolvable node, or whose node has no matching collector, are placed via the +// consistent-hashing fallback if configured, otherwise left unassigned. +// It returns how the target was placed, for per-cycle log summaries. +func (pn *perNodeAllocator) addTargetToTargetItems(tg *target.Item) placement { + // If this is a reassignment, decrement the previous collector's NumTargets. + if previousCol, ok := pn.collectors[tg.CollectorName]; ok && tg.CollectorName != "" { + previousCol.NumTargets-- + delete(pn.targetItemsPerJobPerCollector[tg.CollectorName][tg.JobName], tg.Hash()) + TargetsPerCollector.WithLabelValues(previousCol.String(), perNodeStrategyName).Set(float64(previousCol.NumTargets)) + } + + // Always keep the target in the pool so it can be (re)assigned later. + tg.CollectorName = "" + pn.targetItems[tg.Hash()] = tg + + nodeName := tg.GetNodeName() + colOwner, ok := pn.collectorByNode[nodeName] + if nodeName != "" && ok { + tg.CollectorName = colOwner.Name + pn.addCollectorTargetItemMapping(tg) + colOwner.NumTargets++ + TargetsPerCollector.WithLabelValues(colOwner.String(), perNodeStrategyName).Set(float64(colOwner.NumTargets)) + pn.log.V(2).Info("per-node: assigned target to node-local collector", + "target", strings.Join(tg.TargetURL, ","), "job", tg.JobName, "node", nodeName, "collector", colOwner.Name) + return placedByNode + } + + // No node, or no collector on that node. Use the consistent-hashing fallback + // if configured; otherwise leave the target unassigned. + if pn.fallbackHasher != nil && len(pn.collectors) > 0 { + member := pn.fallbackHasher.LocateKey([]byte(strings.Join(tg.TargetURL, ""))) + if fallbackCol, exists := pn.collectors[member.String()]; exists { + tg.CollectorName = fallbackCol.Name + pn.addCollectorTargetItemMapping(tg) + fallbackCol.NumTargets++ + TargetsPerCollector.WithLabelValues(fallbackCol.String(), perNodeStrategyName).Set(float64(fallbackCol.NumTargets)) + reason := "target has no node label" + if nodeName != "" { + reason = "no collector running on node " + nodeName + } + pn.log.V(1).Info("per-node: no node-local collector, used consistent-hashing fallback", + "target", strings.Join(tg.TargetURL, ","), "job", tg.JobName, "node", nodeName, "collector", fallbackCol.Name, "reason", reason) + return placedByFallback + } + } + pn.log.V(1).Info("per-node: target left UNASSIGNED (no node-local collector and no usable fallback)", + "target", strings.Join(tg.TargetURL, ","), "job", tg.JobName, "node", nodeName) + return unplaced +} + +// handleTargets reconciles added and removed targets against the current state. +func (pn *perNodeAllocator) handleTargets(diff diff.Changes[*target.Item]) { + // Check for removals. + for k, item := range pn.targetItems { + if _, ok := diff.Removals()[k]; ok { + if col, ok := pn.collectors[item.CollectorName]; ok && item.CollectorName != "" { + col.NumTargets-- + delete(pn.targetItemsPerJobPerCollector[item.CollectorName][item.JobName], item.Hash()) + TargetsPerCollector.WithLabelValues(item.CollectorName, perNodeStrategyName).Set(float64(col.NumTargets)) + } + delete(pn.targetItems, k) + } + } + + // Check for additions. + var byNode, byFallback, unassigned, added int + for k, item := range diff.Additions() { + if _, ok := pn.targetItems[k]; ok { + continue + } + added++ + switch pn.addTargetToTargetItems(item) { + case placedByNode: + byNode++ + case placedByFallback: + byFallback++ + case unplaced: + unassigned++ + } + } + + if added > 0 || len(diff.Removals()) > 0 { + pn.log.Info("per-node: target reconcile complete", + "added", added, "removed", len(diff.Removals()), + "assigned_by_node", byNode, "assigned_by_fallback", byFallback, "unassigned", unassigned, + "total_targets", len(pn.targetItems), "collectors", len(pn.collectors)) + } + + pn.recordUnassigned() +} + +// handleCollectors reconciles added and removed collectors, rebuilds the node +// index and re-allocates all known targets. +func (pn *perNodeAllocator) handleCollectors(diff diff.Changes[*Collector]) { + // Clear removed collectors. + for _, k := range diff.Removals() { + delete(pn.collectors, k.Name) + delete(pn.targetItemsPerJobPerCollector, k.Name) + if pn.fallbackHasher != nil { + pn.fallbackHasher.Remove(k.Name) + } + TargetsPerCollector.WithLabelValues(k.Name, perNodeStrategyName).Set(0) + } + // Insert the new collectors. + for _, i := range diff.Additions() { + pn.collectors[i.Name] = NewCollector(i.Name, i.NodeName) + if pn.fallbackHasher != nil { + pn.fallbackHasher.Add(pn.collectors[i.Name]) + } + } + + // Rebuild the node index from the current collector set. + pn.collectorByNode = make(map[string]*Collector) + for _, c := range pn.collectors { + if c.NodeName != "" { + pn.collectorByNode[c.NodeName] = c + } + } + + // Log the node->collector index so it's clear which node each agent owns. + mapping := make([]string, 0, len(pn.collectorByNode)) + for node, c := range pn.collectorByNode { + mapping = append(mapping, node+"="+c.Name) + } + sort.Strings(mapping) + noNode := len(pn.collectors) - len(pn.collectorByNode) + pn.log.Info("per-node: collector node index rebuilt", + "collectors", len(pn.collectors), "nodes_indexed", len(pn.collectorByNode), + "collectors_without_node", noNode, "mapping", strings.Join(mapping, ",")) + + // Re-allocate all targets against the new collector set. + var byNode, byFallback, unassigned int + for _, item := range pn.targetItems { + switch pn.addTargetToTargetItems(item) { + case placedByNode: + byNode++ + case placedByFallback: + byFallback++ + case unplaced: + unassigned++ + } + } + pn.log.Info("per-node: re-allocated all targets after collector change", + "added_collectors", len(diff.Additions()), "removed_collectors", len(diff.Removals()), + "assigned_by_node", byNode, "assigned_by_fallback", byFallback, "unassigned", unassigned, + "total_targets", len(pn.targetItems)) + + pn.recordUnassigned() +} + +// recordUnassigned updates the unassigned-targets gauge. Caller must hold the lock. +func (pn *perNodeAllocator) recordUnassigned() { + var count float64 + for _, item := range pn.targetItems { + if item.CollectorName == "" { + count++ + } + } + targetsUnassigned.Set(count) +} + +// SetTargets accepts a list of targets that will be used to make load balancing +// decisions. This method should be called when there are new targets discovered +// or existing targets are shutdown. +func (pn *perNodeAllocator) SetTargets(targets map[string]*target.Item) { + timer := prometheus.NewTimer(TimeToAssign.WithLabelValues("SetTargets", perNodeStrategyName)) + defer timer.ObserveDuration() + + if pn.filter != nil { + targets = pn.filter.Apply(targets) + } + RecordTargetsKept(targets) + + pn.m.Lock() + defer pn.m.Unlock() + + // If there are no collectors, just track the targets so they can be assigned + // once collectors appear. + if len(pn.collectors) == 0 { + pn.log.Info("No collector instances present, saving targets to allocate to collector(s)") + targetsDiff := diff.Maps(pn.targetItems, targets) + for k, item := range targetsDiff.Additions() { + if _, ok := pn.targetItems[k]; !ok { + pn.targetItems[k] = item + } + } + for k := range targetsDiff.Removals() { + delete(pn.targetItems, k) + } + pn.recordUnassigned() + return + } + + // Check for target changes. + targetsDiff := diff.Maps(pn.targetItems, targets) + if len(targetsDiff.Additions()) != 0 || len(targetsDiff.Removals()) != 0 { + pn.handleTargets(targetsDiff) + } +} + +// SetCollectors sets the set of collectors with key=collectorName, value=Collector object. +// This method is called when Collectors are added or removed. +func (pn *perNodeAllocator) SetCollectors(collectors map[string]*Collector) { + timer := prometheus.NewTimer(TimeToAssign.WithLabelValues("SetCollectors", perNodeStrategyName)) + defer timer.ObserveDuration() + + CollectorsAllocatable.WithLabelValues(perNodeStrategyName).Set(float64(len(collectors))) + if len(collectors) == 0 { + pn.log.Info("No collector instances present") + return + } + + pn.m.Lock() + defer pn.m.Unlock() + + // Check for collector changes. + collectorsDiff := diff.Maps(pn.collectors, collectors) + if len(collectorsDiff.Additions()) != 0 || len(collectorsDiff.Removals()) != 0 { + pn.handleCollectors(collectorsDiff) + } + pn.log.Info("Setting collector completed") +} + +func (pn *perNodeAllocator) GetTargetsForCollectorAndJob(collector string, job string) []*target.Item { + pn.m.RLock() + defer pn.m.RUnlock() + if _, ok := pn.targetItemsPerJobPerCollector[collector]; !ok { + return []*target.Item{} + } + if _, ok := pn.targetItemsPerJobPerCollector[collector][job]; !ok { + return []*target.Item{} + } + targetItemsCopy := make([]*target.Item, len(pn.targetItemsPerJobPerCollector[collector][job])) + index := 0 + for targetHash := range pn.targetItemsPerJobPerCollector[collector][job] { + targetItemsCopy[index] = pn.targetItems[targetHash] + index++ + } + return targetItemsCopy +} + +// TargetItems returns a shallow copy of the targetItems map. +func (pn *perNodeAllocator) TargetItems() map[string]*target.Item { + pn.m.RLock() + defer pn.m.RUnlock() + targetItemsCopy := make(map[string]*target.Item) + for k, v := range pn.targetItems { + targetItemsCopy[k] = v + } + return targetItemsCopy +} + +// Collectors returns a shallow copy of the collectors map. +func (pn *perNodeAllocator) Collectors() map[string]*Collector { + pn.m.RLock() + defer pn.m.RUnlock() + collectorsCopy := make(map[string]*Collector) + for k, v := range pn.collectors { + collectorsCopy[k] = v + } + return collectorsCopy +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go new file mode 100644 index 000000000..67fa3e2d8 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go @@ -0,0 +1,161 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package allocation + +import ( + "testing" + + "github.com/prometheus/common/model" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/target" +) + +// nodeTarget builds a target carrying the pod node-name discovery label. +func nodeTarget(job, url, node string) *target.Item { + lbls := model.LabelSet{} + if node != "" { + lbls[model.LabelName("__meta_kubernetes_pod_node_name")] = model.LabelValue(node) + } + return target.NewItem(job, url, lbls, "") +} + +func newPerNodeTestAllocator() *perNodeAllocator { + return newPerNodeAllocator(logger).(*perNodeAllocator) +} + +// TestPerNodeRegistered ensures the strategy is wired into the registry. +func TestPerNodeRegistered(t *testing.T) { + a, err := New(perNodeStrategyName, logger) + require.NoError(t, err) + require.NotNil(t, a) +} + +// TestPerNodeAssignsToMatchingNode verifies each target lands on the collector +// running on the target's node, and only that collector. +func TestPerNodeAssignsToMatchingNode(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + "collector-b": NewCollector("collector-b", "node-b"), + }) + + targets := map[string]*target.Item{} + for _, tg := range []*target.Item{ + nodeTarget("job1", "10.0.0.1:8080", "node-a"), + nodeTarget("job1", "10.0.0.2:8080", "node-a"), + nodeTarget("job1", "10.0.0.3:8080", "node-b"), + } { + targets[tg.Hash()] = tg + } + c.SetTargets(targets) + + assert.Len(t, c.TargetItems(), 3) + + aTargets := c.GetTargetsForCollectorAndJob("collector-a", "job1") + bTargets := c.GetTargetsForCollectorAndJob("collector-b", "job1") + assert.Len(t, aTargets, 2, "node-a collector should own both node-a targets") + assert.Len(t, bTargets, 1, "node-b collector should own the single node-b target") + + for _, ti := range aTargets { + assert.Equal(t, "collector-a", ti.CollectorName) + assert.Equal(t, "node-a", ti.GetNodeName()) + } + for _, ti := range bTargets { + assert.Equal(t, "collector-b", ti.CollectorName) + assert.Equal(t, "node-b", ti.GetNodeName()) + } +} + +// TestPerNodeLeavesNodelessTargetUnassigned verifies a target without a node +// label is retained but assigned to no collector. +func TestPerNodeLeavesNodelessTargetUnassigned(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + + onNode := nodeTarget("job1", "10.0.0.1:8080", "node-a") + noNode := nodeTarget("job1", "10.0.0.9:8080", "") // external/non-pod endpoint + c.SetTargets(map[string]*target.Item{ + onNode.Hash(): onNode, + noNode.Hash(): noNode, + }) + + // Both are tracked. + assert.Len(t, c.TargetItems(), 2) + // Only the node-matched target is assigned to the collector. + assigned := c.GetTargetsForCollectorAndJob("collector-a", "job1") + assert.Len(t, assigned, 1) + assert.Equal(t, onNode.Hash(), assigned[0].Hash()) + + // The node-less target carries no collector. + for _, ti := range c.TargetItems() { + if ti.Hash() == noNode.Hash() { + assert.Equal(t, "", ti.CollectorName, "node-less target must stay unassigned") + } + } +} + +// TestPerNodeFallbackAssignsNodelessTarget verifies that, with a consistent-hashing +// fallback configured, a target without a node is still placed on some collector +// rather than left unassigned. +func TestPerNodeFallbackAssignsNodelessTarget(t *testing.T) { + a, err := New(perNodeStrategyName, logger, WithFallbackStrategy(consistentHashingStrategyName)) + require.NoError(t, err) + c := a.(*perNodeAllocator) + + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + "collector-b": NewCollector("collector-b", "node-b"), + }) + + onNode := nodeTarget("job1", "10.0.0.1:8080", "node-a") + noNode := nodeTarget("job1", "10.0.0.9:8080", "") // external/non-pod endpoint + c.SetTargets(map[string]*target.Item{ + onNode.Hash(): onNode, + noNode.Hash(): noNode, + }) + + // node-matched target lands on its node's collector. + assert.Equal(t, "collector-a", onNode.CollectorName) + + // node-less target is assigned to *some* collector via the fallback (not ""). + var found *target.Item + for _, ti := range c.TargetItems() { + if ti.Hash() == noNode.Hash() { + found = ti + } + } + require.NotNil(t, found) + assert.NotEqual(t, "", found.CollectorName, "node-less target must be placed by the consistent-hashing fallback") + _, isRealCollector := c.Collectors()[found.CollectorName] + assert.True(t, isRealCollector, "fallback must assign to a known collector") +} + +// target (its node had no collector) gets placed once a matching collector joins. +func TestPerNodeReallocatesWhenCollectorAppears(t *testing.T) { + c := newPerNodeTestAllocator() + // Only node-a has a collector initially. + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + + onB := nodeTarget("job1", "10.0.0.3:8080", "node-b") + c.SetTargets(map[string]*target.Item{onB.Hash(): onB}) + + // node-b target cannot be placed yet. + assert.Empty(t, c.GetTargetsForCollectorAndJob("collector-b", "job1")) + + // node-b collector joins; target should now be assigned to it. + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + "collector-b": NewCollector("collector-b", "node-b"), + }) + + bTargets := c.GetTargetsForCollectorAndJob("collector-b", "job1") + require.Len(t, bTargets, 1) + assert.Equal(t, "collector-b", bTargets[0].CollectorName) +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go index 5f5285ece..9683191ba 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go @@ -38,6 +38,14 @@ var ( Name: "cloudwatch_agent_allocator_targets_remaining", Help: "Number of targets kept after filtering.", }) + // targetsUnassigned records targets that could not be placed on any collector + // (e.g. per-node strategy: target has no node label, or no collector runs on + // the target's node). Such targets are retained and re-evaluated on the next + // collector change, but are not scraped until they can be assigned. + targetsUnassigned = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "cloudwatch_agent_allocator_targets_unassigned", + Help: "Number of targets that could not be assigned to a collector.", + }) ) type AllocationOption func(Allocator) @@ -52,6 +60,26 @@ func WithFilter(filter Filter) AllocationOption { } } +// fallbackStrategySetter is implemented by allocators that support a fallback +// placement strategy for targets the primary strategy cannot assign. +type fallbackStrategySetter interface { + SetFallbackStrategy(name string) +} + +// WithFallbackStrategy configures a fallback placement strategy by name. It is a +// no-op for an empty name or for allocators that don't support a fallback +// (only per-node does today). +func WithFallbackStrategy(name string) AllocationOption { + return func(allocator Allocator) { + if name == "" { + return + } + if setter, ok := allocator.(fallbackStrategySetter); ok { + setter.SetFallbackStrategy(name) + } + } +} + func RecordTargetsKept(targets map[string]*target.Item) { targetsRemaining.Add(float64(len(targets))) } @@ -92,9 +120,12 @@ var _ consistent.Member = Collector{} // Collector Creates a struct that holds Collector information. // This struct will be parsed into endpoint with Collector and jobs info. +// NodeName is the Kubernetes node the collector pod runs on; it is used by the +// per-node allocation strategy to match targets to the collector on their node. // This struct can be extended with information like annotations and labels in the future. type Collector struct { Name string + NodeName string NumTargets int } @@ -106,8 +137,8 @@ func (c Collector) String() string { return c.Name } -func NewCollector(name string) *Collector { - return &Collector{Name: name} +func NewCollector(name, node string) *Collector { + return &Collector{Name: name, NodeName: node} } func init() { @@ -115,4 +146,8 @@ func init() { if err != nil { panic(err) } + err = Register(perNodeStrategyName, newPerNodeAllocator) + if err != nil { + panic(err) + } } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go index 0437ed021..feb9658ab 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go @@ -78,11 +78,11 @@ func Benchmark_Setting(b *testing.B) { } func TestCollectorDiff(t *testing.T) { - collector0 := NewCollector("collector-0") - collector1 := NewCollector("collector-1") - collector2 := NewCollector("collector-2") - collector3 := NewCollector("collector-3") - collector4 := NewCollector("collector-4") + collector0 := NewCollector("collector-0", "") + collector1 := NewCollector("collector-1", "") + collector2 := NewCollector("collector-2", "") + collector3 := NewCollector("collector-3", "") + collector4 := NewCollector("collector-4", "") type args struct { current map[string]*Collector new map[string]*Collector diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go index 60bda6bca..e9a3d6e10 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go @@ -65,8 +65,11 @@ func (k *Client) Watch(ctx context.Context, labelMap map[string]string, fn func( } for i := range pods.Items { pod := pods.Items[i] - if pod.GetObjectMeta().GetDeletionTimestamp() == nil { - collectorMap[pod.Name] = allocation.NewCollector(pod.Name) + // Only register a collector once its pod is scheduled (NodeName set); an + // unscheduled pod has no node, so the per-node strategy could not match + // targets to it. It is picked up later via the Modified watch event. + if pod.GetObjectMeta().GetDeletionTimestamp() == nil && pod.Spec.NodeName != "" { + collectorMap[pod.Name] = allocation.NewCollector(pod.Name, pod.Spec.NodeName) } } fn(collectorMap) @@ -117,8 +120,15 @@ func runWatch(ctx context.Context, k *Client, c <-chan watch.Event, collectorMap } switch event.Type { //nolint:exhaustive - case watch.Added: - collectorMap[pod.Name] = allocation.NewCollector(pod.Name) + case watch.Added, watch.Modified: + // Register/refresh the collector only once its pod is scheduled + // (NodeName set). Handling Modified captures the node when a pod that + // was Added while still unscheduled is later assigned to a node, so the + // per-node strategy can match this collector's node without needing a + // Target Allocator restart after agent (re)scheduling. + if pod.Spec.NodeName != "" { + collectorMap[pod.Name] = allocation.NewCollector(pod.Name, pod.Spec.NodeName) + } case watch.Deleted: delete(collectorMap, pod.Name) } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go index 76fb0b8b9..3b19351a3 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go @@ -58,6 +58,9 @@ func pod(name string) *v1.Pod { Namespace: "test-ns", Labels: labelSet, }, + Spec: v1.PodSpec{ + NodeName: name + "-node", + }, } } @@ -86,13 +89,16 @@ func Test_runWatch(t *testing.T) { }, want: map[string]*allocation.Collector{ "test-pod1": { - Name: "test-pod1", + Name: "test-pod1", + NodeName: "test-pod1-node", }, "test-pod2": { - Name: "test-pod2", + Name: "test-pod2", + NodeName: "test-pod2-node", }, "test-pod3": { - Name: "test-pod3", + Name: "test-pod3", + NodeName: "test-pod3-node", }, }, }, @@ -120,7 +126,8 @@ func Test_runWatch(t *testing.T) { }, want: map[string]*allocation.Collector{ "test-pod1": { - Name: "test-pod1", + Name: "test-pod1", + NodeName: "test-pod1-node", }, }, }, diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 081e646e9..23a29419d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -50,6 +50,7 @@ type Config struct { LabelSelector map[string]string `yaml:"label_selector,omitempty"` PromConfig *promconfig.Config `yaml:"config"` AllocationStrategy *string `yaml:"allocation_strategy,omitempty"` + FallbackAllocationStrategy *string `yaml:"allocation_fallback_strategy,omitempty"` FilterStrategy *string `yaml:"filter_strategy,omitempty"` PrometheusCR PrometheusCRConfig `yaml:"prometheus_cr,omitempty"` PodMonitorSelector map[string]string `yaml:"pod_monitor_selector,omitempty"` @@ -78,6 +79,16 @@ func (c Config) GetAllocationStrategy() string { return DefaultAllocationStrategy } +// GetAllocationFallbackStrategy returns the strategy used to place targets that +// the primary strategy cannot assign (e.g. per-node targets with no node match). +// Empty means no fallback (such targets are left unassigned). +func (c Config) GetAllocationFallbackStrategy() string { + if c.FallbackAllocationStrategy != nil { + return *c.FallbackAllocationStrategy + } + return "" +} + func (c Config) GetTargetsFilterStrategy() string { if c.FilterStrategy != nil { return *c.FilterStrategy diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/main.go b/cmd/amazon-cloudwatch-agent-target-allocator/main.go index 2b553dce6..af7bb0fae 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/main.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/main.go @@ -72,7 +72,9 @@ func main() { log := ctrl.Log.WithName("allocator") allocatorPrehook = prehook.New(cfg.GetTargetsFilterStrategy(), log) - allocator, err = allocation.New(cfg.GetAllocationStrategy(), log, allocation.WithFilter(allocatorPrehook)) + allocator, err = allocation.New(cfg.GetAllocationStrategy(), log, + allocation.WithFilter(allocatorPrehook), + allocation.WithFallbackStrategy(cfg.GetAllocationFallbackStrategy())) if err != nil { setupLog.Error(err, "Unable to initialize allocation strategy") os.Exit(1) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go b/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go index fb49e96ec..db5aa1d4f 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go @@ -10,6 +10,19 @@ import ( "github.com/prometheus/common/model" ) +// nodeLabels are the discovery meta-labels that identify the node a target +// resides on. They mirror the upstream OpenTelemetry target allocator's +// per-node node-label set. See: +// https://prometheus.io/docs/prometheus/latest/configuration/configuration/#kubernetes_sd_config +const ( + nodeNameLabelPod model.LabelName = "__meta_kubernetes_pod_node_name" + nodeNameLabelNode model.LabelName = "__meta_kubernetes_node_name" + nodeNameLabelEndpoint model.LabelName = "__meta_kubernetes_endpoint_node_name" + + endpointSliceTargetKindLabel model.LabelName = "__meta_kubernetes_endpointslice_address_target_kind" + endpointSliceTargetNameLabel model.LabelName = "__meta_kubernetes_endpointslice_address_target_name" +) + // LinkJSON This package contains common structs and methods that relate to scrape targets. type LinkJSON struct { Link string `json:"_link"` @@ -28,6 +41,25 @@ func (t *Item) Hash() string { return t.hash } +// GetNodeName returns the Kubernetes node a target resides on, derived from its +// service-discovery meta labels. Pod targets (PodMonitor, role: pod) always carry +// a node; endpoint targets (ServiceMonitor, role: endpoints/endpointslice) only +// carry one when the endpoint is backed by a pod on a node. Returns "" when no +// node can be determined (e.g. non-pod / external endpoints), in which case the +// per-node strategy leaves the target unassigned. +func (t *Item) GetNodeName() string { + for _, labelName := range []model.LabelName{nodeNameLabelPod, nodeNameLabelNode, nodeNameLabelEndpoint} { + if val := t.Labels[labelName]; val != "" { + return string(val) + } + } + + if t.Labels[endpointSliceTargetKindLabel] != "Node" { + return "" + } + return string(t.Labels[endpointSliceTargetNameLabel]) +} + // NewItem Creates a new target item. // INVARIANTS: // * Item fields must not be modified after creation. diff --git a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml index 6df20a0ff..65f88712d 100644 --- a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml +++ b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml @@ -6405,9 +6405,10 @@ spec: allocationStrategy: description: |- AllocationStrategy determines which strategy the target allocator should use for allocation. - The current option is consistent-hashing. + The options are consistent-hashing and per-node. enum: - consistent-hashing + - per-node type: string enabled: description: Enabled indicates whether to use a target allocation diff --git a/internal/manifests/targetallocator/configmap.go b/internal/manifests/targetallocator/configmap.go index ff5160f53..94e4e72da 100644 --- a/internal/manifests/targetallocator/configmap.go +++ b/internal/manifests/targetallocator/configmap.go @@ -51,7 +51,18 @@ func ConfigMap(params manifests.Params) (*corev1.ConfigMap, error) { taConfig["config"] = prometheusConfig } - taConfig["allocation_strategy"] = v1alpha1.AmazonCloudWatchAgentTargetAllocatorAllocationStrategyConsistentHashing + // Use the strategy from the CR if set, defaulting to consistent-hashing to + // preserve prior behavior. When per-node is selected, configure a + // consistent-hashing fallback so targets without a resolvable node (e.g. + // non-pod ServiceMonitor endpoints) are still allocated rather than dropped. + allocationStrategy := params.OtelCol.Spec.TargetAllocator.AllocationStrategy + if allocationStrategy == "" { + allocationStrategy = v1alpha1.AmazonCloudWatchAgentTargetAllocatorAllocationStrategyConsistentHashing + } + taConfig["allocation_strategy"] = allocationStrategy + if allocationStrategy == v1alpha1.AmazonCloudWatchAgentTargetAllocatorAllocationStrategyPerNode { + taConfig["allocation_fallback_strategy"] = v1alpha1.AmazonCloudWatchAgentTargetAllocatorAllocationStrategyConsistentHashing + } if len(params.OtelCol.Spec.TargetAllocator.FilterStrategy) > 0 { taConfig["filter_strategy"] = params.OtelCol.Spec.TargetAllocator.FilterStrategy diff --git a/internal/manifests/targetallocator/configmap_test.go b/internal/manifests/targetallocator/configmap_test.go index 90f76948b..56d009980 100644 --- a/internal/manifests/targetallocator/configmap_test.go +++ b/internal/manifests/targetallocator/configmap_test.go @@ -145,5 +145,42 @@ prometheus_cr: assert.Equal(t, expectedData, actual.Data) }) + t.Run("should emit per-node strategy with consistent-hashing fallback", func(t *testing.T) { + expectedLables["app.kubernetes.io/component"] = "amazon-cloudwatch-agent-target-allocator" + expectedLables["app.kubernetes.io/name"] = "my-instance-target-allocator" + + expectedData := map[string]string{ + "targetallocator.yaml": `allocation_fallback_strategy: consistent-hashing +allocation_strategy: per-node +config: + scrape_configs: + - job_name: otel-collector + scrape_interval: 10s + static_configs: + - targets: + - 0.0.0.0:8888 + - 0.0.0.0:9999 +label_selector: + app.kubernetes.io/component: amazon-cloudwatch-agent + app.kubernetes.io/instance: default.my-instance + app.kubernetes.io/managed-by: amazon-cloudwatch-agent-operator + app.kubernetes.io/part-of: amazon-cloudwatch-agent +`, + } + instance := collectorInstance() + instance.Spec.TargetAllocator.AllocationStrategy = "per-node" + cfg := config.New() + params := manifests.Params{ + OtelCol: instance, + Config: cfg, + Log: logr.Discard(), + } + actual, err := ConfigMap(params) + assert.NoError(t, err) + + assert.Equal(t, "my-instance-target-allocator", actual.Name) + assert.Equal(t, expectedLables, actual.Labels) + assert.Equal(t, expectedData, actual.Data) + }) } From a10af8b4d99a04e826b6809c2b2c050cab8b76db Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Wed, 15 Jul 2026 15:06:44 +0100 Subject: [PATCH 05/13] test(target-allocator): reach full coverage of per-node allocation logic Cover the remaining branches of the per-node strategy and its supporting code: filter application, unsupported/seeded/empty fallback, target and collector removal, unassigned and reassignment paths, collector-less bootstrapping, the already-tracked guard, GetNodeName label resolution, and the collector watcher skipping unscheduled pods then picking up the node on Modified. --- .../allocation/per_node_test.go | 235 ++++++++++++++++++ .../allocation/strategy_test.go | 20 ++ .../collector/collector_test.go | 58 +++++ .../target/target_test.go | 72 ++++++ 4 files changed, 385 insertions(+) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/target/target_test.go diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go index 67fa3e2d8..8f2be2a39 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/diff" "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/target" ) @@ -159,3 +160,237 @@ func TestPerNodeReallocatesWhenCollectorAppears(t *testing.T) { require.Len(t, bTargets, 1) assert.Equal(t, "collector-b", bTargets[0].CollectorName) } + + +// jobFilter keeps only targets whose job name matches keep. +type jobFilter struct{ keep string } + +func (f jobFilter) Apply(in map[string]*target.Item) map[string]*target.Item { + out := map[string]*target.Item{} + for k, v := range in { + if v.JobName == f.keep { + out[k] = v + } + } + return out +} + +// TestPerNodeSetFilter verifies a configured filter is applied to incoming +// targets before allocation. +func TestPerNodeSetFilter(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetFilter(jobFilter{keep: "keep"}) + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + + keep := nodeTarget("keep", "10.0.0.1:8080", "node-a") + drop := nodeTarget("drop", "10.0.0.2:8080", "node-a") + c.SetTargets(map[string]*target.Item{keep.Hash(): keep, drop.Hash(): drop}) + + require.Len(t, c.TargetItems(), 1, "filtered-out target must not be tracked") + assigned := c.GetTargetsForCollectorAndJob("collector-a", "keep") + require.Len(t, assigned, 1) + assert.Equal(t, keep.Hash(), assigned[0].Hash()) +} + +// TestPerNodeUnsupportedFallbackDisabled verifies that requesting an unsupported +// fallback strategy leaves the fallback disabled, so node-less targets stay +// unassigned. +func TestPerNodeUnsupportedFallbackDisabled(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetFallbackStrategy("bogus-strategy") + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + + noNode := nodeTarget("job1", "10.0.0.9:8080", "") + c.SetTargets(map[string]*target.Item{noNode.Hash(): noNode}) + + require.Len(t, c.TargetItems(), 1) + for _, ti := range c.TargetItems() { + assert.Equal(t, "", ti.CollectorName, "node-less target must stay unassigned when fallback is unsupported") + } +} + +// TestPerNodeFallbackSeedsExistingCollectors verifies enabling the fallback +// after collectors already exist still places node-less targets (the ring is +// seeded with the known collectors). +func TestPerNodeFallbackSeedsExistingCollectors(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + // Enable fallback AFTER collectors are known, exercising the seed loop. + c.SetFallbackStrategy(consistentHashingStrategyName) + + noNode := nodeTarget("job1", "10.0.0.9:8080", "") + c.SetTargets(map[string]*target.Item{noNode.Hash(): noNode}) + + for _, ti := range c.TargetItems() { + assert.Equal(t, "collector-a", ti.CollectorName, "node-less target must be placed by the seeded fallback ring") + } +} + +// TestPerNodeTargetRemoval verifies removed targets are dropped from the pool +// and decremented from their owning collector. +func TestPerNodeTargetRemoval(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + + t1 := nodeTarget("job1", "10.0.0.1:8080", "node-a") + t2 := nodeTarget("job1", "10.0.0.2:8080", "node-a") + c.SetTargets(map[string]*target.Item{t1.Hash(): t1, t2.Hash(): t2}) + require.Len(t, c.GetTargetsForCollectorAndJob("collector-a", "job1"), 2) + + // Remove t2. + c.SetTargets(map[string]*target.Item{t1.Hash(): t1}) + assert.Len(t, c.TargetItems(), 1) + assert.Len(t, c.GetTargetsForCollectorAndJob("collector-a", "job1"), 1) + assert.Equal(t, 1, c.Collectors()["collector-a"].NumTargets) +} + +// TestPerNodeCollectorRemovalReassigns verifies that when a collector is +// removed, its node's targets are re-evaluated (and fall back / unassign), and +// that adding another collector re-allocates already-assigned targets +// (exercising the reassignment decrement path). +func TestPerNodeCollectorRemovalReassigns(t *testing.T) { + a, err := New(perNodeStrategyName, logger, WithFallbackStrategy(consistentHashingStrategyName)) + require.NoError(t, err) + c := a.(*perNodeAllocator) + + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + "collector-b": NewCollector("collector-b", "node-b"), + }) + ta := nodeTarget("job1", "10.0.0.1:8080", "node-a") + tb := nodeTarget("job1", "10.0.0.3:8080", "node-b") + c.SetTargets(map[string]*target.Item{ta.Hash(): ta, tb.Hash(): tb}) + require.Len(t, c.GetTargetsForCollectorAndJob("collector-a", "job1"), 1) + + // Remove collector-a: its node-a target must leave collector-a. With the + // fallback enabled it lands on a remaining collector rather than vanishing. + c.SetCollectors(map[string]*Collector{ + "collector-b": NewCollector("collector-b", "node-b"), + }) + assert.Empty(t, c.GetTargetsForCollectorAndJob("collector-a", "job1")) + // Every target is still tracked and assigned to a live collector. + for _, ti := range c.TargetItems() { + _, live := c.Collectors()[ti.CollectorName] + assert.True(t, live, "target must be owned by a live collector after removal") + } +} + +// TestPerNodeSetTargetsBeforeCollectors verifies targets discovered before any +// collector exists are tracked (unassigned) and later removable in that state. +func TestPerNodeSetTargetsBeforeCollectors(t *testing.T) { + c := newPerNodeTestAllocator() + + t1 := nodeTarget("job1", "10.0.0.1:8080", "node-a") + t2 := nodeTarget("job1", "10.0.0.2:8080", "node-a") + c.SetTargets(map[string]*target.Item{t1.Hash(): t1, t2.Hash(): t2}) + require.Len(t, c.TargetItems(), 2) + for _, ti := range c.TargetItems() { + assert.Equal(t, "", ti.CollectorName, "no collectors yet, targets must be unassigned") + } + + // Removal while still collector-less. + c.SetTargets(map[string]*target.Item{t1.Hash(): t1}) + assert.Len(t, c.TargetItems(), 1) +} + +// TestPerNodeGetTargetsUnknownJob verifies querying a known collector for a job +// it does not own returns an empty slice, not nil-panic. +func TestPerNodeGetTargetsUnknownJob(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + tg := nodeTarget("job1", "10.0.0.1:8080", "node-a") + c.SetTargets(map[string]*target.Item{tg.Hash(): tg}) + + assert.Empty(t, c.GetTargetsForCollectorAndJob("collector-a", "no-such-job")) + assert.Empty(t, c.GetTargetsForCollectorAndJob("no-such-collector", "job1")) +} + +// TestPerNodeFallbackDuringCollectorChange verifies a node-less target already +// placed by the fallback is re-evaluated through the fallback again when the +// collector set changes (exercises the fallback arm of the collector-change +// re-allocation loop). +func TestPerNodeFallbackDuringCollectorChange(t *testing.T) { + a, err := New(perNodeStrategyName, logger, WithFallbackStrategy(consistentHashingStrategyName)) + require.NoError(t, err) + c := a.(*perNodeAllocator) + + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + noNode := nodeTarget("job1", "10.0.0.9:8080", "") + c.SetTargets(map[string]*target.Item{noNode.Hash(): noNode}) + require.NotEqual(t, "", c.TargetItems()[noNode.Hash()].CollectorName) + + // Add another collector: the node-less target is re-allocated via the + // fallback during handleCollectors. + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + "collector-b": NewCollector("collector-b", "node-b"), + }) + placed := c.TargetItems()[noNode.Hash()].CollectorName + assert.NotEqual(t, "", placed, "node-less target must remain fallback-placed after a collector change") + _, live := c.Collectors()[placed] + assert.True(t, live) +} + +// TestPerNodeSetCollectorsEmpty verifies passing an empty collector set is a +// safe no-op (early return) and does not panic or assign anything. +func TestPerNodeSetCollectorsEmpty(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{}) + assert.Empty(t, c.Collectors()) +} + +// TestPerNodeUnplacedDuringCollectorChange verifies that, with no fallback, a +// target whose owning collector is removed becomes unassigned during the +// collector-change re-allocation (exercises the unplaced arm of that loop). +func TestPerNodeUnplacedDuringCollectorChange(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + "collector-b": NewCollector("collector-b", "node-b"), + }) + tb := nodeTarget("job1", "10.0.0.3:8080", "node-b") + c.SetTargets(map[string]*target.Item{tb.Hash(): tb}) + require.Len(t, c.GetTargetsForCollectorAndJob("collector-b", "job1"), 1) + + // Remove collector-b; its node-b target has nowhere to go (no fallback). + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + assert.Empty(t, c.GetTargetsForCollectorAndJob("collector-b", "job1")) + assert.Equal(t, "", c.TargetItems()[tb.Hash()].CollectorName, "target must be unassigned after its collector is removed") +} + +// TestPerNodeHandleTargetsSkipsAlreadyTracked is a white-box test for the +// defensive guard in handleTargets that skips an "addition" whose target is +// already tracked, so it is never counted or placed twice. This state is not +// reachable through SetTargets (which diffs against the current pool), so the +// guard is exercised by driving handleTargets directly. +func TestPerNodeHandleTargetsSkipsAlreadyTracked(t *testing.T) { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + + tg := nodeTarget("job1", "10.0.0.1:8080", "node-a") + // Seed the target as already tracked (and unassigned). + c.targetItems[tg.Hash()] = tg + + // A diff that presents the same target as an addition must be skipped. + changes := diff.Maps(map[string]*target.Item{}, map[string]*target.Item{tg.Hash(): tg}) + c.handleTargets(changes) + + assert.Len(t, c.TargetItems(), 1, "already-tracked target must not be added twice") + assert.Equal(t, 0, c.Collectors()["collector-a"].NumTargets, "guarded target must not be (re)assigned") +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go index feb9658ab..a0136ace3 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy_test.go @@ -123,3 +123,23 @@ func TestCollectorDiff(t *testing.T) { }) } } + + +// TestWithFallbackStrategy covers the option's no-op branches: an empty name, +// and an allocator that does not support a fallback strategy. +func TestWithFallbackStrategy(t *testing.T) { + // Empty name is a no-op: the per-node fallback stays disabled. + a, err := New(perNodeStrategyName, logger, WithFallbackStrategy("")) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if pn := a.(*perNodeAllocator); pn.fallbackHasher != nil { + t.Error("empty fallback name must leave the fallback disabled") + } + + // An allocator that does not implement fallbackStrategySetter (consistent + // hashing) silently ignores the option. + if _, err := New(consistentHashingStrategyName, logger, WithFallbackStrategy(consistentHashingStrategyName)); err != nil { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go index 3b19351a3..bd4a6a916 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go @@ -214,3 +214,61 @@ func Test_closeChannel(t *testing.T) { }) } } + + +// Test_runWatch_UnscheduledThenScheduled verifies an unscheduled collector pod +// (empty NodeName) is skipped when Added, then registered with its node once a +// Modified event reports the assignment. This is the DaemonSet-rollout fix: the +// per-node strategy must pick up a collector's node without a TA restart. +func Test_runWatch_UnscheduledThenScheduled(t *testing.T) { + kubeClient, watcher := getTestClient() + defer func() { + close(kubeClient.close) + watcher.Stop() + }() + + var wg sync.WaitGroup + actual := make(map[string]*allocation.Collector) + go runWatch(context.Background(), &kubeClient, watcher.ResultChan(), map[string]*allocation.Collector{}, func(colMap map[string]*allocation.Collector) { + actual = colMap + wg.Done() + }) + + // Added while unscheduled (no NodeName): must be skipped. + wg.Add(1) + p := pod("test-pod1") + p.Spec.NodeName = "" + created, err := kubeClient.k8sClient.CoreV1().Pods("test-ns").Create(context.Background(), p, metav1.CreateOptions{}) + assert.NoError(t, err) + wg.Wait() + assert.Empty(t, actual, "unscheduled pod (no NodeName) must not be registered") + + // Scheduled later: a Modified event carrying the node must register it. + wg.Add(1) + created.Spec.NodeName = "test-pod1-node" + _, err = kubeClient.k8sClient.CoreV1().Pods("test-ns").Update(context.Background(), created, metav1.UpdateOptions{}) + assert.NoError(t, err) + wg.Wait() + + assert.Equal(t, map[string]*allocation.Collector{ + "test-pod1": {Name: "test-pod1", NodeName: "test-pod1-node"}, + }, actual) +} + + +// Test_runWatch_NonPodEventRestarts verifies runWatch restarts (returns) when an +// event carries an object that is not a Pod, rather than panicking on the type +// assertion. +func Test_runWatch_NonPodEventRestarts(t *testing.T) { + kubeClient, watcher := getTestClient() + defer func() { + close(kubeClient.close) + watcher.Stop() + }() + + events := make(chan watch.Event, 1) + events <- watch.Event{Type: watch.Added, Object: &v1.ConfigMap{}} + msg := runWatch(context.Background(), &kubeClient, events, map[string]*allocation.Collector{}, + func(map[string]*allocation.Collector) {}) + assert.Equal(t, "", msg) +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/target/target_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/target/target_test.go new file mode 100644 index 000000000..7bcca564b --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/target/target_test.go @@ -0,0 +1,72 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package target + +import ( + "testing" + + "github.com/prometheus/common/model" + "github.com/stretchr/testify/assert" +) + +func TestGetNodeName(t *testing.T) { + tests := []struct { + name string + labels model.LabelSet + want string + }{ + { + name: "pod node label", + labels: model.LabelSet{nodeNameLabelPod: "node-a"}, + want: "node-a", + }, + { + name: "node label", + labels: model.LabelSet{nodeNameLabelNode: "node-b"}, + want: "node-b", + }, + { + name: "endpoint node label", + labels: model.LabelSet{nodeNameLabelEndpoint: "node-c"}, + want: "node-c", + }, + { + name: "pod label takes precedence over others", + labels: model.LabelSet{ + nodeNameLabelPod: "node-pod", + nodeNameLabelNode: "node-node", + nodeNameLabelEndpoint: "node-endpoint", + }, + want: "node-pod", + }, + { + name: "endpointslice target kind Node resolves to target name", + labels: model.LabelSet{ + endpointSliceTargetKindLabel: "Node", + endpointSliceTargetNameLabel: "node-d", + }, + want: "node-d", + }, + { + name: "endpointslice target kind Pod is not a node", + labels: model.LabelSet{ + endpointSliceTargetKindLabel: "Pod", + endpointSliceTargetNameLabel: "some-pod", + }, + want: "", + }, + { + name: "no node labels", + labels: model.LabelSet{"__meta_kubernetes_namespace": "test"}, + want: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + item := NewItem("job", "10.0.0.1:8080", tt.labels, "") + assert.Equal(t, tt.want, item.GetNodeName()) + }) + } +} From 0bc59711613c2717af9722520302a29ad46d896e Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Wed, 15 Jul 2026 15:07:47 +0100 Subject: [PATCH 06/13] fix(target-allocator): gofmt config struct and test fallback getter Re-align the Config struct fields after adding FallbackAllocationStrategy so gofmt/CI passes, and add unit coverage for GetAllocationFallbackStrategy. --- .../config/config.go | 30 +++++++++---------- .../config/config_test.go | 9 ++++++ 2 files changed, 24 insertions(+), 15 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 23a29419d..1ca143671 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -42,21 +42,21 @@ const ( ) type Config struct { - ListenAddr string `yaml:"listen_addr,omitempty"` - KubeConfigFilePath string `yaml:"kube_config_file_path,omitempty"` - ClusterConfig *rest.Config `yaml:"-"` - RootLogger logr.Logger `yaml:"-"` - ReloadConfig bool `yaml:"-"` - LabelSelector map[string]string `yaml:"label_selector,omitempty"` - PromConfig *promconfig.Config `yaml:"config"` - AllocationStrategy *string `yaml:"allocation_strategy,omitempty"` - FallbackAllocationStrategy *string `yaml:"allocation_fallback_strategy,omitempty"` - FilterStrategy *string `yaml:"filter_strategy,omitempty"` - PrometheusCR PrometheusCRConfig `yaml:"prometheus_cr,omitempty"` - PodMonitorSelector map[string]string `yaml:"pod_monitor_selector,omitempty"` - ServiceMonitorSelector map[string]string `yaml:"service_monitor_selector,omitempty"` - CollectorSelector *metav1.LabelSelector `yaml:"collector_selector,omitempty"` - HTTPS HTTPSServerConfig `yaml:"https,omitempty"` + ListenAddr string `yaml:"listen_addr,omitempty"` + KubeConfigFilePath string `yaml:"kube_config_file_path,omitempty"` + ClusterConfig *rest.Config `yaml:"-"` + RootLogger logr.Logger `yaml:"-"` + ReloadConfig bool `yaml:"-"` + LabelSelector map[string]string `yaml:"label_selector,omitempty"` + PromConfig *promconfig.Config `yaml:"config"` + AllocationStrategy *string `yaml:"allocation_strategy,omitempty"` + FallbackAllocationStrategy *string `yaml:"allocation_fallback_strategy,omitempty"` + FilterStrategy *string `yaml:"filter_strategy,omitempty"` + PrometheusCR PrometheusCRConfig `yaml:"prometheus_cr,omitempty"` + PodMonitorSelector map[string]string `yaml:"pod_monitor_selector,omitempty"` + ServiceMonitorSelector map[string]string `yaml:"service_monitor_selector,omitempty"` + CollectorSelector *metav1.LabelSelector `yaml:"collector_selector,omitempty"` + HTTPS HTTPSServerConfig `yaml:"https,omitempty"` } type PrometheusCRConfig struct { diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go index ea6332afd..6084b9a98 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go @@ -163,3 +163,12 @@ func TestValidateConfig(t *testing.T) { }) } } + +func TestGetAllocationFallbackStrategy(t *testing.T) { + // Unset: no fallback. + assert.Equal(t, "", Config{}.GetAllocationFallbackStrategy()) + + // Set: returns the configured value. + strategy := "consistent-hashing" + assert.Equal(t, strategy, Config{FallbackAllocationStrategy: &strategy}.GetAllocationFallbackStrategy()) +} From e2aaa3c831196ee538d37f5a5213a7b81bf36430 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 10:53:38 +0100 Subject: [PATCH 07/13] feat(target-allocator): warn when per-node has no fallback newPerNodeAllocator leaves fallbackHasher nil until SetFallbackStrategy runs, so a hand-written per-node config with no fallback silently keeps node-less targets in the pool but never scrapes them (the operator's configmap always emits consistent-hashing, so production is unaffected). Emit a one-time warning when a target is left unassigned because no fallback is configured, pointing at the consistent-hashing fallback. Add TestPerNodeWarnsOnceWhenNoFallback. --- .../allocation/per_node.go | 10 ++++++ .../allocation/per_node_test.go | 36 +++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go index 4d1a1b6c6..64bb9403c 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go @@ -68,6 +68,10 @@ type perNodeAllocator struct { // collectors used to place targets that cannot be matched to a node. fallbackHasher *consistent.Consistent + // warnedNoFallback ensures the "per-node has no fallback" warning is logged + // at most once (guarded by the same lock as the allocation state). + warnedNoFallback bool + log logr.Logger filter Filter @@ -177,6 +181,12 @@ func (pn *perNodeAllocator) addTargetToTargetItems(tg *target.Item) placement { return placedByFallback } } + if pn.fallbackHasher == nil && !pn.warnedNoFallback { + pn.warnedNoFallback = true + pn.log.Info("per-node: no fallback strategy configured; targets that cannot be matched to a " + + "node-local collector (e.g. targets with no node label) will be left UNASSIGNED and never " + + "scraped. Configure a \"consistent-hashing\" fallback to place them.") + } pn.log.V(1).Info("per-node: target left UNASSIGNED (no node-local collector and no usable fallback)", "target", strings.Join(tg.TargetURL, ","), "job", tg.JobName, "node", nodeName) return unplaced diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go index 8f2be2a39..c2d59a3ff 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go @@ -4,8 +4,10 @@ package allocation import ( + "strings" "testing" + "github.com/go-logr/logr/funcr" "github.com/prometheus/common/model" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -394,3 +396,37 @@ func TestPerNodeHandleTargetsSkipsAlreadyTracked(t *testing.T) { assert.Len(t, c.TargetItems(), 1, "already-tracked target must not be added twice") assert.Equal(t, 0, c.Collectors()["collector-a"].NumTargets, "guarded target must not be (re)assigned") } + + +// TestPerNodeWarnsOnceWhenNoFallback verifies that a per-node allocator with no +// fallback logs the "no fallback strategy configured" warning exactly once (not +// per target) and leaves node-less targets unassigned. +func TestPerNodeWarnsOnceWhenNoFallback(t *testing.T) { + var msgs []string + capLogger := funcr.New(func(prefix, args string) { msgs = append(msgs, args) }, funcr.Options{}) + + c := newPerNodeAllocator(capLogger).(*perNodeAllocator) + c.SetCollectors(map[string]*Collector{ + "collector-a": NewCollector("collector-a", "node-a"), + }) + + // Two node-less targets: with no fallback both stay unassigned, and the + // warning must fire exactly once. + n1 := nodeTarget("job1", "10.0.0.9:8080", "") + n2 := nodeTarget("job1", "10.0.0.10:8080", "") + c.SetTargets(map[string]*target.Item{n1.Hash(): n1, n2.Hash(): n2}) + + for _, ti := range c.TargetItems() { + if ti.Hash() == n1.Hash() || ti.Hash() == n2.Hash() { + assert.Equal(t, "", ti.CollectorName, "node-less target must be unassigned without a fallback") + } + } + + warnings := 0 + for _, m := range msgs { + if strings.Contains(m, "no fallback strategy configured") { + warnings++ + } + } + assert.Equal(t, 1, warnings, "no-fallback warning must be logged exactly once") +} From fe394c82a9b683ef911441c6c213f6ce0180221f Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 11:00:51 +0100 Subject: [PATCH 08/13] fix(target-allocator): deterministic tie-break for per-node index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit collectorByNode was rebuilt with last-write-wins over a map, so if two collectors reported the same node (e.g. a transient maxSurge DaemonSet rollout with two pods on a node) ownership — and target placement — flapped with map iteration order. Keep the collector with the smaller pod name deterministically. Add TestPerNodeTwoCollectorsSameNodeTieBreak. --- .../allocation/per_node.go | 12 +++++++++-- .../allocation/per_node_test.go | 20 +++++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go index 64bb9403c..39b8befa8 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go @@ -256,9 +256,17 @@ func (pn *perNodeAllocator) handleCollectors(diff diff.Changes[*Collector]) { // Rebuild the node index from the current collector set. pn.collectorByNode = make(map[string]*Collector) for _, c := range pn.collectors { - if c.NodeName != "" { - pn.collectorByNode[c.NodeName] = c + if c.NodeName == "" { + continue + } + // Deterministic tie-break: normally there is one collector (DaemonSet pod) + // per node, but a maxSurge rollout can briefly place two pods on the same + // node. Keep the one with the smaller pod name so node ownership — and thus + // target placement — doesn't flap with map iteration order. + if existing, ok := pn.collectorByNode[c.NodeName]; ok && existing.Name <= c.Name { + continue } + pn.collectorByNode[c.NodeName] = c } // Log the node->collector index so it's clear which node each agent owns. diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go index c2d59a3ff..c0257e47c 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go @@ -430,3 +430,23 @@ func TestPerNodeWarnsOnceWhenNoFallback(t *testing.T) { } assert.Equal(t, 1, warnings, "no-fallback warning must be logged exactly once") } + + +// TestPerNodeTwoCollectorsSameNodeTieBreak verifies that when two collectors +// report the same node (e.g. a transient maxSurge DaemonSet rollout), the node +// is owned deterministically by the smaller pod name, not by map iteration +// order, so target placement doesn't flap. +func TestPerNodeTwoCollectorsSameNodeTieBreak(t *testing.T) { + // Run several times: SetCollectors iterates a map (random order), so a flaky + // last-write-wins would eventually pick the larger name. + for i := 0; i < 50; i++ { + c := newPerNodeTestAllocator() + c.SetCollectors(map[string]*Collector{ + "collector-b": NewCollector("collector-b", "node-a"), + "collector-a": NewCollector("collector-a", "node-a"), + }) + owner := c.collectorByNode["node-a"] + require.NotNil(t, owner) + assert.Equal(t, "collector-a", owner.Name, "smaller pod name must deterministically own the shared node") + } +} From 6c4a9c526547b242eac99a1306c350189c2e0941 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 11:01:45 +0100 Subject: [PATCH 09/13] docs(target-allocator): explain per-node empty-collectors early return Document that SetCollectors intentionally does not clear the node index/target mappings when the collector set is momentarily empty (parity with consistentHashingAllocator); clearing would drop every target during a transient zero-collector window, and state is corrected on the next non-empty call. --- .../allocation/per_node.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go index 39b8befa8..458059d8c 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go @@ -358,6 +358,11 @@ func (pn *perNodeAllocator) SetCollectors(collectors map[string]*Collector) { CollectorsAllocatable.WithLabelValues(perNodeStrategyName).Set(float64(len(collectors))) if len(collectors) == 0 { + // Intentional parity with consistentHashingAllocator: on a transient drop + // to zero collectors (e.g. a full DaemonSet restart) keep the existing node + // index and target mappings rather than clearing them. Clearing would drop + // every target during that window; the state is corrected on the next + // non-empty SetCollectors. pn.log.Info("No collector instances present") return } From 051a028742f4ce7872b4b49ea988dbbf3c735c32 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 24 Jul 2026 11:06:51 +0100 Subject: [PATCH 10/13] docs(target-allocator): explain Collector.Hash uses name only A NodeName change on the same pod would produce no collector diff (so per-node's collectorByNode would not rebuild), but that cannot happen: the collector watch skips unscheduled pods (empty NodeName) until scheduled, and a scheduled pod's spec.NodeName is immutable. Document the invariant and note to hash Name+NodeName if it ever changes. --- .../allocation/strategy.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go index 9683191ba..f2e2c7a3a 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go @@ -129,6 +129,13 @@ type Collector struct { NumTargets int } +// Hash identifies a Collector by name only. This is safe for per-node's +// collectorByNode index even though a NodeName change would produce no diff: +// unscheduled pods (empty NodeName) are skipped by the collector watch until they +// are scheduled, and a scheduled pod's spec.NodeName is immutable — so a Modified +// event on the same pod never changes NodeName. (See collector.runWatch and +// Test_runWatch_UnscheduledThenScheduled.) If that invariant ever changes, hash +// Name+NodeName here so the node index rebuilds on a node change. func (c Collector) Hash() string { return c.Name } From b53d662df42a40370dad4fb7d94f431d3de8a333 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Tue, 1 Sep 2026 20:54:22 +0100 Subject: [PATCH 11/13] fix(target-allocator): release terminating collector pods The initial pod List skips pods with a DeletionTimestamp, but the watch's Added/Modified branch did not, so a terminating agent kept ownership of its node until the Deleted event landed. Under per-node that leaves the node's targets pinned to a collector that is shutting down, so they go unscraped for the rest of the grace period instead of being re-placed on the node's replacement agent or via the fallback. Apply the same DeletionTimestamp check on Added/Modified and drop the collector as soon as the pod is marked for deletion. Add Test_runWatch_TerminatingPodReleased. --- .../collector/collector.go | 21 +++++++--- .../collector/collector_test.go | 39 ++++++++++++++++++- 2 files changed, 53 insertions(+), 7 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go index e9a3d6e10..261b13c71 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go @@ -121,12 +121,21 @@ func runWatch(ctx context.Context, k *Client, c <-chan watch.Event, collectorMap switch event.Type { //nolint:exhaustive case watch.Added, watch.Modified: - // Register/refresh the collector only once its pod is scheduled - // (NodeName set). Handling Modified captures the node when a pod that - // was Added while still unscheduled is later assigned to a node, so the - // per-node strategy can match this collector's node without needing a - // Target Allocator restart after agent (re)scheduling. - if pod.Spec.NodeName != "" { + // Drop a pod as soon as it is marked for deletion, matching the + // DeletionTimestamp check on the initial List. A terminating agent is + // shutting down and must not keep ownership of its node's targets until + // the Deleted event lands: releasing it immediately lets those targets be + // re-placed (on the node's replacement agent, or via the fallback) + // instead of going unscraped for the rest of the grace period. + // + // Otherwise register/refresh the collector, but only once its pod is + // scheduled (NodeName set). Handling Modified captures the node when a + // pod that was Added while still unscheduled is later assigned to a node, + // so the per-node strategy can match this collector's node without + // needing a Target Allocator restart after agent (re)scheduling. + if pod.GetObjectMeta().GetDeletionTimestamp() != nil { + delete(collectorMap, pod.Name) + } else if pod.Spec.NodeName != "" { collectorMap[pod.Name] = allocation.NewCollector(pod.Name, pod.Spec.NodeName) } case watch.Deleted: diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go index bd4a6a916..314c9e632 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" @@ -215,7 +216,6 @@ func Test_closeChannel(t *testing.T) { } } - // Test_runWatch_UnscheduledThenScheduled verifies an unscheduled collector pod // (empty NodeName) is skipped when Added, then registered with its node once a // Modified event reports the assignment. This is the DaemonSet-rollout fix: the @@ -255,6 +255,43 @@ func Test_runWatch_UnscheduledThenScheduled(t *testing.T) { }, actual) } +// Test_runWatch_TerminatingPodReleased verifies a collector pod that is marked +// for deletion is dropped on the Modified event carrying its DeletionTimestamp, +// rather than keeping ownership of its node's targets until the Deleted event +// lands. This matches the DeletionTimestamp check on the initial List. +func Test_runWatch_TerminatingPodReleased(t *testing.T) { + kubeClient, watcher := getTestClient() + defer func() { + close(kubeClient.close) + watcher.Stop() + }() + + p := pod("test-pod1") + terminating := p.DeepCopy() + now := metav1.Now() + terminating.DeletionTimestamp = &now + + events := make(chan watch.Event, 2) + events <- watch.Event{Type: watch.Added, Object: p} + events <- watch.Event{Type: watch.Modified, Object: terminating} + close(events) + + var updates []map[string]*allocation.Collector + runWatch(context.Background(), &kubeClient, events, map[string]*allocation.Collector{}, + func(colMap map[string]*allocation.Collector) { + snapshot := make(map[string]*allocation.Collector, len(colMap)) + for k, v := range colMap { + snapshot[k] = v + } + updates = append(updates, snapshot) + }) + + require.Len(t, updates, 2) + assert.Equal(t, map[string]*allocation.Collector{ + "test-pod1": {Name: "test-pod1", NodeName: "test-pod1-node"}, + }, updates[0], "scheduled pod must be registered on Added") + assert.Empty(t, updates[1], "pod marked for deletion must be released, not held until Deleted") +} // Test_runWatch_NonPodEventRestarts verifies runWatch restarts (returns) when an // event carries an object that is not a Pod, rather than panicking on the type From 83e96c741e8364ac32cf12c0caf13b4757ee9acb Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Tue, 1 Sep 2026 20:54:22 +0100 Subject: [PATCH 12/13] fix(target-allocator): default the per-node fallback strategy A hand-written per-node config with no allocation_fallback_strategy left targets that carry no node label retained but assigned to no collector, so they were never scraped behind a one-time log line. The operator-generated configmap always emits consistent-hashing, so only hand-written configs were exposed, but silently-unscraped targets is not a safe default. Default the fallback to consistent-hashing in Config.GetAllocationFallbackStrategy whenever the per-node strategy is selected and no fallback is set, following the existing defaulting pattern in that file (GetAllocationStrategy). Setting allocation_fallback_strategy to an empty string remains an explicit opt-out, which is now the only way to reach the no-fallback log; reword it to point at the default and document the invariant on perNodeAllocator. Extend TestGetAllocationFallbackStrategy with the per-node default and opt-out cases. --- .../allocation/per_node.go | 8 +++-- .../config/config.go | 19 +++++++++- .../config/config_test.go | 36 ++++++++++++------- 3 files changed, 47 insertions(+), 16 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go index 458059d8c..a2563c686 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go @@ -46,7 +46,11 @@ const ( // is configured (SetFallbackStrategy("consistent-hashing")); otherwise they are // retained in targetItems but assigned to no collector, re-evaluated on every // collector change, and surfaced via the -// cloudwatch_agent_allocator_targets_unassigned gauge. +// cloudwatch_agent_allocator_targets_unassigned gauge. The Target Allocator +// config defaults the fallback to consistent-hashing whenever the per-node +// strategy is selected (see config.Config.GetAllocationFallbackStrategy), so +// running per-node with no fallback — and therefore with targets that are never +// scraped — requires explicitly setting allocation_fallback_strategy to "". type perNodeAllocator struct { // m protects collectors, targetItems, targetItemsPerJobPerCollector, // collectorByNode and fallbackHasher for concurrent use. @@ -185,7 +189,7 @@ func (pn *perNodeAllocator) addTargetToTargetItems(tg *target.Item) placement { pn.warnedNoFallback = true pn.log.Info("per-node: no fallback strategy configured; targets that cannot be matched to a " + "node-local collector (e.g. targets with no node label) will be left UNASSIGNED and never " + - "scraped. Configure a \"consistent-hashing\" fallback to place them.") + "scraped. Remove allocation_fallback_strategy to get the \"consistent-hashing\" default.") } pn.log.V(1).Info("per-node: target left UNASSIGNED (no node-local collector and no usable fallback)", "target", strings.Join(tg.TargetURL, ","), "job", tg.JobName, "node", nodeName) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 1ca143671..33b68ef75 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -39,6 +39,14 @@ const ( DefaultTLSKeyPath = DefaultCertMountPath + "/server.key" DefaultTLSCertPath = DefaultCertMountPath + "/server.crt" DefaultCABundlePath = DefaultClientCertMountPath + "/tls-ca.crt" + + // PerNodeAllocationStrategy mirrors the allocation package's per-node + // strategy name, which is unexported there. + PerNodeAllocationStrategy = "per-node" + // DefaultPerNodeFallbackStrategy is the fallback strategy applied when the + // per-node strategy is configured without an explicit + // allocation_fallback_strategy, so node-less targets are still scraped. + DefaultPerNodeFallbackStrategy = DefaultAllocationStrategy ) type Config struct { @@ -81,11 +89,20 @@ func (c Config) GetAllocationStrategy() string { // GetAllocationFallbackStrategy returns the strategy used to place targets that // the primary strategy cannot assign (e.g. per-node targets with no node match). -// Empty means no fallback (such targets are left unassigned). +// +// When the per-node strategy is in use and no fallback is configured, it defaults +// to consistent-hashing: without a fallback, targets that carry no node label are +// retained but never scraped, which is not a safe default for a hand-written +// config (the operator-generated config always sets the fallback explicitly). +// Set allocation_fallback_strategy to an empty string to opt out and leave such +// targets unassigned. Empty means no fallback. func (c Config) GetAllocationFallbackStrategy() string { if c.FallbackAllocationStrategy != nil { return *c.FallbackAllocationStrategy } + if c.GetAllocationStrategy() == PerNodeAllocationStrategy { + return DefaultPerNodeFallbackStrategy + } return "" } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go index 6084b9a98..64cc2d69c 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go @@ -19,16 +19,16 @@ func TestLoad(t *testing.T) { file string } tests := []struct { - name string - args args - wantErr assert.ErrorAssertionFunc - wantHTTPS HTTPSServerConfig - wantLabels map[string]string - wantPromCR PrometheusCRConfig - wantAlloc *string - wantPodMonSel map[string]string - wantSvcMonSel map[string]string - wantJobNames []string + name string + args args + wantErr assert.ErrorAssertionFunc + wantHTTPS HTTPSServerConfig + wantLabels map[string]string + wantPromCR PrometheusCRConfig + wantAlloc *string + wantPodMonSel map[string]string + wantSvcMonSel map[string]string + wantJobNames []string }{ { name: "file sd load", @@ -58,8 +58,8 @@ func TestLoad(t *testing.T) { args: args{ file: "./testdata/no_config.yaml", }, - wantErr: assert.NoError, - wantHTTPS: CreateDefaultConfig().HTTPS, + wantErr: assert.NoError, + wantHTTPS: CreateDefaultConfig().HTTPS, wantLabels: nil, wantPromCR: CreateDefaultConfig().PrometheusCR, wantAlloc: CreateDefaultConfig().AllocationStrategy, @@ -165,10 +165,20 @@ func TestValidateConfig(t *testing.T) { } func TestGetAllocationFallbackStrategy(t *testing.T) { - // Unset: no fallback. + // Unset with the default (consistent-hashing) strategy: no fallback. assert.Equal(t, "", Config{}.GetAllocationFallbackStrategy()) // Set: returns the configured value. strategy := "consistent-hashing" assert.Equal(t, strategy, Config{FallbackAllocationStrategy: &strategy}.GetAllocationFallbackStrategy()) + + // Unset with the per-node strategy: defaults to consistent-hashing so + // node-less targets are not silently left unscraped. + perNode := PerNodeAllocationStrategy + assert.Equal(t, DefaultPerNodeFallbackStrategy, + Config{AllocationStrategy: &perNode}.GetAllocationFallbackStrategy()) + + // Explicitly empty with per-node: an opt-out, fallback stays disabled. + none := "" + assert.Equal(t, "", Config{AllocationStrategy: &perNode, FallbackAllocationStrategy: &none}.GetAllocationFallbackStrategy()) } From 56881c03fe41c0449eee2a111d532bc9f1fab5ca Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Tue, 1 Sep 2026 20:54:22 +0100 Subject: [PATCH 13/13] docs(target-allocator): sync allocationStrategy doc with the CRD The generated CRD lists both consistent-hashing and per-node, but the Go doc comment on TargetAllocatorSpec.AllocationStrategy still said the current option is consistent-hashing, so the next make manifests would have reverted the CRD description. Update the comment to match; make manifests now produces no diff for this field. --- apis/v1alpha1/amazoncloudwatchagent_types.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apis/v1alpha1/amazoncloudwatchagent_types.go b/apis/v1alpha1/amazoncloudwatchagent_types.go index a9b7179e4..e4fa75d31 100644 --- a/apis/v1alpha1/amazoncloudwatchagent_types.go +++ b/apis/v1alpha1/amazoncloudwatchagent_types.go @@ -301,7 +301,7 @@ type AmazonCloudWatchAgentTargetAllocator struct { // +optional Resources v1.ResourceRequirements `json:"resources,omitempty"` // AllocationStrategy determines which strategy the target allocator should use for allocation. - // The current option is consistent-hashing. + // The options are consistent-hashing and per-node. // +optional AllocationStrategy AmazonCloudWatchAgentTargetAllocatorAllocationStrategy `json:"allocationStrategy,omitempty"` // FilterStrategy determines how to filter targets before allocating them among the collectors.