Address PR review: tighten version gate and include mode in checksum

- Bump podCollectionOnNodeMinVersion from 7.58 to 7.60 to match the documented
  supported floor (the start-up fix landed in 7.60).
- Gate on BOTH sides: skip the feature if the node-agent override image is
  parseable AND < 7.60 in addition to the existing cluster-side check
  (cluster-agent or CCR). A cluster-agent-only upgrade can no longer mount a
  pod_collection_mode: node_kubelet check into older node-agents.
- Move PodCollectionMode resolution above the default-config checksum and
  include pod_collection_on_node in the checksum input, so toggling the field
  changes the operator-managed cluster-agent pod-template annotation and
  forces a rollout instead of leaving the in-memory ConfigMap stale.
- Tests: add WithNodeAgentImage builder, cover the node-agent-image fallback
  row, and extend ksmClusterAgentWantFunc with a withPodCollectionOnNode()
  option so expected-hash dicts include the new key for the active rows.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: zhuminyi

internal/controller/datadogagent/feature/kubernetesstatecore/const.go

@@ -29,10 +29,11 @@ const (
 	// one of these alongside the cluster-side ConfigMap).
 	defaultKSMPodsOnNodeConf string = "kube-state-metrics-core-pods-on-node-config"
 
-	// Minimum agent / cluster-agent version supporting the
+	// Minimum agent / cluster-agent / node-agent version supporting the
 	// pod_collection_mode field used by PodCollectionMode=node_kubelet.
-	// node_kubelet shipped in 7.58; the startup fix landed in 7.60.
-	podCollectionOnNodeMinVersion = "7.58.0-0"
+	// node_kubelet shipped in 7.58; the startup fix landed in 7.60, so 7.60
+	// is the supported floor across all components that load the check.
+	podCollectionOnNodeMinVersion = "7.60.0-0"
 )
 
 // GetKubeStateMetricsRBACResourceName return the RBAC resources name

internal/controller/datadogagent/feature/kubernetesstatecore/feature.go

@@ -157,69 +157,54 @@ func (f *ksmFeature) Configure(dda metav1.Object, ddaSpec *v2alpha1.DatadogAgent
 			}
 		}
 
+		// Capture the user-supplied custom config (if any) so PodCollectionMode
+		// resolution below can decide whether to mutate the cluster-side YAML.
 		if ddaSpec.Features.KubeStateMetricsCore.Conf != nil {
 			f.customConfig = ddaSpec.Features.KubeStateMetricsCore.Conf
-			hash, err := comparison.GenerateMD5ForSpec(f.customConfig)
-			if err != nil {
-				f.logger.Error(err, "couldn't generate hash for ksm core custom config")
-			} else {
-				f.logger.V(2).Info("built ksm core from custom config", "hash", hash)
-			}
-			f.customConfigAnnotationValue = hash
-			f.customConfigAnnotationKey = object.GetChecksumAnnotationKey(feature.KubernetesStateCoreIDType)
-		} else {
-			// Generate dynamic checksum for default configuration (based on user provided collectCrMetrics field and whether or not APIServices/CRD metrics are collected)
-			defaultConfigData := map[string]any{
-				"collect_crds":        f.collectCRDMetrics,
-				"collect_apiservices": f.collectAPIServiceMetrics,
-				"collect_cr_metrics":  f.collectCrMetrics,
-			}
-
-			hash, err := comparison.GenerateMD5ForSpec(defaultConfigData)
-			if err != nil {
-				f.logger.Error(err, "couldn't generate hash for default ksm core config")
-			} else {
-				f.logger.V(2).Info("generated default ksm core config hash", "hash", hash, "config", defaultConfigData)
-			}
-			f.customConfigAnnotationValue = hash
-			f.customConfigAnnotationKey = object.GetChecksumAnnotationKey(feature.KubernetesStateCoreIDType)
 		}
 
