From 13764510b87ac3a5894ed534c8d28d5bd982dc6b Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 09:21:01 +0000 Subject: [PATCH 1/4] Register enable-prometheus-cr-watcher flag and fix PrometheusCR watcher startup The target-allocator declared the enable-prometheus-cr-watcher flag name as a constant but never registered it on the flag set, while the operator passes --enable-prometheus-cr-watcher whenever PrometheusCR.enabled is true. Because args are parsed with pflag.ExitOnError, the unregistered flag caused the binary to print 'unknown flag' and exit(2), putting the target-allocator pod into CrashLoopBackOff. This change registers the flag and ORs it with the YAML prometheus_cr.enabled setting, then fixes three latent defects that were previously unreachable because the binary crashed first: - promOperator: set a non-empty Namespace on the synthetic Prometheus object so the prometheus-operator config generator no longer panics with 'namespace can't be empty' in store.ForNamespace. - promOperator: set EvaluationInterval so the generated config does not render an empty global.evaluation_interval, which the prometheus config parser rejects with 'empty duration string'. - main: create and register service-discovery metrics and pass them to discovery.NewManager; passing a nil sdMetrics map makes every SD provider fail to register, yielding zero discovered targets. RELEASE_NOTES updated. --- RELEASE_NOTES | 6 ++++++ .../config/config.go | 9 +++++++++ .../config/flags.go | 5 +++++ .../config/flags_test.go | 13 ++++++++++++ .../main.go | 13 +++++++++++- .../watcher/promOperator.go | 20 +++++++++++++++++++ 6 files changed, 65 insertions(+), 1 deletion(-) diff --git a/RELEASE_NOTES b/RELEASE_NOTES index 6128a2210..ff73e4d26 100644 --- a/RELEASE_NOTES +++ b/RELEASE_NOTES @@ -1,3 +1,9 @@ +======================================================================== +Amazon CloudWatch Agent Operator (Unreleased) +======================================================================== +Bug Fixes: +* [TargetAllocator] Register the `--enable-prometheus-cr-watcher` CLI flag so the target-allocator no longer exits with "unknown flag" (CrashLoopBackOff) when the operator passes it for PrometheusCR.enabled. The flag is OR'd with the YAML `prometheus_cr.enabled` setting. + ======================================================================== Amazon CloudWatch Agent Operator v3.5.0 (2026-06-05) ======================================================================== diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 2272eedbe..6b2a3d81d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -119,6 +119,15 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } + // Enable the Prometheus CR watcher when requested via the CLI flag. The YAML + // `prometheus_cr.enabled` value is loaded before this point, so OR the flag in + // rather than overwriting it: the watcher is enabled if either source sets it. + prometheusCREnabled, err := getPrometheusCREnabled(flagSet) + if err != nil { + return err + } + target.PrometheusCR.Enabled = target.PrometheusCR.Enabled || prometheusCREnabled + target.HTTPS.Enabled, err = getHttpsEnabled(flagSet) if err != nil { return err diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go index caf639ed9..294f9bfe6 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go @@ -34,6 +34,7 @@ func getFlagSet(errorHandling pflag.ErrorHandling) *pflag.FlagSet { flagSet := pflag.NewFlagSet(targetAllocatorName, errorHandling) flagSet.String(configFilePathFlagName, DefaultConfigFilePath, "The path to the config file.") flagSet.String(kubeConfigPathFlagName, filepath.Join(homedir.HomeDir(), ".kube", "config"), "absolute path to the KubeconfigPath file") + flagSet.Bool(prometheusCREnabledFlagName, false, "Enable watching of Prometheus Operator custom resources (ServiceMonitor/PodMonitor) to dynamically generate scrape configs.") flagSet.Bool(reloadConfigFlagName, false, "Enable automatic configuration reloading. This functionality is deprecated and will be removed in a future release.") flagSet.Bool(httpsEnabledFlagName, true, "Enable HTTPS additional server") flagSet.String(listenAddrHttpsFlagName, ":8443", "The address where this service serves over HTTPS.") @@ -58,6 +59,10 @@ func getConfigReloadEnabled(flagSet *pflag.FlagSet) (bool, error) { return flagSet.GetBool(reloadConfigFlagName) } +func getPrometheusCREnabled(flagSet *pflag.FlagSet) (bool, error) { + return flagSet.GetBool(prometheusCREnabledFlagName) +} + func getHttpsListenAddr(flagSet *pflag.FlagSet) (string, error) { return flagSet.GetString(listenAddrHttpsFlagName) } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go index 8beed29d0..66c71a268 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go @@ -17,6 +17,7 @@ func TestGetFlagSet(t *testing.T) { // Check if each flag exists assert.NotNil(t, fs.Lookup(configFilePathFlagName), "Flag %s not found", configFilePathFlagName) assert.NotNil(t, fs.Lookup(kubeConfigPathFlagName), "Flag %s not found", kubeConfigPathFlagName) + assert.NotNil(t, fs.Lookup(prometheusCREnabledFlagName), "Flag %s not found", prometheusCREnabledFlagName) } func TestFlagGetters(t *testing.T) { @@ -45,6 +46,18 @@ func TestFlagGetters(t *testing.T) { expectedValue: true, getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getConfigReloadEnabled(fs) }, }, + { + name: "GetPrometheusCREnabled", + flagArgs: []string{"--" + prometheusCREnabledFlagName}, + expectedValue: true, + getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) }, + }, + { + name: "GetPrometheusCREnabledDefault", + flagArgs: []string{}, + expectedValue: false, + getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) }, + }, { name: "InvalidFlag", flagArgs: []string{"--invalid-flag", "value"}, diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/main.go b/cmd/amazon-cloudwatch-agent-target-allocator/main.go index dbc956ebf..57185184d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/main.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/main.go @@ -88,7 +88,18 @@ func main() { srv := server.NewServer(log, allocator, cfg.ListenAddr, httpOptions...) discoveryCtx, discoveryCancel := context.WithCancel(ctx) - discoveryManager = discovery.NewManager(discoveryCtx, nil, prometheus.NewRegistry(), nil) + // Service Discovery metrics MUST be created and passed to the discovery + // manager. Each SD provider's NewDiscoverer fails when its DiscovererMetrics + // is nil, so passing a nil sdMetrics map makes every provider (including + // kubernetes_sd) fail to register, yielding zero discovered targets. Mirror + // the upstream opentelemetry-operator target-allocator. + sdRegistry := prometheus.NewRegistry() + sdMetrics, sdMetricsErr := discovery.CreateAndRegisterSDMetrics(sdRegistry) + if sdMetricsErr != nil { + setupLog.Error(sdMetricsErr, "Unable to register service discovery metrics") + os.Exit(1) + } + discoveryManager = discovery.NewManager(discoveryCtx, nil, sdRegistry, sdMetrics) targetDiscoverer = target.NewDiscoverer(log, discoveryManager, allocatorPrehook, srv) collectorWatcher, collectorWatcherErr := collector.NewClient(log, cfg.ClusterConfig) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 2651e8d29..a15db9d78 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -21,6 +21,7 @@ import ( kubeDiscovery "github.com/prometheus/prometheus/discovery/kubernetes" "gopkg.in/yaml.v2" v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" @@ -49,11 +50,30 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable + // The synthetic Prometheus object must carry a non-empty namespace: the + // prometheus-operator config generator calls store.ForNamespace(prom.Namespace) + // when generating the server configuration, which panics on an empty namespace + // ("namespace can't be empty"). Mirror the upstream opentelemetry-operator + // target-allocator by setting it to the collector/TA namespace, derived from the + // operator-injected OTELCOL_NAMESPACE downward-API env (fallback: amazon-cloudwatch). + collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") + if collectorNamespace == "" { + collectorNamespace = "amazon-cloudwatch" + } prom := &monitoringv1.Prometheus{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: collectorNamespace, + }, Spec: monitoringv1.PrometheusSpec{ CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, + // EvaluationInterval must be non-empty: the prometheus-operator config + // generator renders it verbatim as global.evaluation_interval, and the + // downstream prometheus config parser rejects an empty value with + // "empty duration string". The CWA TA does not evaluate rules, so mirror + // the scrape interval as a sane non-empty default. + EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } From 0405f4decb019dd2ee173333a3985ea517b4ffb5 Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 15:57:02 +0000 Subject: [PATCH 2/4] test(target-allocator): guard scrape_protocols defaulting on config load Add a regression test asserting that loading a Target Allocator config whose static scrape job omits scrape_protocols still yields a non-empty ScrapeProtocols on every loaded scrape config. This is defaulted by the pinned Prometheus library during yaml.UnmarshalStrict into the prometheus Config type, so the distributed /scrape_configs payload is never empty and the agent's prometheus-receiver validation passes. The test fails fast if a future dependency or load-path change drops this defaulting. --- .../scrape_protocols_regression_test.go | 36 +++++++++++++++++++ .../scrape_protocols_omitted_test.yaml | 14 ++++++++ 2 files changed, 50 insertions(+) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go new file mode 100644 index 000000000..9c432d7ee --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go @@ -0,0 +1,36 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestScrapeProtocolsDefaultedOnLoad guards against "scrape_protocols cannot be empty" regression. +func TestScrapeProtocolsDefaultedOnLoad(t *testing.T) { + got := CreateDefaultConfig() + err := LoadFromFile("./testdata/scrape_protocols_omitted_test.yaml", &got) + require.NoError(t, err) + + require.NotNil(t, got.PromConfig) + require.NotEmpty(t, got.PromConfig.ScrapeConfigs) + + for _, sc := range got.PromConfig.ScrapeConfigs { + assert.NotEmpty(t, sc.ScrapeProtocols, + "scrape_protocols must be defaulted for job %q", sc.JobName) + } + + // Spot-check the specific static job from the reproduction by name. + found := false + for _, sc := range got.PromConfig.ScrapeConfigs { + if sc.JobName == "prometheus-sample-app" { + found = true + assert.Greater(t, len(sc.ScrapeProtocols), 0) + } + } + require.True(t, found, "expected the 'prometheus-sample-app' job to be present") +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml new file mode 100644 index 000000000..9405f61e8 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml @@ -0,0 +1,14 @@ +# Regression fixture: a scrape job that omits scrape_protocols. +# The TA config-load path must default scrape_protocols so the agent does not +# reject the config with "scrape_protocols cannot be empty". +config: + scrape_configs: + - job_name: prometheus-sample-app + kubernetes_sd_configs: + - role: pod + relabel_configs: + - source_labels: [__meta_kubernetes_pod_label_app] + regex: prometheus-sample-app + action: keep + - target_label: label1 + replacement: value1 From d61d693fa6f0536786951e8e4d08f378a5e77c82 Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 16:01:54 +0000 Subject: [PATCH 3/4] fix(collector): roll pods when Prometheus config changes The pod-template restart-trigger sha256 was computed from Spec.Config only, so a change to Spec.Prometheus (rendered into a separate ConfigMap) left the pod template byte-identical and the workload controller did not roll the pods. Fold the serialized Spec.Prometheus (PrometheusConfig.Yaml()) into the hash input when it is non-empty, so a Prometheus-only change bumps the pod-template annotation and triggers a rolling restart, matching agent-config behavior. When no Prometheus config is set the hash input is byte-identical to the agent config alone, leaving non-Prometheus agents unaffected. --- RELEASE_NOTES | 6 -- .../config/config.go | 4 +- .../main.go | 6 +- .../watcher/promOperator.go | 22 +++---- internal/manifests/collector/annotations.go | 21 ++++++- .../manifests/collector/annotations_test.go | 61 +++++++++++++++++++ 6 files changed, 92 insertions(+), 28 deletions(-) diff --git a/RELEASE_NOTES b/RELEASE_NOTES index ff73e4d26..6128a2210 100644 --- a/RELEASE_NOTES +++ b/RELEASE_NOTES @@ -1,9 +1,3 @@ -======================================================================== -Amazon CloudWatch Agent Operator (Unreleased) -======================================================================== -Bug Fixes: -* [TargetAllocator] Register the `--enable-prometheus-cr-watcher` CLI flag so the target-allocator no longer exits with "unknown flag" (CrashLoopBackOff) when the operator passes it for PrometheusCR.enabled. The flag is OR'd with the YAML `prometheus_cr.enabled` setting. - ======================================================================== Amazon CloudWatch Agent Operator v3.5.0 (2026-06-05) ======================================================================== diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 6b2a3d81d..081e646e9 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -119,9 +119,7 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } - // Enable the Prometheus CR watcher when requested via the CLI flag. The YAML - // `prometheus_cr.enabled` value is loaded before this point, so OR the flag in - // rather than overwriting it: the watcher is enabled if either source sets it. + // OR the CLI flag into the YAML value so either source can enable the watcher. prometheusCREnabled, err := getPrometheusCREnabled(flagSet) if err != nil { return err diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/main.go b/cmd/amazon-cloudwatch-agent-target-allocator/main.go index 57185184d..2b553dce6 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/main.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/main.go @@ -88,11 +88,7 @@ func main() { srv := server.NewServer(log, allocator, cfg.ListenAddr, httpOptions...) discoveryCtx, discoveryCancel := context.WithCancel(ctx) - // Service Discovery metrics MUST be created and passed to the discovery - // manager. Each SD provider's NewDiscoverer fails when its DiscovererMetrics - // is nil, so passing a nil sdMetrics map makes every provider (including - // kubernetes_sd) fail to register, yielding zero discovered targets. Mirror - // the upstream opentelemetry-operator target-allocator. + // SD metrics must be non-nil; providers fail to register without them. sdRegistry := prometheus.NewRegistry() sdMetrics, sdMetricsErr := discovery.CreateAndRegisterSDMetrics(sdRegistry) if sdMetricsErr != nil { diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index a15db9d78..5dfc314dd 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -29,6 +29,8 @@ import ( allocatorconfig "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/config" ) +const defaultCollectorNamespace = "amazon-cloudwatch" + const minEventInterval = time.Second * 5 func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*PrometheusCRWatcher, error) { @@ -50,15 +52,15 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable - // The synthetic Prometheus object must carry a non-empty namespace: the - // prometheus-operator config generator calls store.ForNamespace(prom.Namespace) - // when generating the server configuration, which panics on an empty namespace - // ("namespace can't be empty"). Mirror the upstream opentelemetry-operator - // target-allocator by setting it to the collector/TA namespace, derived from the - // operator-injected OTELCOL_NAMESPACE downward-API env (fallback: amazon-cloudwatch). + // Namespace must be non-empty; the config generator panics otherwise. collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") if collectorNamespace == "" { - collectorNamespace = "amazon-cloudwatch" + if ns, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil && len(ns) > 0 { + collectorNamespace = string(ns) + } else { + collectorNamespace = defaultCollectorNamespace + } + logger.Info("OTELCOL_NAMESPACE not set, resolved namespace", "namespace", collectorNamespace) } prom := &monitoringv1.Prometheus{ ObjectMeta: metav1.ObjectMeta{ @@ -68,11 +70,7 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, - // EvaluationInterval must be non-empty: the prometheus-operator config - // generator renders it verbatim as global.evaluation_interval, and the - // downstream prometheus config parser rejects an empty value with - // "empty duration string". The CWA TA does not evaluate rules, so mirror - // the scrape interval as a sane non-empty default. + // Must be non-empty; default to scrape interval. EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } diff --git a/internal/manifests/collector/annotations.go b/internal/manifests/collector/annotations.go index a665e465e..364072330 100644 --- a/internal/manifests/collector/annotations.go +++ b/internal/manifests/collector/annotations.go @@ -6,6 +6,7 @@ package collector import ( "crypto/sha256" "fmt" + "log/slog" "github.com/aws/amazon-cloudwatch-agent-operator/apis/v1alpha1" ) @@ -28,7 +29,7 @@ func Annotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string { } // make sure sha256 for configMap is always calculated - annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config) + annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance)) return annotations } @@ -51,11 +52,27 @@ func PodAnnotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string { } // make sure sha256 for configMap is always calculated - podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config) + podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance)) return podAnnotations } +// configHashInput returns the combined config string used for the pod-template restart hash. +func configHashInput(instance v1alpha1.AmazonCloudWatchAgent) string { + config := instance.Spec.Config + if !instance.Spec.Prometheus.IsEmpty() { + promYaml, err := instance.Spec.Prometheus.Yaml() + if err != nil { + // Static sentinel; Yaml() over map[string]interface{} rarely fails in practice. + slog.Warn("failed to serialize Spec.Prometheus for config hash", "error", err) + config += "\x00prometheus-serialize-error" + } else { + config += "\x00" + promYaml // null byte prevents collision between config suffix and promYAML prefix + } + } + return config +} + func getConfigMapSHA(config string) string { h := sha256.Sum256([]byte(config)) return fmt.Sprintf("%x", h) diff --git a/internal/manifests/collector/annotations_test.go b/internal/manifests/collector/annotations_test.go index e02ef1678..1ab8e3bd7 100644 --- a/internal/manifests/collector/annotations_test.go +++ b/internal/manifests/collector/annotations_test.go @@ -79,3 +79,64 @@ func TestAnnotationsPropagateDown(t *testing.T) { assert.Equal(t, "mycomponent", podAnnotations["myapp"]) assert.Equal(t, "pod_annotation_value", podAnnotations["pod_annotation"]) } + +func promConfig(t *testing.T, replacement string) v1alpha1.PrometheusConfig { + t.Helper() + cfg := map[string]interface{}{ + "scrape_configs": []interface{}{ + map[string]interface{}{ + "job_name": "kubernetes-pods-annotated", + "relabel_configs": []interface{}{ + map[string]interface{}{ + "target_label": "bug2probe", + "replacement": replacement, + }, + }, + }, + }, + } + return v1alpha1.PrometheusConfig{ + Config: &v1alpha1.AnyConfig{Object: cfg}, + } +} + +// TestPrometheusConfigChangeBumpsHash verifies a Prometheus-only change bumps the pod-template hash. +func TestPrometheusConfigChangeBumpsHash(t *testing.T) { + base := v1alpha1.AmazonCloudWatchAgent{ + ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"}, + Spec: v1alpha1.AmazonCloudWatchAgentSpec{ + Config: "agent-config", + Prometheus: promConfig(t, "value2"), + }, + } + + // same spec twice -> stable hash + h1 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + h2 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.Equal(t, h1, h2, "hash must be stable when nothing changes") + + // change ONLY the prometheus config -> hash must change + changed := base + changed.Spec.Prometheus = promConfig(t, "value3") + h3 := PodAnnotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.NotEqual(t, h1, h3, "pod annotation hash must change when only Spec.Prometheus changes") + + // metadata annotations hash must also reflect the prometheus change + a1 := Annotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + a3 := Annotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.NotEqual(t, a1, a3, "metadata annotation hash must change when only Spec.Prometheus changes") +} + +// TestEmptyPrometheusHashUnchanged verifies non-Prometheus agents keep a stable hash. +func TestEmptyPrometheusHashUnchanged(t *testing.T) { + otelcol := v1alpha1.AmazonCloudWatchAgent{ + ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"}, + Spec: v1alpha1.AmazonCloudWatchAgentSpec{Config: "test"}, + } + + // sha256("test") == 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08 + assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", + Annotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"]) + assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", + PodAnnotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"]) +} From efd17bbb60f9a76cd71fad8480b537dbf0808e02 Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Fri, 26 Jun 2026 14:21:08 +0100 Subject: [PATCH 4/4] 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) + }) }