From 5a2600a50b04fa659548808667a821175331396c Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 09:21:01 +0000 Subject: [PATCH 1/5] Register enable-prometheus-cr-watcher flag and fix PrometheusCR watcher startup The target-allocator declared the enable-prometheus-cr-watcher flag name as a constant but never registered it on the flag set, while the operator passes --enable-prometheus-cr-watcher whenever PrometheusCR.enabled is true. Because args are parsed with pflag.ExitOnError, the unregistered flag caused the binary to print 'unknown flag' and exit(2), putting the target-allocator pod into CrashLoopBackOff. This change registers the flag and ORs it with the YAML prometheus_cr.enabled setting, then fixes three latent defects that were previously unreachable because the binary crashed first: - promOperator: set a non-empty Namespace on the synthetic Prometheus object so the prometheus-operator config generator no longer panics with 'namespace can't be empty' in store.ForNamespace. - promOperator: set EvaluationInterval so the generated config does not render an empty global.evaluation_interval, which the prometheus config parser rejects with 'empty duration string'. - main: create and register service-discovery metrics and pass them to discovery.NewManager; passing a nil sdMetrics map makes every SD provider fail to register, yielding zero discovered targets. RELEASE_NOTES updated. (cherry picked from commit 13764510b87ac3a5894ed534c8d28d5bd982dc6b) --- .../config/config.go | 9 +++++++++ .../config/flags.go | 5 +++++ .../config/flags_test.go | 13 ++++++++++++ .../watcher/promOperator.go | 20 +++++++++++++++++++ 4 files changed, 47 insertions(+) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 2272eedbe..6b2a3d81d 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -119,6 +119,15 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } + // Enable the Prometheus CR watcher when requested via the CLI flag. The YAML + // `prometheus_cr.enabled` value is loaded before this point, so OR the flag in + // rather than overwriting it: the watcher is enabled if either source sets it. + prometheusCREnabled, err := getPrometheusCREnabled(flagSet) + if err != nil { + return err + } + target.PrometheusCR.Enabled = target.PrometheusCR.Enabled || prometheusCREnabled + target.HTTPS.Enabled, err = getHttpsEnabled(flagSet) if err != nil { return err diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go index caf639ed9..294f9bfe6 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags.go @@ -34,6 +34,7 @@ func getFlagSet(errorHandling pflag.ErrorHandling) *pflag.FlagSet { flagSet := pflag.NewFlagSet(targetAllocatorName, errorHandling) flagSet.String(configFilePathFlagName, DefaultConfigFilePath, "The path to the config file.") flagSet.String(kubeConfigPathFlagName, filepath.Join(homedir.HomeDir(), ".kube", "config"), "absolute path to the KubeconfigPath file") + flagSet.Bool(prometheusCREnabledFlagName, false, "Enable watching of Prometheus Operator custom resources (ServiceMonitor/PodMonitor) to dynamically generate scrape configs.") flagSet.Bool(reloadConfigFlagName, false, "Enable automatic configuration reloading. This functionality is deprecated and will be removed in a future release.") flagSet.Bool(httpsEnabledFlagName, true, "Enable HTTPS additional server") flagSet.String(listenAddrHttpsFlagName, ":8443", "The address where this service serves over HTTPS.") @@ -58,6 +59,10 @@ func getConfigReloadEnabled(flagSet *pflag.FlagSet) (bool, error) { return flagSet.GetBool(reloadConfigFlagName) } +func getPrometheusCREnabled(flagSet *pflag.FlagSet) (bool, error) { + return flagSet.GetBool(prometheusCREnabledFlagName) +} + func getHttpsListenAddr(flagSet *pflag.FlagSet) (string, error) { return flagSet.GetString(listenAddrHttpsFlagName) } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go index 8beed29d0..66c71a268 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/flags_test.go @@ -17,6 +17,7 @@ func TestGetFlagSet(t *testing.T) { // Check if each flag exists assert.NotNil(t, fs.Lookup(configFilePathFlagName), "Flag %s not found", configFilePathFlagName) assert.NotNil(t, fs.Lookup(kubeConfigPathFlagName), "Flag %s not found", kubeConfigPathFlagName) + assert.NotNil(t, fs.Lookup(prometheusCREnabledFlagName), "Flag %s not found", prometheusCREnabledFlagName) } func TestFlagGetters(t *testing.T) { @@ -45,6 +46,18 @@ func TestFlagGetters(t *testing.T) { expectedValue: true, getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getConfigReloadEnabled(fs) }, }, + { + name: "GetPrometheusCREnabled", + flagArgs: []string{"--" + prometheusCREnabledFlagName}, + expectedValue: true, + getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) }, + }, + { + name: "GetPrometheusCREnabledDefault", + flagArgs: []string{}, + expectedValue: false, + getterFunc: func(fs *pflag.FlagSet) (interface{}, error) { return getPrometheusCREnabled(fs) }, + }, { name: "InvalidFlag", flagArgs: []string{"--invalid-flag", "value"}, diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 2651e8d29..a15db9d78 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -21,6 +21,7 @@ import ( kubeDiscovery "github.com/prometheus/prometheus/discovery/kubernetes" "gopkg.in/yaml.v2" v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" @@ -49,11 +50,30 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable + // The synthetic Prometheus object must carry a non-empty namespace: the + // prometheus-operator config generator calls store.ForNamespace(prom.Namespace) + // when generating the server configuration, which panics on an empty namespace + // ("namespace can't be empty"). Mirror the upstream opentelemetry-operator + // target-allocator by setting it to the collector/TA namespace, derived from the + // operator-injected OTELCOL_NAMESPACE downward-API env (fallback: amazon-cloudwatch). + collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") + if collectorNamespace == "" { + collectorNamespace = "amazon-cloudwatch" + } prom := &monitoringv1.Prometheus{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: collectorNamespace, + }, Spec: monitoringv1.PrometheusSpec{ CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, + // EvaluationInterval must be non-empty: the prometheus-operator config + // generator renders it verbatim as global.evaluation_interval, and the + // downstream prometheus config parser rejects an empty value with + // "empty duration string". The CWA TA does not evaluate rules, so mirror + // the scrape interval as a sane non-empty default. + EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } From e8737efc3f60f8cd6acf09007d97680c1bcfb31d Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 15:57:02 +0000 Subject: [PATCH 2/5] test(target-allocator): guard scrape_protocols defaulting on config load Add a regression test asserting that loading a Target Allocator config whose static scrape job omits scrape_protocols still yields a non-empty ScrapeProtocols on every loaded scrape config. This is defaulted by the pinned Prometheus library during yaml.UnmarshalStrict into the prometheus Config type, so the distributed /scrape_configs payload is never empty and the agent's prometheus-receiver validation passes. The test fails fast if a future dependency or load-path change drops this defaulting. (cherry picked from commit 0405f4decb019dd2ee173333a3985ea517b4ffb5) --- .../scrape_protocols_regression_test.go | 36 +++++++++++++++++++ .../scrape_protocols_omitted_test.yaml | 14 ++++++++ 2 files changed, 50 insertions(+) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go new file mode 100644 index 000000000..9c432d7ee --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/scrape_protocols_regression_test.go @@ -0,0 +1,36 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestScrapeProtocolsDefaultedOnLoad guards against "scrape_protocols cannot be empty" regression. +func TestScrapeProtocolsDefaultedOnLoad(t *testing.T) { + got := CreateDefaultConfig() + err := LoadFromFile("./testdata/scrape_protocols_omitted_test.yaml", &got) + require.NoError(t, err) + + require.NotNil(t, got.PromConfig) + require.NotEmpty(t, got.PromConfig.ScrapeConfigs) + + for _, sc := range got.PromConfig.ScrapeConfigs { + assert.NotEmpty(t, sc.ScrapeProtocols, + "scrape_protocols must be defaulted for job %q", sc.JobName) + } + + // Spot-check the specific static job from the reproduction by name. + found := false + for _, sc := range got.PromConfig.ScrapeConfigs { + if sc.JobName == "prometheus-sample-app" { + found = true + assert.Greater(t, len(sc.ScrapeProtocols), 0) + } + } + require.True(t, found, "expected the 'prometheus-sample-app' job to be present") +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml new file mode 100644 index 000000000..9405f61e8 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/testdata/scrape_protocols_omitted_test.yaml @@ -0,0 +1,14 @@ +# Regression fixture: a scrape job that omits scrape_protocols. +# The TA config-load path must default scrape_protocols so the agent does not +# reject the config with "scrape_protocols cannot be empty". +config: + scrape_configs: + - job_name: prometheus-sample-app + kubernetes_sd_configs: + - role: pod + relabel_configs: + - source_labels: [__meta_kubernetes_pod_label_app] + regex: prometheus-sample-app + action: keep + - target_label: label1 + replacement: value1 From 9624a74cf4217457952d99629eee0292f1095327 Mon Sep 17 00:00:00 2001 From: Musa Asad Date: Tue, 16 Jun 2026 16:01:54 +0000 Subject: [PATCH 3/5] fix(collector): roll pods when Prometheus config changes The pod-template restart-trigger sha256 was computed from Spec.Config only, so a change to Spec.Prometheus (rendered into a separate ConfigMap) left the pod template byte-identical and the workload controller did not roll the pods. Fold the serialized Spec.Prometheus (PrometheusConfig.Yaml()) into the hash input when it is non-empty, so a Prometheus-only change bumps the pod-template annotation and triggers a rolling restart, matching agent-config behavior. When no Prometheus config is set the hash input is byte-identical to the agent config alone, leaving non-Prometheus agents unaffected. (cherry picked from commit d61d693fa6f0536786951e8e4d08f378a5e77c82) --- .../config/config.go | 4 +- .../watcher/promOperator.go | 22 +++---- internal/manifests/collector/annotations.go | 21 ++++++- .../manifests/collector/annotations_test.go | 61 +++++++++++++++++++ 4 files changed, 91 insertions(+), 17 deletions(-) diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 6b2a3d81d..081e646e9 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -119,9 +119,7 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } - // Enable the Prometheus CR watcher when requested via the CLI flag. The YAML - // `prometheus_cr.enabled` value is loaded before this point, so OR the flag in - // rather than overwriting it: the watcher is enabled if either source sets it. + // OR the CLI flag into the YAML value so either source can enable the watcher. prometheusCREnabled, err := getPrometheusCREnabled(flagSet) if err != nil { return err diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index a15db9d78..5dfc314dd 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -29,6 +29,8 @@ import ( allocatorconfig "github.com/aws/amazon-cloudwatch-agent-operator/cmd/amazon-cloudwatch-agent-target-allocator/config" ) +const defaultCollectorNamespace = "amazon-cloudwatch" + const minEventInterval = time.Second * 5 func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*PrometheusCRWatcher, error) { @@ -50,15 +52,15 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable - // The synthetic Prometheus object must carry a non-empty namespace: the - // prometheus-operator config generator calls store.ForNamespace(prom.Namespace) - // when generating the server configuration, which panics on an empty namespace - // ("namespace can't be empty"). Mirror the upstream opentelemetry-operator - // target-allocator by setting it to the collector/TA namespace, derived from the - // operator-injected OTELCOL_NAMESPACE downward-API env (fallback: amazon-cloudwatch). + // Namespace must be non-empty; the config generator panics otherwise. collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") if collectorNamespace == "" { - collectorNamespace = "amazon-cloudwatch" + if ns, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace"); err == nil && len(ns) > 0 { + collectorNamespace = string(ns) + } else { + collectorNamespace = defaultCollectorNamespace + } + logger.Info("OTELCOL_NAMESPACE not set, resolved namespace", "namespace", collectorNamespace) } prom := &monitoringv1.Prometheus{ ObjectMeta: metav1.ObjectMeta{ @@ -68,11 +70,7 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, - // EvaluationInterval must be non-empty: the prometheus-operator config - // generator renders it verbatim as global.evaluation_interval, and the - // downstream prometheus config parser rejects an empty value with - // "empty duration string". The CWA TA does not evaluate rules, so mirror - // the scrape interval as a sane non-empty default. + // Must be non-empty; default to scrape interval. EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } diff --git a/internal/manifests/collector/annotations.go b/internal/manifests/collector/annotations.go index 52cad873f..fddf89be8 100644 --- a/internal/manifests/collector/annotations.go +++ b/internal/manifests/collector/annotations.go @@ -6,6 +6,7 @@ package collector import ( "crypto/sha256" "fmt" + "log/slog" "github.com/aws/amazon-cloudwatch-agent-operator/apis/v1alpha1" ) @@ -28,7 +29,7 @@ func Annotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string { } // make sure sha256 for configMap is always calculated - annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config) + annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance)) return annotations } @@ -51,7 +52,7 @@ func PodAnnotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string { } // make sure sha256 for configMap is always calculated - podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config) + podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance)) return podAnnotations } @@ -71,6 +72,22 @@ func PodLabels(instance v1alpha1.AmazonCloudWatchAgent, base map[string]string) return merged } +// configHashInput returns the combined config string used for the pod-template restart hash. +func configHashInput(instance v1alpha1.AmazonCloudWatchAgent) string { + config := instance.Spec.Config + if !instance.Spec.Prometheus.IsEmpty() { + promYaml, err := instance.Spec.Prometheus.Yaml() + if err != nil { + // Static sentinel; Yaml() over map[string]interface{} rarely fails in practice. + slog.Warn("failed to serialize Spec.Prometheus for config hash", "error", err) + config += "\x00prometheus-serialize-error" + } else { + config += "\x00" + promYaml // null byte prevents collision between config suffix and promYAML prefix + } + } + return config +} + func getConfigMapSHA(config string) string { h := sha256.Sum256([]byte(config)) return fmt.Sprintf("%x", h) diff --git a/internal/manifests/collector/annotations_test.go b/internal/manifests/collector/annotations_test.go index e02ef1678..1ab8e3bd7 100644 --- a/internal/manifests/collector/annotations_test.go +++ b/internal/manifests/collector/annotations_test.go @@ -79,3 +79,64 @@ func TestAnnotationsPropagateDown(t *testing.T) { assert.Equal(t, "mycomponent", podAnnotations["myapp"]) assert.Equal(t, "pod_annotation_value", podAnnotations["pod_annotation"]) } + +func promConfig(t *testing.T, replacement string) v1alpha1.PrometheusConfig { + t.Helper() + cfg := map[string]interface{}{ + "scrape_configs": []interface{}{ + map[string]interface{}{ + "job_name": "kubernetes-pods-annotated", + "relabel_configs": []interface{}{ + map[string]interface{}{ + "target_label": "bug2probe", + "replacement": replacement, + }, + }, + }, + }, + } + return v1alpha1.PrometheusConfig{ + Config: &v1alpha1.AnyConfig{Object: cfg}, + } +} + +// TestPrometheusConfigChangeBumpsHash verifies a Prometheus-only change bumps the pod-template hash. +func TestPrometheusConfigChangeBumpsHash(t *testing.T) { + base := v1alpha1.AmazonCloudWatchAgent{ + ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"}, + Spec: v1alpha1.AmazonCloudWatchAgentSpec{ + Config: "agent-config", + Prometheus: promConfig(t, "value2"), + }, + } + + // same spec twice -> stable hash + h1 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + h2 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.Equal(t, h1, h2, "hash must be stable when nothing changes") + + // change ONLY the prometheus config -> hash must change + changed := base + changed.Spec.Prometheus = promConfig(t, "value3") + h3 := PodAnnotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.NotEqual(t, h1, h3, "pod annotation hash must change when only Spec.Prometheus changes") + + // metadata annotations hash must also reflect the prometheus change + a1 := Annotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + a3 := Annotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.NotEqual(t, a1, a3, "metadata annotation hash must change when only Spec.Prometheus changes") +} + +// TestEmptyPrometheusHashUnchanged verifies non-Prometheus agents keep a stable hash. +func TestEmptyPrometheusHashUnchanged(t *testing.T) { + otelcol := v1alpha1.AmazonCloudWatchAgent{ + ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"}, + Spec: v1alpha1.AmazonCloudWatchAgentSpec{Config: "test"}, + } + + // sha256("test") == 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08 + assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", + Annotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"]) + assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", + PodAnnotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"]) +} From 2e4bf12a4278cc4b93c0fa3ffe756ce7061e7cdc Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Wed, 23 Sep 2026 13:40:12 +0100 Subject: [PATCH 4/5] feat(target-allocator): add per-node allocation strategy Adds a per-node allocation strategy so each CloudWatch Agent scrapes only the ServiceMonitor/PodMonitor targets on its own node, eliminating cross-node and cross-AZ scrape traffic. Targets with no resolvable node fall back to consistent-hashing so nothing is silently dropped, and unassigned targets are surfaced via a gauge. Also changes collector registration for ALL strategies, including the existing consistent-hashing default: the initial List now skips pods with an empty spec.NodeName, watch.Modified is handled so a pod scheduled after Add is picked up, and pods are dropped as soon as they carry a DeletionTimestamp rather than on Deleted. This releases a terminating agent's targets immediately instead of at the end of its grace period, at the cost of more target churn during rollouts. Verified: go build ./... clean; make impi and make checklicense pass; unit tests pass across the target-allocator and manifests packages. (cherry picked from commit 1595a876f151425d48b4a001330bea5d26fad84f) --- apis/v1alpha1/allocation_strategy.go | 6 +- apis/v1alpha1/amazoncloudwatchagent_types.go | 2 +- .../allocation/consistent_hashing.go | 2 +- .../allocation/per_node.go | 423 +++++++++++++++++ .../allocation/per_node_test.go | 449 ++++++++++++++++++ .../allocation/strategy.go | 46 +- .../allocation/strategy_test.go | 29 +- .../collector/collector.go | 27 +- .../collector/collector_test.go | 110 ++++- .../config/config.go | 56 ++- .../config/config_test.go | 19 + .../main.go | 4 +- .../target/target.go | 38 ++ .../target/target_test.go | 72 +++ ...aws.amazon.com_amazoncloudwatchagents.yaml | 3 +- .../manifests/targetallocator/configmap.go | 13 +- .../targetallocator/configmap_test.go | 37 ++ 17 files changed, 1301 insertions(+), 35 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 create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/target/target_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/apis/v1alpha1/amazoncloudwatchagent_types.go b/apis/v1alpha1/amazoncloudwatchagent_types.go index 0ee86ea42..7605df20a 100644 --- a/apis/v1alpha1/amazoncloudwatchagent_types.go +++ b/apis/v1alpha1/amazoncloudwatchagent_types.go @@ -316,7 +316,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. 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..a2563c686 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node.go @@ -0,0 +1,423 @@ +// 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. 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. + 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 + + // 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 +} + +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 + } + } + 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. 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) + 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 == "" { + 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. + 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 { + // 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 + } + + 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..c1d9430b1 --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go @@ -0,0 +1,449 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +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" + + "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" +) + +// 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) +} + +// 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") +} + +// 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") +} + +// 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") + } +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/strategy.go index 5f5285ece..f2e2c7a3a 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,12 +120,22 @@ 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 } +// 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 } @@ -106,8 +144,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 +153,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..d86309c0c 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 @@ -123,3 +123,22 @@ 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.go b/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.go index 60bda6bca..261b13c71 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,24 @@ 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: + // 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: 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..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" @@ -58,6 +59,9 @@ func pod(name string) *v1.Pod { Namespace: "test-ns", Labels: labelSet, }, + Spec: v1.PodSpec{ + NodeName: name + "-node", + }, } } @@ -86,13 +90,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 +127,8 @@ func Test_runWatch(t *testing.T) { }, want: map[string]*allocation.Collector{ "test-pod1": { - Name: "test-pod1", + Name: "test-pod1", + NodeName: "test-pod1-node", }, }, }, @@ -207,3 +215,97 @@ 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_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 +// 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/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 081e646e9..33b68ef75 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -39,23 +39,32 @@ 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 { - 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"` - 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 { @@ -78,6 +87,25 @@ 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). +// +// 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 "" +} + func (c Config) GetTargetsFilterStrategy() string { if c.FilterStrategy != nil { return *c.FilterStrategy 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 5c6f2fec2..64cc2d69c 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,22 @@ func TestValidateConfig(t *testing.T) { }) } } + +func TestGetAllocationFallbackStrategy(t *testing.T) { + // 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()) +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/main.go b/cmd/amazon-cloudwatch-agent-target-allocator/main.go index 4271832e1..1170831b8 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/main.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/main.go @@ -73,7 +73,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..1093c1388 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go @@ -10,6 +10,25 @@ 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" +) + +// nodeNameLabels is the ordered set of discovery meta-labels consulted by +// GetNodeName. Package-level so the per-node allocator does not allocate a fresh +// slice for every target on every allocation cycle (handleCollectors re-allocates +// every known target on any collector change). +var nodeNameLabels = []model.LabelName{nodeNameLabelPod, nodeNameLabelNode, nodeNameLabelEndpoint} + // LinkJSON This package contains common structs and methods that relate to scrape targets. type LinkJSON struct { Link string `json:"_link"` @@ -28,6 +47,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 nodeNameLabels { + 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/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()) + }) + } +} diff --git a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml index 41e9f1e83..ed7e7be6c 100644 --- a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml +++ b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml @@ -6648,9 +6648,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 afa7000903dc21adaf51523ae4717b0f187918fe Mon Sep 17 00:00:00 2001 From: Praise Wenegieme Date: Wed, 23 Sep 2026 13:40:12 +0100 Subject: [PATCH 5/5] feat(target-allocator): route SM/PM by cloudwatch.aws/scraper annotation Partitions ServiceMonitor/PodMonitor discovery across CloudWatch Agents by the cloudwatch.aws/scraper annotation on the monitor CR. A monitor annotated cluster-scraper is scraped only by the cluster-scraper agent's Target Allocator; all others only by the per-node agent's. The two roles are complementary, so every monitor is owned by exactly one agent with no double-scrape and no gap. Adds spec.targetAllocator.prometheusCR.scraperRole (Target Allocator config key scraper_role) and a client-side annotation filter applied during monitor discovery in LoadConfig. Note: scraperRole is constrained by an enum with no empty member, so the default per-node role is expressed by omitting the field; setting scraperRole: "" explicitly is rejected by the API server. Companion: aws-observability/helm-charts#339. Verified: go build ./... clean; make impi and make checklicense pass; unit tests pass across the target-allocator and manifests packages. (cherry picked from commit cd3e40b63e1095dba8072017783a8db100143e03) --- apis/v1alpha1/amazoncloudwatchagent_types.go | 8 + .../config/config.go | 1 + .../watcher/promOperator.go | 57 ++++++ .../watcher/promOperator_annotation_test.go | 190 ++++++++++++++++++ ...aws.amazon.com_amazoncloudwatchagents.yaml | 10 + .../manifests/targetallocator/configmap.go | 4 + .../targetallocator/configmap_test.go | 40 ++++ 7 files changed, 310 insertions(+) create mode 100644 cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_annotation_test.go diff --git a/apis/v1alpha1/amazoncloudwatchagent_types.go b/apis/v1alpha1/amazoncloudwatchagent_types.go index 7605df20a..f511a3929 100644 --- a/apis/v1alpha1/amazoncloudwatchagent_types.go +++ b/apis/v1alpha1/amazoncloudwatchagent_types.go @@ -381,6 +381,14 @@ type AmazonCloudWatchAgentTargetAllocatorPrometheusCR struct { // ServiceMonitor's meta labels. The requirements are ANDed. // +optional ServiceMonitorSelector map[string]string `json:"serviceMonitorSelector,omitempty"` + // ScraperRole partitions ServiceMonitor/PodMonitor discovery across CloudWatch agents by the + // "cloudwatch.aws/scraper" annotation on the monitor CR. "cluster-scraper" selects only monitors + // annotated cloudwatch.aws/scraper: cluster-scraper; empty (default) selects only monitors that are + // not so annotated. This lets a heavy/singleton monitor be routed to the central cluster-scraper + // agent while all others stay on the per-node agent. + // +optional + // +kubebuilder:validation:Enum=cluster-scraper + ScraperRole string `json:"scraperRole,omitempty"` } // ScaleSubresourceStatus defines the observed state of the AmazonCloudWatchAgent's diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go index 33b68ef75..12dad1034 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config.go @@ -63,6 +63,7 @@ type Config struct { PrometheusCR PrometheusCRConfig `yaml:"prometheus_cr,omitempty"` PodMonitorSelector map[string]string `yaml:"pod_monitor_selector,omitempty"` ServiceMonitorSelector map[string]string `yaml:"service_monitor_selector,omitempty"` + ScraperRole string `yaml:"scraper_role,omitempty"` CollectorSelector *metav1.LabelSelector `yaml:"collector_selector,omitempty"` HTTPS HTTPSServerConfig `yaml:"https,omitempty"` } diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 5dfc314dd..64dd41884 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go @@ -98,6 +98,7 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr kubeConfigPath: cfg.KubeConfigFilePath, serviceMonitorSelector: servMonSelector, podMonitorSelector: podMonSelector, + scraperRole: cfg.ScraperRole, }, nil } @@ -114,6 +115,7 @@ type PrometheusCRWatcher struct { serviceMonitorSelector labels.Selector podMonitorSelector labels.Selector + scraperRole string } func getSelector(s map[string]string) labels.Selector { @@ -123,6 +125,53 @@ func getSelector(s map[string]string) labels.Selector { return labels.SelectorFromSet(s) } +// ScraperAnnotationKey and clusterScraperRole implement annotation-based routing of +// ServiceMonitor/PodMonitor CRs across CloudWatch agents. A monitor annotated +// cloudwatch.aws/scraper: cluster-scraper is scraped only by the cluster-scraper agent's Target +// Allocator; all others are scraped only by the per-node agent's Target Allocator. +const ( + ScraperAnnotationKey = "cloudwatch.aws/scraper" + clusterScraperRole = "cluster-scraper" +) + +// annotationRoleMatches reports whether a monitor with the given annotations +// belongs to scraperRole. The cluster-scraper role claims monitors annotated +// cloudwatch.aws/scraper: cluster-scraper; the default role (empty) claims only +// monitors that are not so annotated — so the two roles partition monitors with +// no overlap and no gap. +// +// Routing is intentionally BINARY today: every role other than cluster-scraper +// (i.e. the default per-node agent) claims the rest. NOTE: before a third role is +// introduced (e.g. a "gpu-scraper"), this must become an explicit +// role-to-annotation-value match — otherwise any unrecognized role would +// silently fall through to the per-node bucket here. +func annotationRoleMatches(scraperRole string, annotations map[string]string) bool { + routed := annotations[ScraperAnnotationKey] == clusterScraperRole + if scraperRole == clusterScraperRole { + return routed + } + return !routed +} + +// selectsMonitor reports whether a discovered ServiceMonitor/PodMonitor belongs to this Target +// Allocator's scraper role (see annotationRoleMatches). When this allocator is the cluster-scraper, +// it logs each monitor it claims, because that monitor was explicitly overridden onto the +// cluster-scraper via the cloudwatch.aws/scraper annotation. +func (w *PrometheusCRWatcher) selectsMonitor(kind, namespace, name string, annotations map[string]string) bool { + if !annotationRoleMatches(w.scraperRole, annotations) { + return false + } + if w.scraperRole == clusterScraperRole { + // V(1): LoadConfig reruns on every reconcile, so this fires per claimed + // monitor each cycle — keep it at debug verbosity to avoid log spam on + // large annotated sets. + w.logger.V(1).Info("routing monitor to cluster-scraper via annotation", + "kind", kind, "namespace", namespace, "name", name, + "annotation", ScraperAnnotationKey+"="+clusterScraperRole) + } + return true +} + // getInformers returns a map of informers for the given resources. func getInformers(factory informers.FactoriesForNamespaces) (map[string]*informers.ForResource, error) { serviceMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.ServiceMonitorName)) @@ -231,6 +280,10 @@ func (w *PrometheusCRWatcher) LoadConfig(ctx context.Context) (*promconfig.Confi serviceMonitorInstances := make(map[string]*monitoringv1.ServiceMonitor) smRetrieveErr := w.informers[monitoringv1.ServiceMonitorName].ListAll(w.serviceMonitorSelector, func(sm interface{}) { monitor := sm.(*monitoringv1.ServiceMonitor) + // Annotation-based routing: skip monitors that belong to the other agent's scraper role. + if !w.selectsMonitor("ServiceMonitor", monitor.Namespace, monitor.Name, monitor.GetAnnotations()) { + return + } key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor) w.addStoreAssetsForServiceMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.Endpoints, store) serviceMonitorInstances[key] = monitor @@ -242,6 +295,10 @@ func (w *PrometheusCRWatcher) LoadConfig(ctx context.Context) (*promconfig.Confi podMonitorInstances := make(map[string]*monitoringv1.PodMonitor) pmRetrieveErr := w.informers[monitoringv1.PodMonitorName].ListAll(w.podMonitorSelector, func(pm interface{}) { monitor := pm.(*monitoringv1.PodMonitor) + // Annotation-based routing: skip monitors that belong to the other agent's scraper role. + if !w.selectsMonitor("PodMonitor", monitor.Namespace, monitor.Name, monitor.GetAnnotations()) { + return + } key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor) w.addStoreAssetsForPodMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.PodMetricsEndpoints, store) podMonitorInstances[key] = monitor diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_annotation_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_annotation_test.go new file mode 100644 index 000000000..44251434a --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator_annotation_test.go @@ -0,0 +1,190 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: Apache-2.0 + +package watcher + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/go-logr/logr/funcr" + monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/utils/ptr" +) + +// TestAnnotationRoleMatches verifies the annotation-based routing partition: the cluster-scraper +// role selects only monitors annotated cloudwatch.aws/scraper: cluster-scraper, and the default +// role (empty) selects only monitors that are not so annotated. The two roles are complementary, +// so every monitor is owned by exactly one role (no overlap, no gap). +func TestAnnotationRoleMatches(t *testing.T) { + routed := map[string]string{ScraperAnnotationKey: clusterScraperRole} + other := map[string]string{ScraperAnnotationKey: "something-else"} + none := map[string]string{"unrelated": "x"} + + cases := []struct { + name string + role string + annotations map[string]string + want bool + }{ + {"cluster-scraper claims routed", clusterScraperRole, routed, true}, + {"cluster-scraper skips unannotated", clusterScraperRole, none, false}, + {"cluster-scraper skips other value", clusterScraperRole, other, false}, + {"cluster-scraper skips nil", clusterScraperRole, nil, false}, + {"default claims unannotated", "", none, true}, + {"default claims nil", "", nil, true}, + {"default claims other value", "", other, true}, + {"default skips routed", "", routed, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := annotationRoleMatches(tc.role, tc.annotations); got != tc.want { + t.Fatalf("annotationRoleMatches(%q, %v) = %v, want %v", tc.role, tc.annotations, got, tc.want) + } + }) + } + + // Partition invariant: for any monitor, exactly one role claims it. + for _, ann := range []map[string]string{routed, other, none, nil} { + cs := annotationRoleMatches(clusterScraperRole, ann) + def := annotationRoleMatches("", ann) + if cs == def { + t.Fatalf("partition broken for %v: cluster-scraper=%v default=%v (must differ)", ann, cs, def) + } + } +} + +// TestSelectsMonitor covers PrometheusCRWatcher.selectsMonitor for both scraper roles, and asserts +// the cluster-scraper agent logs each monitor it claims (the override event) while the per-node +// agent logs nothing. +func TestSelectsMonitor(t *testing.T) { + routed := map[string]string{ScraperAnnotationKey: clusterScraperRole} + none := map[string]string{} + + cases := []struct { + name string + role string + annotations map[string]string + wantSelect bool + wantLog bool + }{ + {"cluster-scraper claims routed (logs override)", clusterScraperRole, routed, true, true}, + {"cluster-scraper skips unannotated", clusterScraperRole, none, false, false}, + {"default claims unannotated (no log)", "", none, true, false}, + {"default skips routed", "", routed, false, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var logs []string + // Verbosity 1 so the V(1) cluster-scraper routing log is captured. + logger := funcr.New(func(prefix, args string) { logs = append(logs, args) }, funcr.Options{Verbosity: 1}) + w := &PrometheusCRWatcher{logger: logger, scraperRole: tc.role} + + got := w.selectsMonitor("PodMonitor", "ns", "mon", tc.annotations) + assert.Equal(t, tc.wantSelect, got, "selectsMonitor result") + + logged := strings.Join(logs, "\n") + if tc.wantLog { + assert.Contains(t, logged, "cluster-scraper", "expected an override log mentioning cluster-scraper") + assert.Contains(t, logged, "mon", "override log should identify the monitor") + } else { + assert.Empty(t, logs, "no override log expected for role=%q select=%v", tc.role, got) + } + }) + } +} + +// TestLoadConfigScraperRouting exercises the annotation filter through the real LoadConfig path +// (matching TestLoadConfig's harness): an annotated ServiceMonitor is discovered by the +// cluster-scraper role and excluded by the default role. +func TestLoadConfigScraperRouting(t *testing.T) { + const monName = "routed" + newSM := func(annotated bool) *monitoringv1.ServiceMonitor { + ann := map[string]string{} + if annotated { + ann[ScraperAnnotationKey] = clusterScraperRole + } + return &monitoringv1.ServiceMonitor{ + ObjectMeta: metav1.ObjectMeta{Name: monName, Namespace: "test", Annotations: ann}, + Spec: monitoringv1.ServiceMonitorSpec{Endpoints: []monitoringv1.Endpoint{{Port: "metrics"}}}, + } + } + newPM := func(annotated bool) *monitoringv1.PodMonitor { + ann := map[string]string{} + if annotated { + ann[ScraperAnnotationKey] = clusterScraperRole + } + return &monitoringv1.PodMonitor{ + ObjectMeta: metav1.ObjectMeta{Name: monName, Namespace: "test", Annotations: ann}, + Spec: monitoringv1.PodMonitorSpec{PodMetricsEndpoints: []monitoringv1.PodMetricsEndpoint{{Port: ptr.To("metrics")}}}, + } + } + + // PodMonitor discovery mirrors ServiceMonitor, so both kinds are exercised + // through LoadConfig for both roles, annotated and unannotated. + cases := []struct { + name string + kind string // "ServiceMonitor" or "PodMonitor" + annotated bool + role string + wantSelected bool + }{ + {"cluster-scraper claims annotated ServiceMonitor", "ServiceMonitor", true, clusterScraperRole, true}, + {"default excludes annotated ServiceMonitor", "ServiceMonitor", true, "", false}, + {"default claims unannotated ServiceMonitor", "ServiceMonitor", false, "", true}, + {"cluster-scraper excludes unannotated ServiceMonitor", "ServiceMonitor", false, clusterScraperRole, false}, + {"cluster-scraper claims annotated PodMonitor", "PodMonitor", true, clusterScraperRole, true}, + {"default excludes annotated PodMonitor", "PodMonitor", true, "", false}, + {"default claims unannotated PodMonitor", "PodMonitor", false, "", true}, + {"cluster-scraper excludes unannotated PodMonitor", "PodMonitor", false, clusterScraperRole, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var sm *monitoringv1.ServiceMonitor + var pm *monitoringv1.PodMonitor + if tc.kind == "ServiceMonitor" { + sm = newSM(tc.annotated) + } else { + pm = newPM(tc.annotated) + } + + w := getTestPrometheusCRWatcher(t, sm, pm) + w.logger = logr.Discard() + w.scraperRole = tc.role + defer func() { _ = w.Close() }() + + for _, informer := range w.informers { + informer.Start(w.stopChannel) + } + // Bounded sync wait instead of an unbounded HasSynced spin loop, so a + // stuck informer fails fast rather than hanging until CI kills it. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + for _, informer := range w.informers { + require.True(t, cache.WaitForCacheSync(ctx.Done(), informer.HasSynced), + "informer cache did not sync before timeout") + } + + got, err := w.LoadConfig(context.Background()) + require.NoError(t, err) + + var found bool + for _, sc := range got.ScrapeConfigs { + if strings.Contains(sc.JobName, monName) { + found = true + break + } + } + assert.Equalf(t, tc.wantSelected, found, + "%s discovered=%v, want %v (annotated=%v, scraperRole=%q)", + tc.kind, found, tc.wantSelected, tc.annotated, tc.role) + }) + } +} diff --git a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml index ed7e7be6c..381ec2ddb 100644 --- a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml +++ b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml @@ -6858,6 +6858,16 @@ spec: Default: "30s" format: duration type: string + scraperRole: + description: |- + ScraperRole partitions ServiceMonitor/PodMonitor discovery across CloudWatch agents by the + "cloudwatch.aws/scraper" annotation on the monitor CR. "cluster-scraper" selects only monitors + annotated cloudwatch.aws/scraper: cluster-scraper; empty (default) selects only monitors that are + not so annotated. This lets a heavy/singleton monitor be routed to the central cluster-scraper + agent while all others stay on the per-node agent. + enum: + - cluster-scraper + type: string serviceMonitorSelector: additionalProperties: type: string diff --git a/internal/manifests/targetallocator/configmap.go b/internal/manifests/targetallocator/configmap.go index 94e4e72da..fc1245ebf 100644 --- a/internal/manifests/targetallocator/configmap.go +++ b/internal/manifests/targetallocator/configmap.go @@ -80,6 +80,10 @@ func ConfigMap(params manifests.Params) (*corev1.ConfigMap, error) { taConfig["pod_monitor_selector"] = ¶ms.OtelCol.Spec.TargetAllocator.PrometheusCR.PodMonitorSelector } + if len(params.OtelCol.Spec.TargetAllocator.PrometheusCR.ScraperRole) > 0 { + taConfig["scraper_role"] = params.OtelCol.Spec.TargetAllocator.PrometheusCR.ScraperRole + } + if len(prometheusCRConfig) > 0 { taConfig["prometheus_cr"] = prometheusCRConfig } diff --git a/internal/manifests/targetallocator/configmap_test.go b/internal/manifests/targetallocator/configmap_test.go index 56d009980..5e1d5f8aa 100644 --- a/internal/manifests/targetallocator/configmap_test.go +++ b/internal/manifests/targetallocator/configmap_test.go @@ -80,6 +80,7 @@ label_selector: app.kubernetes.io/part-of: amazon-cloudwatch-agent pod_monitor_selector: release: my-instance +scraper_role: cluster-scraper service_monitor_selector: release: my-instance `, @@ -91,6 +92,7 @@ service_monitor_selector: instance.Spec.TargetAllocator.PrometheusCR.ServiceMonitorSelector = map[string]string{ "release": "my-instance", } + instance.Spec.TargetAllocator.PrometheusCR.ScraperRole = "cluster-scraper" cfg := config.New() params := manifests.Params{ OtelCol: instance, @@ -183,4 +185,42 @@ label_selector: assert.Equal(t, expectedData, actual.Data) }) + t.Run("should emit scraper_role when set on the prometheusCR", 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_strategy: consistent-hashing +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 +scraper_role: cluster-scraper +`, + } + instance := collectorInstance() + instance.Spec.TargetAllocator.PrometheusCR.ScraperRole = "cluster-scraper" + 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) + }) + }