address review comments

- normalized the input to lower case for consistency
- added validations for kind collision
- add flag for unresolved kinds, and defer skip decision base on that
- move peek-and-map test cases to restore_policies_test.go

Signed-off-by: Adam Zhang <adam.zhang@broadcom.com>

design/restore-filter-enhancement/fine-grained-restore-filters-design.md

@@ -150,12 +150,14 @@ Only resource kinds listed in `resourceFilters` entries are restored for the mat
 
 The `kinds` field accepts both plural resource names (e.g., `configmaps`, `mycustomkinds.mygroup.io`) and singular `Kind` names (e.g., `ConfigMap`, `MyCustomKind`). 
 
+To ensure consistent case-insensitive behavior across all code paths, Velero normalizes all input `kinds` to lowercase *before* attempting discovery or fallback matching.
+
 During a restore, Velero attempts to resolve `Kind` names to fully-qualified plural resource names using the cluster's discovery helper. However, for Custom Resources (CRDs), the CRD might not exist in the cluster yet when the restore begins. 
 
 To handle this, Velero implements a **peek-and-map fallback**:
-1. If a `Kind` cannot be resolved via the discovery helper at the start of the restore, Velero stores the raw string as provided in the policy.
+1. If a normalized `Kind` cannot be resolved via the discovery helper at the start of the restore, Velero stores the normalized string as provided in the policy.
 2. Later, when iterating through the backup tarball, if Velero encounters a resource type (e.g., `mycustomkinds.mygroup.io`) that doesn't match any resolved filters, it peeks at the `Kind` of the first item in the tarball for that resource type.
-3. It then checks if this actual `Kind` matches any of the unresolved strings in the user's policy (case-insensitive).
+3. It then checks if this actual `Kind` (case-insensitively) matches any of the unresolved normalized strings in the user's policy.
 4. If a match is found, the filter is applied and cached for subsequent lookups. 
 
 This ensures that users can intuitively write `kinds: [MyCustomKind]` and it will work reliably, even if the CRD hasn't been restored yet. This logic applies to both `namespacedFilterPolicies` and `clusterScopedFilterPolicy`.

pkg/controller/restore_controller_test.go

@@ -18,7 +18,6 @@ package controller
 
 import (
 	"bytes"
-	"context"
 	"io"
 	"testing"
 	"time"
@@ -786,7 +785,7 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 		Phase(velerov1api.BackupPhaseCompleted).
 		Result()))
 
-	r.validateAndComplete(context.Background(), restore)
+	r.validateAndComplete(t.Context(), restore)
 	assert.Contains(t, restore.Status.ValidationErrors, "No backups found for schedule")
 	assert.Empty(t, restore.Spec.BackupName)
 
@@ -802,7 +801,7 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 			Result(),
 	))
 
-	r.validateAndComplete(context.Background(), restore)
+	r.validateAndComplete(t.Context(), restore)
 	assert.Contains(t, restore.Status.ValidationErrors, "No completed backups found for schedule")
 	assert.Empty(t, restore.Spec.BackupName)
 
@@ -833,7 +832,7 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 			ScheduleName: "schedule-1",
 		},
 	}
-	r.validateAndComplete(context.Background(), restore)
+	r.validateAndComplete(t.Context(), restore)
 	assert.Nil(t, restore.Status.ValidationErrors)
 	assert.Equal(t, "foo", restore.Spec.BackupName)
 }
@@ -893,7 +892,7 @@ func TestValidateAndCompleteWithResourcePolicySpecified(t *testing.T) {
 			Result(),
 	))
 
-	r.validateAndComplete(context.Background(), restore)
+	r.validateAndComplete(t.Context(), restore)
 	assert.Contains(t, restore.Status.ValidationErrors[0], "fail to get ResourcePolicies velero/test-configmap ConfigMap")
 
 	restore1 := &velerov1api.Restore{
@@ -926,7 +925,7 @@ clusterScopedFilterPolicy:
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), cm1))
 
-	r.validateAndComplete(context.Background(), restore1)
+	r.validateAndComplete(t.Context(), restore1)
 	assert.Nil(t, restore1.Status.ValidationErrors)
 
 	restore2 := &velerov1api.Restore{
@@ -963,7 +962,7 @@ volumePolicies:
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), cm2))
 
-	r.validateAndComplete(context.Background(), restore2)
+	r.validateAndComplete(t.Context(), restore2)
 	assert.Contains(t, restore2.Status.ValidationErrors[0], "fail to validate ResourcePolicies in ConfigMap velero/test-configmap-invalid")
 }
 
