[CONTP-1596]Improve GKE Autopilot support

config/rbac/role.yaml

@@ -188,6 +188,7 @@ rules:
   - create
   - get
   - list
+  - patch
   - watch
 - apiGroups:
   - autoscaling

internal/controller/datadogagent/common/volumes.go

@@ -295,6 +295,7 @@ func GetVolumeMountForSecurity() corev1.VolumeMount {
 	return corev1.VolumeMount{
 		Name:      SeccompSecurityVolumeName,
 		MountPath: SeccompSecurityVolumePath,
+		ReadOnly:  true,
 	}
 }
 

internal/controller/datadogagent/component/agent/rbac.go

@@ -14,22 +14,36 @@ import (
 
 // RBAC for Agent
 
-// GetDefaultAgentClusterRolePolicyRules returns the default policy rules for the Agent cluster role
-func GetDefaultAgentClusterRolePolicyRules(excludeNonResourceRules bool, useFineGrainedAuthorization bool) []rbacv1.PolicyRule {
+// GetDefaultAgentClusterRolePolicyRules returns the default policy rules for the Agent cluster role.
+// kubeletUseAPIServer adds get/list on pods so the Agent can discover pods via
+// the API server when the kubelet endpoint is not reachable (e.g. GKE Autopilot).
+func GetDefaultAgentClusterRolePolicyRules(excludeNonResourceRules bool, useFineGrainedAuthorization bool, kubeletUseAPIServer bool) []rbacv1.PolicyRule {
 	policyRule := []rbacv1.PolicyRule{
 		getKubeletPolicyRule(useFineGrainedAuthorization),
 		getEndpointsPolicyRule(),
 		getLeaderElectionPolicyRule(),
 		component.GetEKSControlPlaneMetricsPolicyRule(),
 	}
 
+	if kubeletUseAPIServer {
+		policyRule = append(policyRule, getPodsPolicyRule())
+	}
+
 	if !excludeNonResourceRules {
 		policyRule = append(policyRule, getMetricsEndpointPolicyRule())
 	}
 
 	return policyRule
 }
 
+func getPodsPolicyRule() rbacv1.PolicyRule {
+	return rbacv1.PolicyRule{
+		APIGroups: []string{rbac.CoreAPIGroup},
+		Resources: []string{rbac.PodsResource},
+		Verbs:     []string{rbac.GetVerb, rbac.ListVerb},
+	}
+}
+
 func getMetricsEndpointPolicyRule() rbacv1.PolicyRule {
 	return rbacv1.PolicyRule{
 		NonResourceURLs: []string{

internal/controller/datadogagent/component/agent/rbac_test.go

@@ -0,0 +1,45 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-present Datadog, Inc.
+
+package agent
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/DataDog/datadog-operator/pkg/kubernetes/rbac"
+)
+
+func TestGetDefaultAgentClusterRolePolicyRules_KubeletUseAPIServer(t *testing.T) {
+	tests := []struct {
+		name                string
+		kubeletUseAPIServer bool
+		expectPodsRule      bool
+	}{
+		{name: "no kubelet api server -> no pods rule", kubeletUseAPIServer: false, expectPodsRule: false},
+		{name: "kubelet api server -> pods get/list rule", kubeletUseAPIServer: true, expectPodsRule: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rules := GetDefaultAgentClusterRolePolicyRules(false, false, tt.kubeletUseAPIServer)
+
+			found := false
+			for _, r := range rules {
+				for _, res := range r.Resources {
+					if res != rbac.PodsResource {
+						continue
+					}
+					found = true
+					assert.ElementsMatch(t, []string{rbac.GetVerb, rbac.ListVerb}, r.Verbs,
+						"pods rule should grant get and list")
+					assert.Equal(t, []string{rbac.CoreAPIGroup}, r.APIGroups)
+				}
+			}
+			assert.Equal(t, tt.expectPodsRule, found)
+		})
+	}
+}

internal/controller/datadogagent/controller_v2_test.go

@@ -1248,6 +1248,9 @@ func Test_AutopilotOverrides(t *testing.T) {
 				ds := &appsv1.DaemonSet{}
 				err := c.Get(context.TODO(), types.NamespacedName{Namespace: resourcesNamespace, Name: dsName}, ds)
 				assert.NoError(t, err, "Failed to get DaemonSet %s/%s", resourcesNamespace, dsName)
+				assertNoDanglingVolumeMounts(t, ds.Spec.Template.Spec)
+				assertNoEnvVarInPodSpec(t, ds.Spec.Template.Spec, common.DDAuthTokenFilePath)
+				assertSeccompSecurityMountReadOnly(t, ds.Spec.Template.Spec)
 
 				forbiddenVolumes := map[string]struct{}{
 					common.AuthVolumeName:            {},
@@ -1334,6 +1337,9 @@ func Test_AutopilotOverrides(t *testing.T) {
 				ds := &appsv1.DaemonSet{}
 				err := c.Get(context.TODO(), types.NamespacedName{Namespace: resourcesNamespace, Name: dsName}, ds)
 				assert.NoError(t, err, "Failed to get DaemonSet %s/%s", resourcesNamespace, dsName)
+				assertNoDanglingVolumeMounts(t, ds.Spec.Template.Spec)
+				assertNoEnvVarInPodSpec(t, ds.Spec.Template.Spec, common.DDAuthTokenFilePath)
+				assertSeccompSecurityMountReadOnly(t, ds.Spec.Template.Spec)
 
 				traceAgentFound := false
 				for _, ctn := range ds.Spec.Template.Spec.Containers {
@@ -1389,6 +1395,59 @@ func verifyDaemonsetContainers(t *testing.T, c client.Client, resourcesNamespace
 	assert.Equal(t, expectedContainers, dsContainers, "Container names don't match")
 }
 
+func assertNoDanglingVolumeMounts(t *testing.T, podSpec corev1.PodSpec) {
+	t.Helper()
+
+	volumes := map[string]struct{}{}
+	for _, volume := range podSpec.Volumes {
+		volumes[volume.Name] = struct{}{}
+	}
+
+	for _, container := range podSpec.InitContainers {
+		for _, mount := range container.VolumeMounts {
+			_, found := volumes[mount.Name]
+			assert.True(t, found, "init container %s has mount %s without a matching volume", container.Name, mount.Name)
+		}
+	}
+	for _, container := range podSpec.Containers {
+		for _, mount := range container.VolumeMounts {
+			_, found := volumes[mount.Name]
+			assert.True(t, found, "container %s has mount %s without a matching volume", container.Name, mount.Name)
+		}
+	}
+}
+
+func assertNoEnvVarInPodSpec(t *testing.T, podSpec corev1.PodSpec, name string) {
+	t.Helper()
+
+	for _, container := range podSpec.InitContainers {
+		for _, env := range container.Env {
+			assert.NotEqual(t, name, env.Name, "init container %s should not have env var %s", container.Name, name)
+		}
+	}
+	for _, container := range podSpec.Containers {
+		for _, env := range container.Env {
+			assert.NotEqual(t, name, env.Name, "container %s should not have env var %s", container.Name, name)
+		}
+	}
+}
+
+func assertSeccompSecurityMountReadOnly(t *testing.T, podSpec corev1.PodSpec) {
+	t.Helper()
+
+	for _, container := range podSpec.InitContainers {
+		if container.Name != string(apicommon.SeccompSetupContainerName) {
+			continue
+		}
+		for _, mount := range container.VolumeMounts {
+			if mount.Name == common.SeccompSecurityVolumeName {
+				assert.True(t, mount.ReadOnly, "init container %s should mount %s read-only", container.Name, mount.Name)
+				return
+			}
+		}
+	}
+}
+
 func verifyPDB(t *testing.T, c client.Client) {
 	pdbList := policyv1.PodDisruptionBudgetList{}
 	err := c.List(context.TODO(), &pdbList)