[pkg/clusteragent/autoscaling] Filter template labels before applying nodepool (#49938)

### What does this PR do?

Filter template metadata before creating the NodePool. Any labels/annotations containing keys with `kubernetes.io`/`k8s.io`/`karpenter.sh` are dropped with a warning log

### Motivation

Ensure we don't attempt to create a NodePool with restricted template metadata labels/annotations - this ensures NodePools we create are valid and can be applied 

### Describe how you validated your changes
Unit tests cover this small change

### Additional Notes


Co-authored-by: jennifer.chen <jennifer.chen@datadoghq.com>

Author: jennchenn

pkg/clusteragent/autoscaling/cluster/model/node_pool_internal.go

@@ -8,6 +8,8 @@
 package model
 
 import (
+	"strings"
+
 	// AWS Karpenter provider registers some variables in shared packages
 	_ "github.com/aws/karpenter-provider-aws/pkg/apis/v1"
 
@@ -26,6 +28,41 @@ const (
 	KarpenterNodePoolHashAnnotationKey = "karpenter.sh/nodepool-hash"
 )
 
+// templateMetadataBlocklistedDomains lists base domains blocked from template_metadata
+// labels and annotations.
+var templateMetadataBlocklistedDomains = []string{
+	"kubernetes.io",
+	"k8s.io",
+	"karpenter.sh",
+}
+
+// filterTemplateMetadataKeys filters out reserved keys and returns a map that can be applied
+// to the nodepool template labels or annotations.
+func filterTemplateMetadataKeys(kvs []KeyValue) map[string]string {
+	out := make(map[string]string, len(kvs))
+	for _, kv := range kvs {
+		if isBlocklistedTemplateMetadataKey(kv.Key) {
+			log.Warnf("Dropping template_metadata key %q", kv.Key)
+			continue
+		}
+		out[kv.Key] = kv.Value
+	}
+	return out
+}
+
+func isBlocklistedTemplateMetadataKey(key string) bool {
+	domain := karpenterv1.GetLabelDomain(key)
+	if domain == "" {
+		return false
+	}
+	for _, blocked := range templateMetadataBlocklistedDomains {
+		if domain == blocked || strings.HasSuffix(domain, "."+blocked) {
+			return true
+		}
+	}
+	return false
+}
+
 type NodePoolInternal struct {
 	// targetName is the user-created NodePool the Datadog-managed NodePool is derived from
 	targetName string
@@ -66,17 +103,9 @@ func buildKarpenterNodePoolFromManifest(kv1 *KarpenterV1NodePool) *karpenterv1.N
 	// Handle template metadata labels/annotations
 	spec := *kv1.Spec
 	if kv1.TemplateMetadata != nil {
-		templateLabels := make(map[string]string, len(kv1.TemplateMetadata.Labels))
-		for _, kv := range kv1.TemplateMetadata.Labels {
-			templateLabels[kv.Key] = kv.Value
-		}
-		templateAnnotations := make(map[string]string, len(kv1.TemplateMetadata.Annotations))
-		for _, kv := range kv1.TemplateMetadata.Annotations {
-			templateAnnotations[kv.Key] = kv.Value
-		}
 		spec.Template.ObjectMeta = karpenterv1.ObjectMeta{
-			Labels:      templateLabels,
-			Annotations: templateAnnotations,
+			Labels:      filterTemplateMetadataKeys(kv1.TemplateMetadata.Labels),
+			Annotations: filterTemplateMetadataKeys(kv1.TemplateMetadata.Annotations),
 		}
 	}
 

pkg/clusteragent/autoscaling/cluster/model/node_pool_internal_test.go

@@ -144,6 +144,88 @@ func TestBuildKarpenterNodePoolFromManifest_WithTemplateMetadata(t *testing.T) {
 	assert.Equal(t, map[string]string{"node-ann": "node-ann-val"}, knp.Spec.Template.ObjectMeta.Annotations)
 }
 
+func TestBuildKarpenterNodePoolFromManifest_TemplateMetadataBlocklist(t *testing.T) {
+	kv1 := &KarpenterV1NodePool{
+		Metadata: Metadata{Name: "test-pool"},
+		TemplateMetadata: &Metadata{
+			Labels: Labels{
+				{Key: "app", Value: "my-app"},
+				{Key: "karpenter.sh/capacity-type", Value: "spot"},
+				{Key: "node-role.kubernetes.io/control-plane", Value: ""},
+				{Key: "k8s.io/foo", Value: "bar"},
+			},
+			Annotations: Annotations{
+				{Key: "team", Value: "infra"},
+				{Key: "kubernetes.io/created-by", Value: "value"},
+				{Key: "karpenter.sh/do-not-disrupt", Value: "true"},
+			},
+		},
+		Spec: &karpenterv1.NodePoolSpec{},
+	}
+
+	knp := buildKarpenterNodePoolFromManifest(kv1)
+
+	require.NotNil(t, knp)
+	assert.Equal(t, map[string]string{"app": "my-app"}, knp.Spec.Template.ObjectMeta.Labels)
+	assert.Equal(t, map[string]string{"team": "infra"}, knp.Spec.Template.ObjectMeta.Annotations)
+}
+
+func TestFilterTemplateMetadataKeys(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    []KeyValue
+		expected map[string]string
+	}{
+		{
+			name:     "empty input",
+			input:    nil,
+			expected: map[string]string{},
+		},
+		{
+			name: "all allowed",
+			input: []KeyValue{
+				{Key: "app", Value: "web"},
+				{Key: "team", Value: "platform"},
+			},
+			expected: map[string]string{"app": "web", "team": "platform"},
+		},
+		{
+			name: "keys that contain blocked strings but are not in blocked domains are allowed",
+			input: []KeyValue{
+				{Key: "teamk8s.io/role", Value: "worker"},
+				{Key: "example.com/karpenter.sh-mode", Value: "fast"},
+			},
+			expected: map[string]string{"teamk8s.io/role": "worker", "example.com/karpenter.sh-mode": "fast"},
+		},
+		{
+			name: "reserved labels are dropped",
+			input: []KeyValue{
+				{Key: "app", Value: "web"},
+				{Key: "node-role.kubernetes.io/control-plane", Value: ""},
+				{Key: "k8s.io/foo", Value: "bar"},
+				{Key: "karpenter.sh/capacity-type", Value: "spot"},
+				{Key: "karpenter.sh/injected", Value: "bad"},
+			},
+			expected: map[string]string{"app": "web"},
+		},
+		{
+			name: "reserved annotations are dropped",
+			input: []KeyValue{
+				{Key: "team", Value: "infra"},
+				{Key: "karpenter.sh/do-not-disrupt", Value: "true"},
+				{Key: "kubernetes.io/created-by", Value: "value"},
+			},
+			expected: map[string]string{"team": "infra"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.expected, filterTemplateMetadataKeys(tt.input))
+		})
+	}
+}
+
 func TestBuildKarpenterNodePoolFromManifest_NilSpec(t *testing.T) {
 	kv1 := &KarpenterV1NodePool{
 		Metadata: Metadata{Name: "test-pool"},