-		f.configConfigMapName = constants.GetConfName(dda, f.customConfig, defaultKubeStateMetricsCoreConf)
-
-		// Resolve PodCollectionMode. When node_kubelet is requested AND the
-		// agent version supports it, switch the cluster-side instance to
-		// pod_collection_mode: cluster_unassigned (only when the operator owns
-		// the cluster-side config) and deploy a pods-only check to every node
-		// agent. When the user supplies their own .Conf, the operator still
-		// deploys the node-side check but does not mutate the user's YAML;
-		// they are responsible for setting cluster_unassigned themselves.
+		// Resolve PodCollectionMode. When node_kubelet is requested AND every
+		// component that loads the check is version-compatible, switch the
+		// cluster-side instance to pod_collection_mode: cluster_unassigned
+		// (only when the operator owns the cluster-side config) and deploy a
+		// pods-only check to every node agent. When the user supplies their
+		// own .Conf, the operator still deploys the node-side check but does
+		// not mutate the user's YAML; they are responsible for setting
+		// cluster_unassigned themselves.
 		if mode := ddaSpec.Features.KubeStateMetricsCore.PodCollectionMode; mode != nil &&
 			*mode == v2alpha1.KSMPodCollectionModeNodeKubelet {
 			f.podCollectionOnNode = true
-			// Version compatibility check on the component that runs the
-			// cluster-side check (CCR if cluster-checks-runners are enabled,
-			// otherwise the cluster-agent). Unparseable tags (`:dev`, custom
-			// registries, etc.) are assumed compatible — matches the existing
-			// pattern for the apiservices/CRD checks above.
-			var override *v2alpha1.DatadogAgentComponentOverride
-			if ovr, ok := ddaSpec.Override[v2alpha1.ClusterAgentComponentName]; ok {
-				override = ovr
-			}
+			// Version compatibility check on BOTH sides: the cluster-side
+			// component that runs the cluster_unassigned check (CCR if cluster-
+			// checks-runners are enabled, otherwise the cluster-agent) AND the
+			// node-agent that runs the node_kubelet check. If either image tag
+			// is parseable AND below the supported floor, skip the feature so
+			// the operator doesn't mount an unsupported file into a node-agent
+			// that would silently fall back to default mode and double-collect.
+			// Unparseable tags (`:dev`, custom registries, etc.) are assumed
+			// compatible — matches the existing pattern for the apiservices/CRD
+			// checks above.
+			componentsToCheck := []v2alpha1.ComponentName{v2alpha1.NodeAgentComponentName}
 			if f.runInClusterChecksRunner {
-				if ovr, ok := ddaSpec.Override[v2alpha1.ClusterChecksRunnerComponentName]; ok {
-					override = ovr
-				}
+				componentsToCheck = append(componentsToCheck, v2alpha1.ClusterChecksRunnerComponentName)
+			} else {
+				componentsToCheck = append(componentsToCheck, v2alpha1.ClusterAgentComponentName)
 			}
-			if override != nil && override.Image != nil {
-				agentVersion := common.GetAgentVersionFromImage(*override.Image)
+			for _, comp := range componentsToCheck {
+				ovr, ok := ddaSpec.Override[comp]
+				if !ok || ovr == nil || ovr.Image == nil {
+					continue
+				}
+				agentVersion := common.GetAgentVersionFromImage(*ovr.Image)
 				fallback := true // assume compatible when unparseable
 				if !utils.IsAboveMinVersion(agentVersion, podCollectionOnNodeMinVersion, &fallback) {
 					f.logger.Info(
-						"PodCollectionMode=node_kubelet requires agent >= 7.58; falling back to default",
+						"PodCollectionMode=node_kubelet requires agent >= 7.60; falling back to default",
+						"component", string(comp),
 						"version", agentVersion,
 					)
 					f.podCollectionOnNode = false
+					break
 				}
 			}
 			if f.podCollectionOnNode {
@@ -237,6 +222,41 @@ func (f *ksmFeature) Configure(dda metav1.Object, ddaSpec *v2alpha1.DatadogAgent
 				}
 			}
 		}
+
+		// Compute the checksum annotation. With f.podCollectionOnNode resolved
+		// above, toggling the field changes the input here, which propagates
+		// to the cluster-agent pod-template annotation and forces a rollout.
+		if f.customConfig != nil {
+			hash, err := comparison.GenerateMD5ForSpec(f.customConfig)
+			if err != nil {
+				f.logger.Error(err, "couldn't generate hash for ksm core custom config")
+			} else {
+				f.logger.V(2).Info("built ksm core from custom config", "hash", hash)
+			}
+			f.customConfigAnnotationValue = hash
+			f.customConfigAnnotationKey = object.GetChecksumAnnotationKey(feature.KubernetesStateCoreIDType)
+		} else {
+			// Dynamic checksum for the default configuration. Includes every
+			// input that affects the rendered cluster-side ConfigMap so that
+			// toggling any of them forces a rollout of the consumer.
+			defaultConfigData := map[string]any{
+				"collect_crds":           f.collectCRDMetrics,
+				"collect_apiservices":    f.collectAPIServiceMetrics,
+				"collect_cr_metrics":     f.collectCrMetrics,
+				"pod_collection_on_node": f.podCollectionOnNode,
+			}
+
+			hash, err := comparison.GenerateMD5ForSpec(defaultConfigData)
+			if err != nil {
+				f.logger.Error(err, "couldn't generate hash for default ksm core config")
+			} else {
+				f.logger.V(2).Info("generated default ksm core config hash", "hash", hash, "config", defaultConfigData)
+			}
+			f.customConfigAnnotationValue = hash
+			f.customConfigAnnotationKey = object.GetChecksumAnnotationKey(feature.KubernetesStateCoreIDType)
+		}
+
+		f.configConfigMapName = constants.GetConfName(dda, f.customConfig, defaultKubeStateMetricsCoreConf)
 	}
 
 	return output

