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 a9b7179e4..e4fa75d31 100644 --- a/apis/v1alpha1/amazoncloudwatchagent_types.go +++ b/apis/v1alpha1/amazoncloudwatchagent_types.go @@ -301,7 +301,7 @@ type AmazonCloudWatchAgentTargetAllocator struct { // +optional Resources v1.ResourceRequirements `json:"resources,omitempty"` // AllocationStrategy determines which strategy the target allocator should use for allocation. - // The current option is consistent-hashing. + // The options are consistent-hashing and per-node. // +optional AllocationStrategy AmazonCloudWatchAgentTargetAllocatorAllocationStrategy `json:"allocationStrategy,omitempty"` // FilterStrategy determines how to filter targets before allocating them among the collectors. 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..c0257e47c --- /dev/null +++ b/cmd/amazon-cloudwatch-agent-target-allocator/allocation/per_node_test.go @@ -0,0 +1,452 @@ +// 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..a0136ace3 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,23 @@ func TestCollectorDiff(t *testing.T) { }) } } + + +// TestWithFallbackStrategy covers the option's no-op branches: an empty name, +// and an allocator that does not support a fallback strategy. +func TestWithFallbackStrategy(t *testing.T) { + // Empty name is a no-op: the per-node fallback stays disabled. + a, err := New(perNodeStrategyName, logger, WithFallbackStrategy("")) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if pn := a.(*perNodeAllocator); pn.fallbackHasher != nil { + t.Error("empty fallback name must leave the fallback disabled") + } + + // An allocator that does not implement fallbackStrategySetter (consistent + // hashing) silently ignores the option. + if _, err := New(consistentHashingStrategyName, logger, WithFallbackStrategy(consistentHashingStrategyName)); err != nil { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/cmd/amazon-cloudwatch-agent-target-allocator/collector/collector.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 2272eedbe..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 @@ -119,6 +147,13 @@ func LoadFromCLI(target *Config, flagSet *pflag.FlagSet) error { return err } + // OR the CLI flag into the YAML value so either source can enable the watcher. + 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/config_test.go b/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go index ea6332afd..64cc2d69c 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/config/config_test.go @@ -19,16 +19,16 @@ func TestLoad(t *testing.T) { file string } tests := []struct { - name string - args args - wantErr assert.ErrorAssertionFunc - wantHTTPS HTTPSServerConfig - wantLabels map[string]string - wantPromCR PrometheusCRConfig - wantAlloc *string - wantPodMonSel map[string]string - wantSvcMonSel map[string]string - wantJobNames []string + name string + args args + wantErr assert.ErrorAssertionFunc + wantHTTPS HTTPSServerConfig + wantLabels map[string]string + wantPromCR PrometheusCRConfig + wantAlloc *string + wantPodMonSel map[string]string + wantSvcMonSel map[string]string + wantJobNames []string }{ { name: "file sd load", @@ -58,8 +58,8 @@ func TestLoad(t *testing.T) { args: args{ file: "./testdata/no_config.yaml", }, - wantErr: assert.NoError, - wantHTTPS: CreateDefaultConfig().HTTPS, + wantErr: assert.NoError, + wantHTTPS: CreateDefaultConfig().HTTPS, wantLabels: nil, wantPromCR: CreateDefaultConfig().PrometheusCR, wantAlloc: CreateDefaultConfig().AllocationStrategy, @@ -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/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/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 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..db5aa1d4f 100644 --- a/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go +++ b/cmd/amazon-cloudwatch-agent-target-allocator/target/target.go @@ -10,6 +10,19 @@ import ( "github.com/prometheus/common/model" ) +// nodeLabels are the discovery meta-labels that identify the node a target +// resides on. They mirror the upstream OpenTelemetry target allocator's +// per-node node-label set. See: +// https://prometheus.io/docs/prometheus/latest/configuration/configuration/#kubernetes_sd_config +const ( + nodeNameLabelPod model.LabelName = "__meta_kubernetes_pod_node_name" + nodeNameLabelNode model.LabelName = "__meta_kubernetes_node_name" + nodeNameLabelEndpoint model.LabelName = "__meta_kubernetes_endpoint_node_name" + + endpointSliceTargetKindLabel model.LabelName = "__meta_kubernetes_endpointslice_address_target_kind" + endpointSliceTargetNameLabel model.LabelName = "__meta_kubernetes_endpointslice_address_target_name" +) + // LinkJSON This package contains common structs and methods that relate to scrape targets. type LinkJSON struct { Link string `json:"_link"` @@ -28,6 +41,25 @@ func (t *Item) Hash() string { return t.hash } +// GetNodeName returns the Kubernetes node a target resides on, derived from its +// service-discovery meta labels. Pod targets (PodMonitor, role: pod) always carry +// a node; endpoint targets (ServiceMonitor, role: endpoints/endpointslice) only +// carry one when the endpoint is backed by a pod on a node. Returns "" when no +// node can be determined (e.g. non-pod / external endpoints), in which case the +// per-node strategy leaves the target unassigned. +func (t *Item) GetNodeName() string { + for _, labelName := range []model.LabelName{nodeNameLabelPod, nodeNameLabelNode, nodeNameLabelEndpoint} { + if val := t.Labels[labelName]; val != "" { + return string(val) + } + } + + if t.Labels[endpointSliceTargetKindLabel] != "Node" { + return "" + } + return string(t.Labels[endpointSliceTargetNameLabel]) +} + // NewItem Creates a new target item. // INVARIANTS: // * Item fields must not be modified after creation. diff --git a/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/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go b/cmd/amazon-cloudwatch-agent-target-allocator/watcher/promOperator.go index 2651e8d29..5dfc314dd 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" @@ -28,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) { @@ -49,11 +52,26 @@ func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config) (*Pr } // TODO: We should make these durations configurable + // Namespace must be non-empty; the config generator panics otherwise. + collectorNamespace := os.Getenv("OTELCOL_NAMESPACE") + if collectorNamespace == "" { + 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{ + Namespace: collectorNamespace, + }, Spec: monitoringv1.PrometheusSpec{ CommonPrometheusFields: monitoringv1.CommonPrometheusFields{ ScrapeInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, + // Must be non-empty; default to scrape interval. + EvaluationInterval: monitoringv1.Duration(cfg.PrometheusCR.ScrapeInterval.String()), }, } diff --git a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml index d3271ac1c..e816d834d 100644 --- a/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml +++ b/config/crd/bases/cloudwatch.aws.amazon.com_amazoncloudwatchagents.yaml @@ -6630,9 +6630,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/collector/annotations.go b/internal/manifests/collector/annotations.go index a665e465e..364072330 100644 --- a/internal/manifests/collector/annotations.go +++ b/internal/manifests/collector/annotations.go @@ -6,6 +6,7 @@ package collector import ( "crypto/sha256" "fmt" + "log/slog" "github.com/aws/amazon-cloudwatch-agent-operator/apis/v1alpha1" ) @@ -28,7 +29,7 @@ func Annotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string { } // make sure sha256 for configMap is always calculated - annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config) + annotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance)) return annotations } @@ -51,11 +52,27 @@ func PodAnnotations(instance v1alpha1.AmazonCloudWatchAgent) map[string]string { } // make sure sha256 for configMap is always calculated - podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(instance.Spec.Config) + podAnnotations["amazon-cloudwatch-agent-operator-config/sha256"] = getConfigMapSHA(configHashInput(instance)) return podAnnotations } +// configHashInput returns the combined config string used for the pod-template restart hash. +func configHashInput(instance v1alpha1.AmazonCloudWatchAgent) string { + config := instance.Spec.Config + if !instance.Spec.Prometheus.IsEmpty() { + promYaml, err := instance.Spec.Prometheus.Yaml() + if err != nil { + // Static sentinel; Yaml() over map[string]interface{} rarely fails in practice. + slog.Warn("failed to serialize Spec.Prometheus for config hash", "error", err) + config += "\x00prometheus-serialize-error" + } else { + config += "\x00" + promYaml // null byte prevents collision between config suffix and promYAML prefix + } + } + return config +} + func getConfigMapSHA(config string) string { h := sha256.Sum256([]byte(config)) return fmt.Sprintf("%x", h) diff --git a/internal/manifests/collector/annotations_test.go b/internal/manifests/collector/annotations_test.go index e02ef1678..1ab8e3bd7 100644 --- a/internal/manifests/collector/annotations_test.go +++ b/internal/manifests/collector/annotations_test.go @@ -79,3 +79,64 @@ func TestAnnotationsPropagateDown(t *testing.T) { assert.Equal(t, "mycomponent", podAnnotations["myapp"]) assert.Equal(t, "pod_annotation_value", podAnnotations["pod_annotation"]) } + +func promConfig(t *testing.T, replacement string) v1alpha1.PrometheusConfig { + t.Helper() + cfg := map[string]interface{}{ + "scrape_configs": []interface{}{ + map[string]interface{}{ + "job_name": "kubernetes-pods-annotated", + "relabel_configs": []interface{}{ + map[string]interface{}{ + "target_label": "bug2probe", + "replacement": replacement, + }, + }, + }, + }, + } + return v1alpha1.PrometheusConfig{ + Config: &v1alpha1.AnyConfig{Object: cfg}, + } +} + +// TestPrometheusConfigChangeBumpsHash verifies a Prometheus-only change bumps the pod-template hash. +func TestPrometheusConfigChangeBumpsHash(t *testing.T) { + base := v1alpha1.AmazonCloudWatchAgent{ + ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"}, + Spec: v1alpha1.AmazonCloudWatchAgentSpec{ + Config: "agent-config", + Prometheus: promConfig(t, "value2"), + }, + } + + // same spec twice -> stable hash + h1 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + h2 := PodAnnotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.Equal(t, h1, h2, "hash must be stable when nothing changes") + + // change ONLY the prometheus config -> hash must change + changed := base + changed.Spec.Prometheus = promConfig(t, "value3") + h3 := PodAnnotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.NotEqual(t, h1, h3, "pod annotation hash must change when only Spec.Prometheus changes") + + // metadata annotations hash must also reflect the prometheus change + a1 := Annotations(base)["amazon-cloudwatch-agent-operator-config/sha256"] + a3 := Annotations(changed)["amazon-cloudwatch-agent-operator-config/sha256"] + assert.NotEqual(t, a1, a3, "metadata annotation hash must change when only Spec.Prometheus changes") +} + +// TestEmptyPrometheusHashUnchanged verifies non-Prometheus agents keep a stable hash. +func TestEmptyPrometheusHashUnchanged(t *testing.T) { + otelcol := v1alpha1.AmazonCloudWatchAgent{ + ObjectMeta: metav1.ObjectMeta{Name: "my-instance", Namespace: "my-ns"}, + Spec: v1alpha1.AmazonCloudWatchAgentSpec{Config: "test"}, + } + + // sha256("test") == 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08 + assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", + Annotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"]) + assert.Equal(t, "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08", + PodAnnotations(otelcol)["amazon-cloudwatch-agent-operator-config/sha256"]) +} 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) + }) }