@@ -1022,7 +1021,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 			Result(),
 	))
 
-	r.validateAndComplete(context.Background(), restore)
+	r.validateAndComplete(t.Context(), restore)
 	assert.Contains(t, restore.Status.ValidationErrors[0], "failed to get resource modifiers configmap")
 
 	restore1 := &velerov1api.Restore{
@@ -1050,7 +1049,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), cm1))
 
-	r.validateAndComplete(context.Background(), restore1)
+	r.validateAndComplete(t.Context(), restore1)
 	assert.Nil(t, restore1.Status.ValidationErrors)
 
 	restore2 := &velerov1api.Restore{
@@ -1079,7 +1078,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), invalidVersionCm))
 
-	r.validateAndComplete(context.Background(), restore2)
+	r.validateAndComplete(t.Context(), restore2)
 	assert.Contains(t, restore2.Status.ValidationErrors[0], "Error in parsing resource modifiers provided in configmap")
 
 	restore3 := &velerov1api.Restore{
@@ -1107,7 +1106,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), invalidOperatorCm))
 
-	r.validateAndComplete(context.Background(), restore3)
+	r.validateAndComplete(t.Context(), restore3)
 	assert.Contains(t, restore3.Status.ValidationErrors[0], "Validation error in resource modifiers provided in configmap")
 }
 

pkg/restore/restore.go

@@ -456,6 +456,9 @@ type resolvedNamespaceFilter struct {
 	// catchAllFilter holds the resolved filter for a catch-all entry (empty kinds or ["*"]).
 	// nil when no catch-all entry is defined.
 	catchAllFilter *resolvedResourceFilter
+	// hasUnresolvedKinds is true if any kind in the policy failed discovery.
+	// This is used to bypass the fast-path skip so the peek-and-map fallback can run.
+	hasUnresolvedKinds bool
 }
 
 // namespacedFilterPattern pairs a namespace pattern string with its pre-compiled