internal/controller/datadogagent/feature/kubernetesstatecore/feature_test.go

@@ -153,7 +153,7 @@ func Test_ksmFeature_Configure(t *testing.T) {
 				WithKSMPodCollectionMode(v2alpha1.KSMPodCollectionModeNodeKubelet).
 				Build(),
 			WantConfigure: true,
-			ClusterAgent:  ksmClusterAgentWantFunc(false),
+			ClusterAgent:  ksmClusterAgentWantFunc(false, withPodCollectionOnNode()),
 			Agent:         test.NewDefaultComponentTest().WithWantFunc(ksmAgentNodeWantFuncWithPodsOnNode),
 		},
 		{
@@ -164,7 +164,7 @@ func Test_ksmFeature_Configure(t *testing.T) {
 				WithSingleContainerStrategy(true).
 				Build(),
 			WantConfigure: true,
-			ClusterAgent:  ksmClusterAgentWantFunc(false),
+			ClusterAgent:  ksmClusterAgentWantFunc(false, withPodCollectionOnNode()),
 			Agent:         test.NewDefaultComponentTest().WithWantFunc(ksmAgentSingleAgentWantFuncWithPodsOnNode),
 		},
 		{
@@ -179,11 +179,22 @@ func Test_ksmFeature_Configure(t *testing.T) {
 			Agent:         test.NewDefaultComponentTest().WithWantFunc(ksmAgentNodeWantFuncWithPodsOnNode),
 		},
 		{
-			Name: "ksm-core enabled, podCollectionMode=node_kubelet but cluster-agent image < 7.58 -> fall back",
+			Name: "ksm-core enabled, podCollectionMode=node_kubelet but cluster-agent image < 7.60 -> fall back",
 			DDA: testutils.NewDatadogAgentBuilder().
 				WithKSMEnabled(true).
 				WithKSMPodCollectionMode(v2alpha1.KSMPodCollectionModeNodeKubelet).
-				WithClusterAgentImage("gcr.io/datadoghq/cluster-agent:7.57.0").
+				WithClusterAgentImage("gcr.io/datadoghq/cluster-agent:7.59.0").
+				Build(),
+			WantConfigure: true,
+			ClusterAgent:  ksmClusterAgentWantFunc(false),
+			Agent:         test.NewDefaultComponentTest().WithWantFunc(ksmAgentNodeWantFunc),
+		},
+		{
+			Name: "ksm-core enabled, podCollectionMode=node_kubelet but node-agent image < 7.60 -> fall back",
+			DDA: testutils.NewDatadogAgentBuilder().
+				WithKSMEnabled(true).
+				WithKSMPodCollectionMode(v2alpha1.KSMPodCollectionModeNodeKubelet).
+				WithNodeAgentImage("gcr.io/datadoghq/agent:7.59.0").
 				Build(),
 			WantConfigure: true,
 			ClusterAgent:  ksmClusterAgentWantFunc(false),
@@ -194,7 +205,21 @@ func Test_ksmFeature_Configure(t *testing.T) {
 	tests.Run(t, buildKSMFeature)
 }
 
-func ksmClusterAgentWantFunc(hasCustomConfig bool) *test.ComponentTest {
+type ksmClusterAgentWantConfig struct {
+	podCollectionOnNode bool
+}
+
+type ksmClusterAgentOption func(*ksmClusterAgentWantConfig)
+
+func withPodCollectionOnNode() ksmClusterAgentOption {
+	return func(c *ksmClusterAgentWantConfig) { c.podCollectionOnNode = true }
+}
+
+func ksmClusterAgentWantFunc(hasCustomConfig bool, opts ...ksmClusterAgentOption) *test.ComponentTest {
+	cfg := ksmClusterAgentWantConfig{}
+	for _, o := range opts {
+		o(&cfg)
+	}
 	return test.NewDefaultComponentTest().WithWantFunc(
 		func(t testing.TB, mgrInterface feature.PodTemplateManagers) {
 			mgr := mgrInterface.(*fake.PodTemplateManagers)
@@ -226,9 +251,10 @@ func ksmClusterAgentWantFunc(hasCustomConfig bool) *test.ComponentTest {
 			} else {
 				// Verify default config annotation - CRDs and APIServices collected, no custom resource metrics
 				defaultConfigData := map[string]any{
-					"collect_crds":        true,
-					"collect_apiservices": true,
-					"collect_cr_metrics":  nil,
+					"collect_crds":           true,
+					"collect_apiservices":    true,
+					"collect_cr_metrics":     nil,
+					"pod_collection_on_node": cfg.podCollectionOnNode,
 				}
 				hash, err := comparison.GenerateMD5ForSpec(defaultConfigData)
 				assert.NoError(t, err)

pkg/testutils/builder.go

@@ -1184,6 +1184,21 @@ func (builder *DatadogAgentBuilder) WithClusterAgentImage(image string) *Datadog
 	return builder
 }
 
+func (builder *DatadogAgentBuilder) WithNodeAgentImage(image string) *DatadogAgentBuilder {
+	if builder.datadogAgent.Spec.Override == nil {
+		builder.datadogAgent.Spec.Override = map[v2alpha1.ComponentName]*v2alpha1.DatadogAgentComponentOverride{}
+	}
+
+	if builder.datadogAgent.Spec.Override[v2alpha1.NodeAgentComponentName] == nil {
+		builder.datadogAgent.Spec.Override[v2alpha1.NodeAgentComponentName] = &v2alpha1.DatadogAgentComponentOverride{}
+	}
+
+	builder.datadogAgent.Spec.Override[v2alpha1.NodeAgentComponentName].Image = &v2alpha1.AgentImageConfig{
+		Name: image,
+	}
+	return builder
+}
+
 func (builder *DatadogAgentBuilder) WithClusterAgentDisabled(disabled bool) *DatadogAgentBuilder {
 	builder.WithComponentOverride(v2alpha1.ClusterAgentComponentName, v2alpha1.DatadogAgentComponentOverride{
 		Disabled: ptr.To(disabled),