@@ -514,17 +517,24 @@ func resolveRestoreClusterScopedFilterPolicy(
 		if err != nil {
 			return nil, err
 		}
-		for _, kind := range rf.Kinds {
-			gr, resource, err := helper.ResourceFor(schema.GroupVersionResource{Resource: kind})
+		for _, kind := range resolved.originalKinds {
+			gr, resource, err := helper.ResourceFor(schema.ParseGroupResource(kind).WithVersion(""))
+
+			key := kind
 			if err != nil {
 				log.WithField("kind", kind).Warnf("Cannot resolve kind via discovery, using as-is")
-				result[kind] = resolved
-				continue
+			} else {
+				if resource.Namespaced {
+					log.Warnf("kind %q in clusterScopedFilterPolicy is a namespace-scoped resource; it will never match in a cluster-scoped filter — did you mean namespacedFilterPolicies?", kind)
+				}
+				key = gr.GroupResource().String()
 			}
-			if resource.Namespaced {
-				log.Warnf("kind %q in clusterScopedFilterPolicy is a namespace-scoped resource; it will never match in a cluster-scoped filter — did you mean namespacedFilterPolicies?", kind)
+
+			if _, exists := result[key]; exists {
+				return nil, fmt.Errorf("ambiguous policy: duplicate kind %q detected", key)
 			}
-			result[gr.GroupResource().String()] = resolved
+
+			result[key] = resolved
 		}
 	}
 	return result, nil
@@ -548,6 +558,7 @@ func resolveRestoreNamespacedFilterPolicies(
 	for _, policy := range policies {
 		rfMap := make(map[string]*resolvedResourceFilter)
 		var catchAll *resolvedResourceFilter
+		hasUnresolvedKinds := false
 
 		for _, rf := range policy.ResourceFilters {
 			resolved, err := resolveResourceFilter(rf)
@@ -560,35 +571,42 @@ func resolveRestoreNamespacedFilterPolicies(
 				continue
 			}
 
-			for _, kind := range rf.Kinds {
+			for _, kind := range resolved.originalKinds {
 				gr, resource, err := helper.ResourceFor(
-					schema.GroupVersionResource{Resource: kind},
+					schema.ParseGroupResource(kind).WithVersion(""),
 				)
+
+				key := kind
 				if err != nil {
 					log.WithField("kind", kind).Warnf(
 						"Cannot resolve kind via discovery, using as-is")
-					rfMap[kind] = resolved
-					continue
+					hasUnresolvedKinds = true
+				} else {
+					if !resource.Namespaced {
+						log.Warnf("kind %q in namespacedFilterPolicies is a cluster-scoped resource; it will never match in a namespace-scoped filter — did you mean clusterScopedFilterPolicy?", kind)
+					}
+
+					if globalExcludes[kind] || globalExcludes[gr.GroupResource().String()] {
+						log.WithFields(logrus.Fields{
+							"kind":             kind,
+							"namespacePattern": strings.Join(policy.Namespaces, ","),
+						}).Warn("namespacedFilterPolicies entry lists a kind that is globally excluded by RestoreSpec.ExcludedResources; the per-namespace filter entry has no effect")
+					}
+					key = gr.GroupResource().String()
 				}
 
-				if !resource.Namespaced {
-					log.Warnf("kind %q in namespacedFilterPolicies is a cluster-scoped resource; it will never match in a namespace-scoped filter — did you mean clusterScopedFilterPolicy?", kind)
+				if _, exists := rfMap[key]; exists {
+					return nil, nil, fmt.Errorf("ambiguous policy: duplicate kind %q detected", key)
 				}
 
-				if globalExcludes[kind] || globalExcludes[gr.GroupResource().String()] {
-					log.WithFields(logrus.Fields{
-						"kind":             kind,
-						"namespacePattern": strings.Join(policy.Namespaces, ","),
-					}).Warn("namespacedFilterPolicies entry lists a kind that is globally excluded by RestoreSpec.ExcludedResources; the per-namespace filter entry has no effect")
-				}
-
-				rfMap[gr.GroupResource().String()] = resolved
+				rfMap[key] = resolved
 			}
 		}
 
 		nsFilter := &resolvedNamespaceFilter{
-			resourceFilterMap: rfMap,
-			catchAllFilter:    catchAll,
+			resourceFilterMap:  rfMap,
+			catchAllFilter:     catchAll,
+			hasUnresolvedKinds: hasUnresolvedKinds,
 		}
 		for _, nsPattern := range policy.Namespaces {
 			result[nsPattern] = nsFilter
@@ -634,11 +652,17 @@ func resolveResourceFilter(
 	if len(rf.Names) > 0 || len(rf.ExcludedNames) > 0 {
 		nameIE = collections.NewIncludesExcludes().Includes(rf.Names...).Excludes(rf.ExcludedNames...)
 	}
+
+	normalizedKinds := make([]string, len(rf.Kinds))
+	for i, k := range rf.Kinds {
+		normalizedKinds[i] = strings.ToLower(k)
+	}
+
 	return &resolvedResourceFilter{
 		labelSelector:    selector,
 		orLabelSelectors: orSelectors,
 		nameIE:           nameIE,
-		originalKinds:    rf.Kinds,
+		originalKinds:    normalizedKinds,
 	}, nil
 }
 
@@ -2543,7 +2567,7 @@ func (ctx *restoreContext) getOrderedResourceCollection(
 			if namespace != "" && !ctx.resourceMustHave.Has(groupResource.String()) {
 				if nsFilter := ctx.getNamespaceFilter(namespace); nsFilter != nil {
 					_, kindListed := nsFilter.resourceFilterMap[groupResource.String()]
-					if !kindListed && nsFilter.catchAllFilter == nil {
+					if !kindListed && nsFilter.catchAllFilter == nil && !nsFilter.hasUnresolvedKinds {
 						ctx.log.Infof("Skipping resource %s in namespace %s: not in resourceFilters",
 							resource, namespace)
 						continue
@@ -2643,6 +2667,11 @@ func (ctx *restoreContext) getSelectedRestoreableItems(resource string, original
 					rf = nsFilter.catchAllFilter // may be nil if no catch-all
 				}
 				useFilterPolicy = true
+
+				if rf == nil {
+					ctx.log.Infof("Skipping resource %s in namespace %s: not in resourceFilters", resource, originalNamespace)
+					return restorable, warnings, errs
+				}
 			}
 		} else if ctx.clusterScopedFilterMap != nil {
 			// Cluster-scoped path: only applies if kind is listed (refinement overlay)
@@ -2650,7 +2679,10 @@ func (ctx *restoreContext) getSelectedRestoreableItems(resource string, original
 				rf = listedRF
 				useFilterPolicy = true
 			} else if len(items) > 0 {
-				// Peek-and-map logic for unresolvable kinds
+				// Peek-and-map logic for unresolvable kinds.
+				// Note: Unlike the namespaced path, this fallback is always reachable
+				// because the main restore loop does not have a fast-path skip for
+				// unlisted cluster-scoped resources.
 				peekPath := archive.GetItemFilePath(ctx.restoreDir, resourceForPath, originalNamespace, items[0])
 				// Ignore unmarshal errors during peek; the main restore loop will catch and report them
 				if obj, err := archive.Unmarshal(ctx.fileSystem, peekPath); err == nil {

pkg/restore/restore_policies_test.go

@@ -1,7 +1,6 @@
 package restore
 
 import (
-	"context"
 	"io"
 	"testing"
 
@@ -9,6 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
@@ -19,6 +19,9 @@ import (
 )
 
 func TestRestoreResourcePoliciesFiltering(t *testing.T) {
+	customKindRes := &test.APIResource{Group: "mygroup.io", Version: "v1", Name: "mycustomkinds", Kind: "MyCustomKind", Namespaced: true}
+	clusterCustomKindRes := &test.APIResource{Group: "mygroup.io", Version: "v1", Name: "myclustercustomkinds", Kind: "MyClusterCustomKind", Namespaced: false}
+
 	tests := []struct {
 		name         string
 		restore      *velerov1api.Restore
@@ -150,6 +153,45 @@ namespacedFilterPolicies:
 				test.Deployments(): {"ns-1/deploy-1"},
 			},
 		},
+		{
+			name:    "unresolved kind in namespaced filter policy is still restored via peek-and-map",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			policyYAML: `version: v1
+namespacedFilterPolicies:
+  - namespaces: ["ns-1"]
+    resourceFilters:
+      - kinds: ["MyCustomKind"]
+`,
+			tarball: test.NewTarWriter(t).AddItems("mycustomkinds.mygroup.io",
+				&unstructured.Unstructured{Object: map[string]any{"apiVersion": "mygroup.io/v1", "kind": "MyCustomKind", "metadata": map[string]any{"namespace": "ns-1", "name": "my-cr"}}},
+			).Done(),
+			apiResources: []*test.APIResource{
+				customKindRes,
+			},
+			want: map[*test.APIResource][]string{
+				customKindRes: {"ns-1/my-cr"},
+			},
+		},
+		{
+			name:    "unresolved kind in cluster-scoped filter policy is still restored via peek-and-map",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			policyYAML: `version: v1
+clusterScopedFilterPolicy:
+  resourceFilters:
+    - kinds: ["MyClusterCustomKind"]
+`,
+			tarball: test.NewTarWriter(t).AddItems("myclustercustomkinds.mygroup.io",
+				&unstructured.Unstructured{Object: map[string]any{"apiVersion": "mygroup.io/v1", "kind": "MyClusterCustomKind", "metadata": map[string]any{"name": "my-cluster-cr"}}},
+			).Done(),
+			apiResources: []*test.APIResource{
+				clusterCustomKindRes,
+			},
+			want: map[*test.APIResource][]string{
+				clusterCustomKindRes: {"/my-cluster-cr"},
+			},
+		},
 	}
 
 	for _, tc := range tests {
@@ -180,7 +222,7 @@ namespacedFilterPolicies:
 					Name: "test-policies",
 				}
 				var err error
-				resPolicies, err = resourcepolicies.GetResourcePoliciesFromRestore(context.Background(), restore, client, logrus.New())
+				resPolicies, err = resourcepolicies.GetResourcePoliciesFromRestore(t.Context(), restore, client, logrus.New())
 				require.NoError(t, err)
 			}
 
@@ -204,3 +246,40 @@ namespacedFilterPolicies:
 		})
 	}
 }
+
+func TestResolveRestoreNamespacedFilterPolicies_Validation(t *testing.T) {
+	log := logrus.New()
+	helper := test.NewFakeDiscoveryHelper(true, nil)
+
+	policies := []resourcepolicies.NamespacedFilterPolicy{
+		{
+			Namespaces: []string{"ns-1"},
+			ResourceFilters: []resourcepolicies.ResourceFilter{
+				{
+					Kinds: []string{"MyKind", "mykind"},
+				},
+			},
+		},
+	}
+
+	_, _, err := resolveRestoreNamespacedFilterPolicies(policies, nil, helper, log)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "ambiguous policy: duplicate kind")
+}
+
+func TestResolveRestoreClusterScopedFilterPolicy_Validation(t *testing.T) {
+	log := logrus.New()
+	helper := test.NewFakeDiscoveryHelper(true, nil)
+
+	policy := &resourcepolicies.ClusterScopedFilterPolicy{
+		ResourceFilters: []resourcepolicies.ResourceFilter{
+			{
+				Kinds: []string{"MyKind", "mykind"},
+			},
+		},
+	}
+
+	_, err := resolveRestoreClusterScopedFilterPolicy(policy, helper, log)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "ambiguous policy: duplicate kind")
+}

pkg/restore/restore_test.go

@@ -754,26 +754,6 @@ func TestRestoreResourceFiltering(t *testing.T) {
 			apiResources: []*test.APIResource{test.ServiceAccounts()},
 			want:         map[*test.APIResource][]string{test.ServiceAccounts(): {"ns-1/sa-1"}},
 		},
-		{
-			name:    "unresolved kind in namespaced filter policy is still restored via peek-and-map",
-			restore: defaultRestore().ResourcePoliciesConfigmap("test-policy").Result(),
-			backup:  defaultBackup().Result(),
-			tarball: test.NewTarWriter(t).AddItems("mycustomkinds.mygroup.io",
-				&unstructured.Unstructured{Object: map[string]any{"apiVersion": "mygroup.io/v1", "kind": "MyCustomKind", "metadata": map[string]any{"namespace": "ns-1", "name": "my-cr"}}},
-			).Done(),
-			apiResources: []*test.APIResource{},            // Empty to simulate discovery failure
-			want:         map[*test.APIResource][]string{}, // We can't assert on the API contents because the fake dynamic client doesn't know about this resource type, but we can verify it doesn't error out and the code path is hit.
-		},
-		{
-			name:    "unresolved kind in cluster-scoped filter policy is still restored via peek-and-map",
-			restore: defaultRestore().ResourcePoliciesConfigmap("test-policy").Result(),
-			backup:  defaultBackup().Result(),
-			tarball: test.NewTarWriter(t).AddItems("myclustercustomkinds.mygroup.io",
-				&unstructured.Unstructured{Object: map[string]any{"apiVersion": "mygroup.io/v1", "kind": "MyClusterCustomKind", "metadata": map[string]any{"name": "my-cluster-cr"}}},
-			).Done(),
-			apiResources: []*test.APIResource{},            // Empty to simulate discovery failure
-			want:         map[*test.APIResource][]string{}, // Same here
-		},
 	}
 
 	for _, tc := range tests {
@@ -785,34 +765,8 @@ func TestRestoreResourceFiltering(t *testing.T) {
 			}
 			require.NoError(t, h.restorer.discoveryHelper.Refresh())
 
-			if tc.restore.Spec.ResourcePolicy != nil {
-				var yamlData string
-				if tc.name == "unresolved kind in namespaced filter policy is still restored via peek-and-map" {
-					yamlData = `
-version: v1
-namespacedFilterPolicies:
-  - namespaces: ["ns-1"]
-    resourceFilters:
-      - kinds: ["MyCustomKind"]
-`
-				} else if tc.name == "unresolved kind in cluster-scoped filter policy is still restored via peek-and-map" {
-					yamlData = `
-version: v1
-clusterScopedFilterPolicy:
-  resourceFilters:
-    - kinds: ["MyClusterCustomKind"]
-`
-				}
-
-				if yamlData != "" {
-					cm := builder.ForConfigMap(tc.restore.Namespace, tc.restore.Spec.ResourcePolicy.Name).Data("yaml", yamlData).Result()
-					err := h.restorer.kbClient.Create(context.TODO(), cm)
-					require.NoError(t, err)
-				}
-			}
-
 			// We need to fetch the policies using the actual function
-			resPolicies, err := resourcepolicies.GetResourcePoliciesFromRestore(context.TODO(), tc.restore, h.restorer.kbClient, h.log)
+			resPolicies, err := resourcepolicies.GetResourcePoliciesFromRestore(t.Context(), tc.restore, h.restorer.kbClient, h.log)
 			require.NoError(t, err)
 
 			data := &Request{

pkg/test/api_server.go

@@ -56,6 +56,8 @@ func NewAPIServer(t *testing.T) *APIServer {
 				{Group: "extensions", Version: "v1", Resource: "deployments"}:                              "ExtDeploymentsList",
 				{Group: "velero.io", Version: "v1", Resource: "deployments"}:                               "VeleroDeploymentsList",
 				{Group: "velero.io", Version: "v2alpha1", Resource: "datauploads"}:                         "DataUploadsList",
+				{Group: "mygroup.io", Version: "v1", Resource: "mycustomkinds"}:                            "MyCustomKindList",
+				{Group: "mygroup.io", Version: "v1", Resource: "myclustercustomkinds"}:                     "MyClusterCustomKindList",
 			})
 		discoveryClient = &DiscoveryClient{FakeDiscovery: kubeClient.Discovery().(*discoveryfake.FakeDiscovery)}
 	)