implement fine-grained backup filter policies

implement backup logic to support backups with
NamespacedFilterPolicies and ClusterScopedFilterPolicy

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

changelogs/unreleased/9880-adam-jian-zhang

@@ -0,0 +1 @@
+Fix issue #9815, implement core logic of backup with ClusterScopedFilterPolicy and NamespacedFilterPolicies

pkg/backup/backup.go

@@ -26,9 +26,11 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/gobwas/glob"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
@@ -36,6 +38,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	kubeerrs "k8s.io/apimachinery/pkg/util/errors"
@@ -43,6 +46,7 @@ import (
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/vmware-tanzu/velero/internal/hook"
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/volume"
 	"github.com/vmware-tanzu/velero/internal/volumehelper"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
@@ -352,6 +356,52 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		backupRequest.ResourceIncludesExcludes = srie
 	}
 
+	if backupRequest.ResPolicies != nil {
+		clusterScopedFilterPolicy := backupRequest.ResPolicies.GetClusterScopedFilterPolicy()
+		if clusterScopedFilterPolicy != nil {
+			backupRequest.ClusterScopedFilterMap, err = resolveClusterScopedFilterPolicy(
+				clusterScopedFilterPolicy,
+				kb.discoveryHelper,
+				log,
+			)
+			if err != nil {
+				return err
+			}
+			log.Infof("Resolved clusterScopedFilterPolicy: %d kind group(s) in cluster-scoped filter map",
+				len(backupRequest.ClusterScopedFilterMap))
+		}
+
+		nfPolicies := backupRequest.ResPolicies.GetNamespacedFilterPolicies()
+		if len(nfPolicies) > 0 {
+			backupRequest.NamespacedFilterMap, backupRequest.NamespacedFilterPatterns, err = resolveNamespacedFilterPolicies(
+				nfPolicies,
+				kb.discoveryHelper,
+				log,
+			)
+			if err != nil {
+				return err
+			}
+			log.Infof("Resolved namespacedFilterPolicies: %d namespace pattern(s) registered",
+				len(backupRequest.NamespacedFilterPatterns))
+			for _, p := range backupRequest.NamespacedFilterPatterns {
+				nsf := backupRequest.NamespacedFilterMap[p.Pattern]
+				log.WithFields(logrus.Fields{
+					"namespacePattern": p.Pattern,
+					"kindCount":        len(nsf.ResourceFilterMap),
+					"hasCatchAll":      nsf.CatchAllFilter != nil,
+				}).Debug("namespacedFilterPolicies: namespace pattern registered")
+				for kind := range nsf.ResourceFilterMap {
+					if backupRequest.ResourceIncludesExcludes.ShouldExclude(kind) {
+						log.WithFields(logrus.Fields{
+							"namespacePattern": p.Pattern,
+							"kind":             kind,
+						}).Warn("namespacedFilterPolicies entry lists a kind that is globally excluded by includeExcludePolicy; the per-namespace filter entry has no effect")
+					}
+				}
+			}
+		}
+	}
+
 	log.Infof("Backing up all volumes using pod volume backup: %t", boolptr.IsSetToTrue(backupRequest.Backup.Spec.DefaultVolumesToFsBackup))
 
 	backupRequest.ResourceHooks, err = getResourceHooks(backupRequest.Spec.Hooks.Resources, kb.discoveryHelper)
@@ -1341,3 +1391,141 @@ func putVolumeInfos(
 
 	return backupStore.PutBackupVolumeInfos(backupName, backupVolumeInfoBuf)
 }
+
+func resolveClusterScopedFilterPolicy(
+	policy *resourcepolicies.ClusterScopedFilterPolicy,
+	helper discovery.Helper,
+	log logrus.FieldLogger,
+) (map[string]*ResolvedResourceFilter, error) {
+	rfMap := make(map[string]*ResolvedResourceFilter)
+
+	for _, rf := range policy.ResourceFilters {
+		resolved, err := resolveResourceFilter(rf)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, kind := range rf.Kinds {
+			gr, apiResource, err := helper.ResourceFor(
+				schema.GroupVersionResource{Resource: kind},
+			)
+			if err != nil {
+				log.WithField("kind", kind).Warnf(
+					"Cannot resolve kind via discovery, using as-is: %v", err)
+				rfMap[kind] = resolved
+				continue
+			}
+			if apiResource.Namespaced {
+				log.WithField("kind", kind).Warnf(
+					"kind %q in clusterScopedFilterPolicy is namespace-scoped; "+
+						"it will never match in a cluster-scoped filter — did you mean namespacedFilterPolicies?", kind)
+			}
+			rfMap[gr.GroupResource().String()] = resolved
+		}
+	}
+
+	return rfMap, nil
+}
+
+func resolveResourceFilter(rf resourcepolicies.ResourceFilter) (*ResolvedResourceFilter, error) {
+	var selector labels.Selector
+	if len(rf.LabelSelector) > 0 {
+		var err error
+		selector, err = labels.ValidatedSelectorFromSet(labels.Set(rf.LabelSelector))
+		if err != nil {
+			return nil, fmt.Errorf("invalid label selector in resource filter: %w", err)
+		}
+	}
+
+	var orSelectors []labels.Selector
+	for _, ols := range rf.OrLabelSelectors {
+		s, err := labels.ValidatedSelectorFromSet(labels.Set(ols))
+		if err != nil {
+			return nil, fmt.Errorf("invalid OR label selector in resource filter: %w", err)
+		}
+		orSelectors = append(orSelectors, s)
+	}
+
+	var nameIE *collections.IncludesExcludes
+	if len(rf.Names) > 0 || len(rf.ExcludedNames) > 0 {
+		nameIE = collections.NewIncludesExcludes()
+		nameIE.Includes(rf.Names...)
+		nameIE.Excludes(rf.ExcludedNames...)
+	}
+
+	return &ResolvedResourceFilter{
+		LabelSelector:    selector,
+		OrLabelSelectors: orSelectors,
+		NameIE:           nameIE,
+	}, nil
+}
+
+func resolveNamespacedFilterPolicies(
+	policies []resourcepolicies.NamespacedFilterPolicy,
+	helper discovery.Helper,
+	log logrus.FieldLogger,
+) (map[string]*ResolvedNamespaceFilter, []NamespacedFilterPattern, error) {
+	result := make(map[string]*ResolvedNamespaceFilter)
+	var patternOrder []NamespacedFilterPattern
+
+	for _, policy := range policies {
+		rfMap := make(map[string]*ResolvedResourceFilter)
+		var nsFilter *ResolvedNamespaceFilter
+
+		for _, rf := range policy.ResourceFilters {
+			resolved, err := resolveResourceFilter(rf)
+			if err != nil {
+				return nil, nil, err
+			}
+
+			if rf.IsCatchAll() {
+				if nsFilter == nil {
+					nsFilter = &ResolvedNamespaceFilter{ResourceFilterMap: rfMap}
+				}
+				nsFilter.CatchAllFilter = resolved
+			} else {
+				// Resolve each kind to a fully-qualified group-resource string with improved error handling
+				for _, kind := range rf.Kinds {
+					gr, apiResource, err := helper.ResourceFor(
+						schema.GroupVersionResource{Resource: kind},
+					)
+					if err != nil {
+						// Log warning but continue - allows for forward compatibility
+						log.WithField("kind", kind).Warnf(
+							"Cannot resolve kind via discovery, using as-is: %v", err)
+						rfMap[kind] = resolved
+						continue
+					}
+					if !apiResource.Namespaced {
+						log.WithField("kind", kind).Warnf(
+							"kind %q in namespacedFilterPolicies is cluster-scoped; "+
+								"it will never match in a namespace-scoped filter — did you mean clusterScopedFilterPolicy?", kind)
+					}
+					rfMap[gr.GroupResource().String()] = resolved
+				}
+			}
+		}
+
+		if nsFilter == nil {
+			nsFilter = &ResolvedNamespaceFilter{ResourceFilterMap: rfMap}
+		} else {
+			nsFilter.ResourceFilterMap = rfMap
+		}
+		for _, nsPattern := range policy.Namespaces {
+			result[nsPattern] = nsFilter
+			// Pre-compile glob patterns once here; exact names are matched via map
+			// and never reach the pattern loop, so only wildcard patterns need Compiled set.
+			entry := NamespacedFilterPattern{Pattern: nsPattern}
+			if strings.ContainsAny(nsPattern, "*?[") {
+				if compiled, cerr := glob.Compile(nsPattern); cerr == nil {
+					entry.Compiled = compiled
+				} else {
+					// Pattern already validated; this branch should not be reached
+					log.WithField("pattern", nsPattern).Warnf("Failed to pre-compile glob pattern: %v", cerr)
+				}
+			}
+			patternOrder = append(patternOrder, entry)
+		}
+	}
+	return result, patternOrder, nil
+}

pkg/backup/backup_test.go

@@ -39,7 +39,9 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/volume"
@@ -5728,3 +5730,321 @@ func (f *fakeSingleObjectBackupStoreGetter) Get(*velerov1.BackupStorageLocation,
 func NewFakeSingleObjectBackupStoreGetter(store persistence.BackupStore) persistence.ObjectBackupStoreGetter {
 	return &fakeSingleObjectBackupStoreGetter{store: store}
 }
+func TestResolveResourceFilter(t *testing.T) {
+	tests := []struct {
+		name        string
+		rf          resourcepolicies.ResourceFilter
+		expectErr   bool
+		checkResult func(*testing.T, *ResolvedResourceFilter)
+	}{
+		{
+			name: "valid label selector",
+			rf: resourcepolicies.ResourceFilter{
+				LabelSelector: map[string]string{"app": "foo"},
+			},
+			expectErr: false,
+			checkResult: func(t *testing.T, r *ResolvedResourceFilter) {
+				t.Helper()
+				require.NotNil(t, r)
+				require.NotNil(t, r.LabelSelector)
+				assert.True(t, r.LabelSelector.Matches(labels.Set{"app": "foo"}))
+			},
+		},
+		{
+			name: "invalid label selector",
+			rf: resourcepolicies.ResourceFilter{
+				LabelSelector: map[string]string{"invalid/label/key": "value"},
+			},
+			expectErr: true,
+		},
+		{
+			name: "valid or label selectors",
+			rf: resourcepolicies.ResourceFilter{
+				OrLabelSelectors: []map[string]string{
+					{"app": "foo"},
+					{"app": "bar"},
+				},
+			},
+			expectErr: false,
+			checkResult: func(t *testing.T, r *ResolvedResourceFilter) {
+				t.Helper()
+				require.NotNil(t, r)
+				require.Len(t, r.OrLabelSelectors, 2)
+			},
+		},
+		{
+			name: "invalid or label selectors",
+			rf: resourcepolicies.ResourceFilter{
+				OrLabelSelectors: []map[string]string{
+					{"invalid/label/key": "value"},
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "names and excluded names",
+			rf: resourcepolicies.ResourceFilter{
+				Names:         []string{"inc1", "inc2"},
+				ExcludedNames: []string{"exc1"},
+			},
+			expectErr: false,
+			checkResult: func(t *testing.T, r *ResolvedResourceFilter) {
+				t.Helper()
+				require.NotNil(t, r)
+				require.NotNil(t, r.NameIE)
+				assert.True(t, r.NameIE.ShouldInclude("inc1"))
+				assert.False(t, r.NameIE.ShouldInclude("exc1"))
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			res, err := resolveResourceFilter(tc.rf)
+			if tc.expectErr {
+				require.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+				if tc.checkResult != nil {
+					tc.checkResult(t, res)
+				}
+			}
+		})
+	}
+}
+
+type mockDiscoveryHelper struct {
+	discovery.Helper
+	ResourceForFunc func(input schema.GroupVersionResource) (schema.GroupVersionResource, metav1.APIResource, error)
+}
+
+func (m *mockDiscoveryHelper) ResourceFor(input schema.GroupVersionResource) (schema.GroupVersionResource, metav1.APIResource, error) {
+	if m.ResourceForFunc != nil {
+		return m.ResourceForFunc(input)
+	}
+	return m.Helper.ResourceFor(input)
+}
+
+func TestResolveClusterScopedFilterPolicy(t *testing.T) {
+	helper := test.NewFakeDiscoveryHelper(true, nil)
+	log := test.NewLogger()
+
+	policy := &resourcepolicies.ClusterScopedFilterPolicy{
+		ResourceFilters: []resourcepolicies.ResourceFilter{
+			{
+				Kinds:         []string{"pods", "secrets"},
+				LabelSelector: map[string]string{"app": "foo"},
+			},
+			{
+				Kinds:         []string{"invalid-kind"},
+				LabelSelector: map[string]string{"invalid/label/key": "value"},
+			},
+		},
+	}
+
+	// Test with invalid label selector to trigger error
+	_, err := resolveClusterScopedFilterPolicy(policy, helper, log)
+	require.Error(t, err)
+
+	// Test valid policy
+	validPolicy := &resourcepolicies.ClusterScopedFilterPolicy{
+		ResourceFilters: []resourcepolicies.ResourceFilter{
+			{
+				Kinds:         []string{"pods", "secrets"},
+				LabelSelector: map[string]string{"app": "foo"},
+			},
+		},
+	}
+	res, err := resolveClusterScopedFilterPolicy(validPolicy, helper, log)
+	require.NoError(t, err)
+	require.Len(t, res, 2)
+	assert.Contains(t, res, "pods")
+	assert.Contains(t, res, "secrets")
+	assert.True(t, res["pods"].LabelSelector.Matches(labels.Set{"app": "foo"}))
+
+	// Test warning branches
+	mockHelper := &mockDiscoveryHelper{
+		Helper: helper,
+		ResourceForFunc: func(input schema.GroupVersionResource) (schema.GroupVersionResource, metav1.APIResource, error) {
+			if input.Resource == "invalid-resource" {
+				return schema.GroupVersionResource{}, metav1.APIResource{}, errors.New("cannot resolve")
+			}
+			if input.Resource == "namespaced-resource" {
+				return schema.GroupVersionResource{Resource: "namespaced-resource"}, metav1.APIResource{Namespaced: true, Name: "namespaced-resource"}, nil
+			}
+			return helper.ResourceFor(input)
+		},
+	}
+
+	policyWithWarns := &resourcepolicies.ClusterScopedFilterPolicy{
+		ResourceFilters: []resourcepolicies.ResourceFilter{
+			{
+				Kinds: []string{"invalid-resource", "namespaced-resource"},
+			},
+		},
+	}
+	res2, err2 := resolveClusterScopedFilterPolicy(policyWithWarns, mockHelper, log)
+	require.NoError(t, err2)
+	assert.Contains(t, res2, "invalid-resource")
+	assert.Contains(t, res2, "namespaced-resource")
+}
+
+func TestResolveNamespacedFilterPolicies(t *testing.T) {
+	helper := test.NewFakeDiscoveryHelper(true, nil)
+	log := test.NewLogger()
+
+	policies := []resourcepolicies.NamespacedFilterPolicy{
+		{
+			Namespaces: []string{"ns1", "ns-*"},
+			ResourceFilters: []resourcepolicies.ResourceFilter{
+				{
+					Kinds:         []string{"pods"},
+					LabelSelector: map[string]string{"app": "foo"},
+				},
+				{
+					Kinds:         []string{"*"},
+					LabelSelector: map[string]string{"catch": "all"},
+				},
+			},
+		},
+	}
+
+	res, patterns, err := resolveNamespacedFilterPolicies(policies, helper, log)
+	require.NoError(t, err)
+	require.Len(t, res, 2)
+	require.Len(t, patterns, 2)
+
+	assert.Contains(t, res, "ns1")
+	assert.Contains(t, res, "ns-*")
+
+	ns1Filter := res["ns1"]
+	require.NotNil(t, ns1Filter)
+	require.NotNil(t, ns1Filter.CatchAllFilter)
+	assert.True(t, ns1Filter.CatchAllFilter.LabelSelector.Matches(labels.Set{"catch": "all"}))
+	require.Contains(t, ns1Filter.ResourceFilterMap, "pods")
+	assert.True(t, ns1Filter.ResourceFilterMap["pods"].LabelSelector.Matches(labels.Set{"app": "foo"}))
+
+	// Test with invalid label selector
+	invalidPolicies := []resourcepolicies.NamespacedFilterPolicy{
+		{
+			Namespaces: []string{"ns1"},
+			ResourceFilters: []resourcepolicies.ResourceFilter{
+				{
+					Kinds:         []string{"pods"},
+					LabelSelector: map[string]string{"invalid/label/key": "value"},
+				},
+			},
+		},
+	}
+	_, _, err = resolveNamespacedFilterPolicies(invalidPolicies, helper, log)
+	require.Error(t, err)
+
+	// Test warning branches
+	mockHelper := &mockDiscoveryHelper{
+		Helper: helper,
+		ResourceForFunc: func(input schema.GroupVersionResource) (schema.GroupVersionResource, metav1.APIResource, error) {
+			if input.Resource == "invalid-resource" {
+				return schema.GroupVersionResource{}, metav1.APIResource{}, errors.New("cannot resolve")
+			}
+			if input.Resource == "cluster-scoped-resource" {
+				return schema.GroupVersionResource{Resource: "cluster-scoped-resource"}, metav1.APIResource{Namespaced: false, Name: "cluster-scoped-resource"}, nil
+			}
+			return schema.GroupVersionResource{Resource: input.Resource}, metav1.APIResource{Namespaced: true, Name: input.Resource}, nil
+		},
+	}
+
+	policyWithWarns := []resourcepolicies.NamespacedFilterPolicy{
+		{
+			Namespaces: []string{"ns1"},
+			ResourceFilters: []resourcepolicies.ResourceFilter{
+				{
+					Kinds: []string{"invalid-resource", "cluster-scoped-resource"},
+				},
+			},
+		},
+	}
+	resWarns, _, errWarns := resolveNamespacedFilterPolicies(policyWithWarns, mockHelper, log)
+	require.NoError(t, errWarns)
+	require.Contains(t, resWarns["ns1"].ResourceFilterMap, "invalid-resource")
+	require.Contains(t, resWarns["ns1"].ResourceFilterMap, "cluster-scoped-resource")
+}
+
+func TestBackupWithResPoliciesLogs(t *testing.T) {
+	itemBlockPool := StartItemBlockWorkerPool(t.Context(), 1, logrus.StandardLogger())
+	defer itemBlockPool.Stop()
+
+	h := newHarness(t, itemBlockPool)
+
+	// Add some resources so discovery helper knows about them
+	h.addItems(t, test.Pods(builder.ForPod("ns1", "pod-1").Result()))
+	h.addItems(t, test.PVs(builder.ForPersistentVolume("pv-1").Result()))
+
+	backupReq := &Request{
+		Backup:           defaultBackup().ExcludedNamespaceScopedResources("pods").Result(),
+		SkippedPVTracker: NewSkipPVTracker(),
+		BackedUpItems:    NewBackedUpItemsMap(),
+		WorkerPool:       itemBlockPool,
+	}
+
+	p := new(resourcepolicies.Policies)
+	inputPolicy := &resourcepolicies.ResourcePolicies{
+		Version: "v1",
+		ClusterScopedFilterPolicy: &resourcepolicies.ClusterScopedFilterPolicy{
+			ResourceFilters: []resourcepolicies.ResourceFilter{
+				{Kinds: []string{"pods", "invalid-cluster-kind"}},
+			},
+		},
+		NamespacedFilterPolicies: []resourcepolicies.NamespacedFilterPolicy{
+			{
+				Namespaces: []string{"ns1"},
+				ResourceFilters: []resourcepolicies.ResourceFilter{
+					{Kinds: []string{"persistentvolumes", "pods", "invalid-ns-kind"}},
+				},
+			},
+		},
+	}
+	require.NoError(t, p.BuildPolicy(inputPolicy))
+	backupReq.ResPolicies = p
+
+	backupFile := bytes.NewBuffer([]byte{})
+	err := h.backupper.Backup(h.log, backupReq, backupFile, nil, nil, nil)
+	require.NoError(t, err)
+
+	// Add test to cover error returns from resolve policies
+	badClusterPol := &resourcepolicies.ClusterScopedFilterPolicy{
+		ResourceFilters: []resourcepolicies.ResourceFilter{
+			{
+				Kinds:         []string{"pods"},
+				LabelSelector: map[string]string{"invalid/label/key": "value"},
+			},
+		},
+	}
+	pBadCluster := new(resourcepolicies.Policies)
+	require.NoError(t, pBadCluster.BuildPolicy(&resourcepolicies.ResourcePolicies{
+		Version:                   "v1",
+		ClusterScopedFilterPolicy: badClusterPol,
+	}))
+	backupReq.ResPolicies = pBadCluster
+	err = h.backupper.Backup(h.log, backupReq, backupFile, nil, nil, nil)
+	require.Error(t, err)
+
+	badNsPol := []resourcepolicies.NamespacedFilterPolicy{
+		{
+			Namespaces: []string{"ns1"},
+			ResourceFilters: []resourcepolicies.ResourceFilter{
+				{
+					Kinds:         []string{"pods"},
+					LabelSelector: map[string]string{"invalid/label/key": "value"},
+				},
+			},
+		},
+	}
+	pBadNs := new(resourcepolicies.Policies)
+	require.NoError(t, pBadNs.BuildPolicy(&resourcepolicies.ResourcePolicies{
+		Version:                  "v1",
+		NamespacedFilterPolicies: badNsPol,
+	}))
+	backupReq.ResPolicies = pBadNs
+	err = h.backupper.Backup(h.log, backupReq, backupFile, nil, nil, nil)
+	require.Error(t, err)
+}

pkg/backup/item_backupper.go

@@ -142,6 +142,42 @@ func (ib *itemBackupper) itemInclusionChecks(log logrus.FieldLogger, mustInclude
 			log.Info("Excluding item because resource is excluded")
 			return false
 		}
+
+		// Per-kind name filter from ResourcePolicy namespace filter.
+		if namespace != "" {
+			if nsFilter := ib.backupRequest.GetNamespaceFilter(namespace); nsFilter != nil {
+				rf := nsFilter.ResourceFilterMap[groupResource.String()]
+				if rf == nil {
+					rf = nsFilter.CatchAllFilter
+				}
+				// When rf is still nil the item's kind is not listed in the namespace filter and
+				// there is no catch-all entry. This is an intentional permissive passthrough:
+				// plugin-injected additional items (returned by BackupItemAction) must be able
+				// to reach the archive even when their kind was not explicitly listed in
+				// namespacedFilterPolicies, because excluding them at Stage 2 would break backup
+				// completeness. For example, a CSI plugin may inject a VolumeSnapshotContent
+				// as an additional item that is required for a correct restore. Kind-level
+				// exclusion for the primary collection pass is enforced earlier in
+				// item_collector.go (Stage 1).
+				if rf != nil && rf.NameIE != nil {
+					if !rf.NameIE.ShouldInclude(metadata.GetName()) {
+						log.Infof("Excluding item: name does not match resource filter for kind %s",
+							groupResource)
+						return false
+					}
+				}
+			}
+		} else {
+			// Cluster-scoped resource name filter
+			if ib.backupRequest.ClusterScopedFilterMap != nil {
+				if rf, ok := ib.backupRequest.ClusterScopedFilterMap[groupResource.String()]; ok && rf.NameIE != nil {
+					if !rf.NameIE.ShouldInclude(metadata.GetName()) {
+						log.Infof("Excluding item: name does not match clusterScopedFilterPolicy for kind %s", groupResource)
+						return false
+					}
+				}
+			}
+		}
 	}
 
 	if metadata.GetDeletionTimestamp() != nil {

pkg/backup/item_backupper_test.go

@@ -21,20 +21,20 @@ import (
 	"testing"
 
 	"github.com/sirupsen/logrus"
-	"github.com/stretchr/testify/require"
-	"k8s.io/apimachinery/pkg/runtime/schema"
-	ctrlfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
-
-	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
-	"github.com/vmware-tanzu/velero/pkg/kuberesource"
-
 	"github.com/stretchr/testify/assert"
+	"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/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	ctrlfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/util/collections"
 )
 
 func Test_resourceKey(t *testing.T) {
@@ -494,3 +494,284 @@ func TestUnTrackSkippedPV_PendingLostPVC(t *testing.T) {
 		})
 	}
 }
+
+// includeAllIE is a minimal IncludesExcludesInterface that includes everything —
+// used in tests where the global resource include/exclude logic is not under test.
+type includeAllIE struct{}
+
+func (includeAllIE) ShouldInclude(string) bool { return true }
+func (includeAllIE) ShouldExclude(string) bool { return false }
+
+// makeTestUnstructured creates an unstructured object with the given namespace, name, and labels.
+func makeTestUnstructured(namespace, name string, labels map[string]string) *unstructured.Unstructured {
+	obj := &unstructured.Unstructured{}
+	obj.SetNamespace(namespace)
+	obj.SetName(name)
+	if labels != nil {
+		obj.SetLabels(labels)
+	}
+	return obj
+}
+
+// makeNameIE creates an IncludesExcludes that includes only the given glob patterns.
+func makeNameIE(include ...string) *collections.IncludesExcludes {
+	ie := collections.NewIncludesExcludes()
+	ie.Includes(include...)
+	return ie
+}
+
+// newTestItemBackupper builds a minimal itemBackupper suitable for itemInclusionChecks tests.
+func newTestItemBackupper(req *Request) *itemBackupper {
+	return &itemBackupper{
+		backupRequest: req,
+	}
+}
+
+// baseRequest returns a Request with NamespaceIncludesExcludes and ResourceIncludesExcludes
+// configured to include everything, so only the filter-map logic under test is exercised.
+func baseRequest() *Request {
+	return &Request{
+		Backup:                    builder.ForBackup("velero", "test-backup").Result(),
+		NamespaceIncludesExcludes: collections.NewNamespaceIncludesExcludes().Includes("*"),
+		ResourceIncludesExcludes:  includeAllIE{},
+		SkippedPVTracker:          NewSkipPVTracker(),
+	}
+}
+
+var configMapsGR = schema.GroupResource{Group: "", Resource: "configmaps"}
+var clusterRolesGR = schema.GroupResource{Group: "rbac.authorization.k8s.io", Resource: "clusterroles"}
+
+// TestItemInclusionChecks_ExcludeLabel_OverridesNamespaceFilter verifies that
+// velero.io/exclude-from-backup=true takes precedence over a namespacedFilterPolicies
+// entry that would otherwise include the resource.
+func TestItemInclusionChecks_ExcludeLabel_OverridesNamespaceFilter(t *testing.T) {
+	req := baseRequest()
+	req.NamespacedFilterMap = map[string]*ResolvedNamespaceFilter{
+		"ns-a": {
+			ResourceFilterMap: map[string]*ResolvedResourceFilter{
+				configMapsGR.String(): {}, // include all ConfigMaps in ns-a
+			},
+		},
+	}
+	req.NamespacedFilterPatterns = []NamespacedFilterPattern{}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	obj := makeTestUnstructured("ns-a", "my-config", map[string]string{
+		velerov1api.ExcludeFromBackupLabel: "true",
+	})
+
+	result := ib.itemInclusionChecks(log, false, obj, obj, configMapsGR)
+	assert.False(t, result, "resource with exclude-from-backup=true must be excluded even when matched by namespacedFilterPolicies")
+}
+
+// TestItemInclusionChecks_ExcludeLabel_OverridesCatchAll verifies that
+// velero.io/exclude-from-backup=true takes precedence over the catch-all filter.
+func TestItemInclusionChecks_ExcludeLabel_OverridesCatchAll(t *testing.T) {
+	catchAllFilter := &ResolvedResourceFilter{} // include everything via catch-all
+	req := baseRequest()
+	req.NamespacedFilterMap = map[string]*ResolvedNamespaceFilter{
+		"ns-a": {
+			ResourceFilterMap: map[string]*ResolvedResourceFilter{},
+			CatchAllFilter:    catchAllFilter,
+		},
+	}
+	req.NamespacedFilterPatterns = []NamespacedFilterPattern{}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	obj := makeTestUnstructured("ns-a", "my-config", map[string]string{
+		velerov1api.ExcludeFromBackupLabel: "true",
+	})
+
+	result := ib.itemInclusionChecks(log, false, obj, obj, configMapsGR)
+	assert.False(t, result, "resource with exclude-from-backup=true must be excluded even when matched by catch-all filter")
+}
+
+// TestItemInclusionChecks_ExcludeLabel_OverridesClusterScopedFilter verifies that
+// velero.io/exclude-from-backup=true takes precedence over clusterScopedFilterPolicy.
+func TestItemInclusionChecks_ExcludeLabel_OverridesClusterScopedFilter(t *testing.T) {
+	req := baseRequest()
+	req.ClusterScopedFilterMap = map[string]*ResolvedResourceFilter{
+		clusterRolesGR.String(): {}, // include all ClusterRoles
+	}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	// Cluster-scoped object: no namespace
+	obj := makeTestUnstructured("", "my-role", map[string]string{
+		velerov1api.ExcludeFromBackupLabel: "true",
+	})
+
+	result := ib.itemInclusionChecks(log, false, obj, obj, clusterRolesGR)
+	assert.False(t, result, "cluster-scoped resource with exclude-from-backup=true must be excluded even when in clusterScopedFilterPolicy")
+}
+
+// TestItemInclusionChecks_ClusterScoped_NotInFilterMap_PassesThrough verifies that
+// a dynamically injected cluster-scoped resource NOT listed in ClusterScopedFilterMap
+// passes through itemInclusionChecks (permissive passthrough at Stage 2).
+func TestItemInclusionChecks_ClusterScoped_NotInFilterMap_PassesThrough(t *testing.T) {
+	req := baseRequest()
+	req.ClusterScopedFilterMap = map[string]*ResolvedResourceFilter{
+		clusterRolesGR.String(): {}, // only ClusterRoles are listed
+	}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	// VolumeSnapshotClass is NOT in the filter map
+	volumeSnapshotClassGR := schema.GroupResource{Group: "snapshot.storage.k8s.io", Resource: "volumesnapshotclasses"}
+	obj := makeTestUnstructured("", "standard", nil)
+
+	result := ib.itemInclusionChecks(log, false, obj, obj, volumeSnapshotClassGR)
+	assert.True(t, result, "cluster-scoped resource not in ClusterScopedFilterMap must pass through (permissive Stage 2 for unlisted kinds)")
+}
+
+// TestItemInclusionChecks_ClusterScoped_NameIE_Matching verifies that a cluster-scoped
+// resource listed in ClusterScopedFilterMap with a NameIE filter is included/excluded
+// based on its name.
+func TestItemInclusionChecks_ClusterScoped_NameIE_Matching(t *testing.T) {
+	req := baseRequest()
+	req.ClusterScopedFilterMap = map[string]*ResolvedResourceFilter{
+		clusterRolesGR.String(): {
+			NameIE: makeNameIE("my-app-*"),
+		},
+	}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	// Matching name
+	matching := makeTestUnstructured("", "my-app-reader", nil)
+	assert.True(t, ib.itemInclusionChecks(log, false, matching, matching, clusterRolesGR),
+		"ClusterRole matching name pattern must be included")
+
+	// Non-matching name
+	nonMatching := makeTestUnstructured("", "other-role", nil)
+	assert.False(t, ib.itemInclusionChecks(log, false, nonMatching, nonMatching, clusterRolesGR),
+		"ClusterRole not matching name pattern must be excluded")
+}
+
+// TestItemInclusionChecks_GlobalExclusion_OverridesNamespaceFilter verifies that
+// a resource kind globally excluded by includeExcludePolicy is rejected at Stage 2
+// even when a namespacedFilterPolicies entry lists that kind. The global
+// ResourceIncludesExcludes.ShouldInclude check fires before the per-namespace filter.
+func TestItemInclusionChecks_GlobalExclusion_OverridesNamespaceFilter(t *testing.T) {
+	// excludeSecretsIE excludes "secrets" globally, includes everything else.
+	excludeSecretsIE := &excludeResourceIE{excluded: "secrets"}
+
+	req := &Request{
+		Backup:                    builder.ForBackup("velero", "test-backup").Result(),
+		NamespaceIncludesExcludes: collections.NewNamespaceIncludesExcludes().Includes("*"),
+		ResourceIncludesExcludes:  excludeSecretsIE,
+		SkippedPVTracker:          NewSkipPVTracker(),
+		// namespacedFilterPolicies says to back up Secrets in ns-a
+		NamespacedFilterMap: map[string]*ResolvedNamespaceFilter{
+			"ns-a": {
+				ResourceFilterMap: map[string]*ResolvedResourceFilter{
+					"secrets.": {}, // Secret listed in per-namespace filter
+				},
+			},
+		},
+		NamespacedFilterPatterns: []NamespacedFilterPattern{},
+	}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	secretsGR := schema.GroupResource{Group: "", Resource: "secrets"}
+	obj := makeTestUnstructured("ns-a", "my-secret", nil)
+
+	result := ib.itemInclusionChecks(log, false, obj, obj, secretsGR)
+	assert.False(t, result,
+		"Secret must be excluded because it is globally excluded by ResourceIncludesExcludes, "+
+			"even though namespacedFilterPolicies lists it")
+}
+
+// TestItemInclusionChecks_PluginItem_UnlistedKind_NoCatchAll_PassesThrough verifies the
+// intentional permissive passthrough at Stage 2 for plugin-injected additional items.
+// When a namespace has a namespacedFilterPolicies entry but the item's kind is not listed
+// in that policy and there is no catch-all entry, itemInclusionChecks must still allow
+// the item through.
+//
+// Rationale: plugin-injected additional items (returned by BackupItemAction) must be able
+// to reach the archive even when their kind was not explicitly listed in the filter policy,
+// because rejecting them here would break backup completeness. For example, a CSI plugin
+// may inject a VolumeSnapshotContent that is required for a correct restore.
+// Kind-level exclusion for the primary collection pass is enforced at Stage 1 in
+// item_collector.go, not at Stage 2 here.
+func TestItemInclusionChecks_PluginItem_UnlistedKind_NoCatchAll_PassesThrough(t *testing.T) {
+	req := baseRequest()
+	// Namespace filter only lists ConfigMaps; Secrets are not listed and there is no catch-all.
+	req.NamespacedFilterMap = map[string]*ResolvedNamespaceFilter{
+		"ns-a": {
+			ResourceFilterMap: map[string]*ResolvedResourceFilter{
+				configMapsGR.String(): {},
+			},
+			CatchAllFilter: nil,
+		},
+	}
+	req.NamespacedFilterPatterns = []NamespacedFilterPattern{}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	secretsGR := schema.GroupResource{Group: "", Resource: "secrets"}
+	obj := makeTestUnstructured("ns-a", "plugin-injected-secret", nil)
+
+	result := ib.itemInclusionChecks(log, false, obj, obj, secretsGR)
+	assert.True(t, result,
+		"plugin-injected additional item of an unlisted kind must pass through Stage 2 "+
+			"even when its namespace has a namespacedFilterPolicies entry with no catch-all; "+
+			"kind exclusion is enforced at Stage 1 (item_collector.go), not here")
+}
+
+// TestItemInclusionChecks_PluginItem_UnlistedKind_WithCatchAll_PassesThrough verifies that
+// a plugin-injected additional item of a kind not listed in the namespace filter also passes
+// through Stage 2 when a catch-all entry is present. The catch-all is validated to never
+// carry a NameIE (names/excludedNames are prohibited on catch-all entries), so the name
+// check is always a no-op for catch-all-matched items and the item is included.
+func TestItemInclusionChecks_PluginItem_UnlistedKind_WithCatchAll_PassesThrough(t *testing.T) {
+	req := baseRequest()
+	// Namespace filter lists ConfigMaps explicitly; a catch-all covers everything else.
+	// The catch-all has no NameIE — this is enforced by validation.
+	req.NamespacedFilterMap = map[string]*ResolvedNamespaceFilter{
+		"ns-a": {
+			ResourceFilterMap: map[string]*ResolvedResourceFilter{
+				configMapsGR.String(): {},
+			},
+			CatchAllFilter: &ResolvedResourceFilter{
+				// NameIE intentionally nil: validation forbids names/excludedNames on catch-all
+				NameIE: nil,
+			},
+		},
+	}
+	req.NamespacedFilterPatterns = []NamespacedFilterPattern{}
+
+	ib := newTestItemBackupper(req)
+	log := logrus.New()
+
+	secretsGR := schema.GroupResource{Group: "", Resource: "secrets"}
+	obj := makeTestUnstructured("ns-a", "plugin-injected-secret", nil)
+
+	result := ib.itemInclusionChecks(log, false, obj, obj, secretsGR)
+	assert.True(t, result,
+		"plugin-injected additional item matched by catch-all must pass through Stage 2; "+
+			"the catch-all has no NameIE so the name check is a no-op")
+}
+
+// excludeResourceIE is an IncludesExcludesInterface that excludes a single resource
+// type and includes everything else. Used to simulate includeExcludePolicy global exclusions.
+type excludeResourceIE struct {
+	excluded string
+}
+
+func (e *excludeResourceIE) ShouldInclude(typeName string) bool {
+	return typeName != e.excluded
+}
+func (e *excludeResourceIE) ShouldExclude(typeName string) bool {
+	return typeName == e.excluded
+}

pkg/backup/item_collector.go

@@ -462,6 +462,7 @@ func (r *itemCollector) getResourceItems(
 	}
 
 	clusterScoped := !resource.Namespaced
+
 	namespacesToList := getNamespacesToList(r.backupRequest.NamespaceIncludesExcludes)
 
 	// If we get here, we're backing up something other than namespaces
@@ -472,6 +473,16 @@ func (r *itemCollector) getResourceItems(
 	var items []*kubernetesResource
 
 	for _, namespace := range namespacesToList {
+		// Check per-namespace resource type filter from ResourcePolicy
+		if nsFilter := r.backupRequest.GetNamespaceFilter(namespace); nsFilter != nil {
+			_, hasSpecific := nsFilter.ResourceFilterMap[gr.String()]
+			if !hasSpecific && nsFilter.CatchAllFilter == nil {
+				log.Debugf("Skipping resource %s in namespace %s: not in resourceFilters",
+					gr, namespace)
+				continue
+			}
+		}
+
 		unstructuredItems, err := r.listResourceByLabelsPerNamespace(
 			namespace, gr, gv, resource, log)
 		if err != nil {
@@ -527,13 +538,47 @@ func (r *itemCollector) listResourceByLabelsPerNamespace(
 		return nil, err
 	}
 
+	// Determine label selectors — per-namespace/per-kind or global
 	var orLabelSelectors []string
-	if r.backupRequest.Spec.OrLabelSelectors != nil {
-		for _, s := range r.backupRequest.Spec.OrLabelSelectors {
-			orLabelSelectors = append(orLabelSelectors, metav1.FormatLabelSelector(s))
+	var labelSelector string
+
+	if !resource.Namespaced && r.backupRequest.ClusterScopedFilterMap != nil {
+		rf := r.backupRequest.ClusterScopedFilterMap[gr.String()]
+		if rf != nil {
+			if rf.LabelSelector != nil {
+				labelSelector = rf.LabelSelector.String()
+			}
+			if len(rf.OrLabelSelectors) > 0 {
+				for _, s := range rf.OrLabelSelectors {
+					orLabelSelectors = append(orLabelSelectors, s.String())
+				}
+			}
+		}
+	} else if nsFilter := r.backupRequest.GetNamespaceFilter(namespace); nsFilter != nil {
+		rf := nsFilter.ResourceFilterMap[gr.String()]
+		if rf == nil {
+			rf = nsFilter.CatchAllFilter
+		}
+		if rf != nil {
+			if rf.LabelSelector != nil {
+				labelSelector = rf.LabelSelector.String()
+			}
+			if len(rf.OrLabelSelectors) > 0 {
+				for _, s := range rf.OrLabelSelectors {
+					orLabelSelectors = append(orLabelSelectors, s.String())
+				}
+			}
 		}
 	} else {
-		orLabelSelectors = []string{}
+		// Use global selectors (existing behavior)
+		if r.backupRequest.Spec.OrLabelSelectors != nil {
+			for _, s := range r.backupRequest.Spec.OrLabelSelectors {
+				orLabelSelectors = append(orLabelSelectors, metav1.FormatLabelSelector(s))
+			}
+		}
+		if selector := r.backupRequest.Spec.LabelSelector; selector != nil {
+			labelSelector = metav1.FormatLabelSelector(selector)
+		}
 	}
 
 	logger.Info("Listing items")
@@ -553,11 +598,6 @@ func (r *itemCollector) listResourceByLabelsPerNamespace(
 		return nil, err
 	}
 
-	var labelSelector string
-	if selector := r.backupRequest.Spec.LabelSelector; selector != nil {
-		labelSelector = metav1.FormatLabelSelector(selector)
-	}
-
 	// Listing items for labelSelector (singular)
 	if len(orLabelSelectors) == 0 {
 		unstructuredItems, err = r.listItemsForLabel(

pkg/backup/item_collector_test.go

@@ -26,7 +26,9 @@ import (
 	corev1api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/builder"
@@ -279,8 +281,9 @@ func TestItemCollectorBackupNamespaces(t *testing.T) {
 					Backup:                    tc.backup,
 					NamespaceIncludesExcludes: tc.ie,
 				},
-				dynamicFactory: factory,
-				dir:            tempDir,
+				dynamicFactory:  factory,
+				discoveryHelper: test.NewFakeDiscoveryHelper(true, nil),
+				dir:             tempDir,
 			}
 
 			if tc.converter == nil {
@@ -305,3 +308,140 @@ func TestItemCollectorBackupNamespaces(t *testing.T) {
 		})
 	}
 }
+
+// TestNamespacedFilterMap_GlobalExclusionPrecedence verifies the precedence rule:
+// ResourceIncludesExcludes (set by includeExcludePolicy) is checked before the
+// NamespacedFilterMap. This is enforced at both Stage 1 (item_collector.go line ~430)
+// and Stage 2 (item_backupper.go itemInclusionChecks). The unit below confirms that
+// GetNamespaceFilter still returns a filter for the namespace — it is the caller's
+// responsibility to check ResourceIncludesExcludes first, which item_collector does.
+//
+// Full coverage of the Stage 2 enforcement is in item_backupper_test.go
+// TestItemInclusionChecks_GlobalExclusion_OverridesNamespaceFilter.
+func TestNamespacedFilterMap_GlobalExclusionPrecedence(t *testing.T) {
+	req := &Request{
+		Backup:                    builder.ForBackup("velero", "test-backup").Result(),
+		NamespaceIncludesExcludes: collections.NewNamespaceIncludesExcludes().Includes("ns-a"),
+		NamespacedFilterMap: map[string]*ResolvedNamespaceFilter{
+			"ns-a": {
+				ResourceFilterMap: map[string]*ResolvedResourceFilter{
+					"secrets.": {},
+				},
+			},
+		},
+		NamespacedFilterPatterns: []NamespacedFilterPattern{},
+	}
+
+	// GetNamespaceFilter returns the filter regardless of global exclusions.
+	// The caller (item_collector) is responsible for checking ResourceIncludesExcludes first.
+	nsFilter := req.GetNamespaceFilter("ns-a")
+	require.NotNil(t, nsFilter, "GetNamespaceFilter should return a filter for ns-a")
+	_, hasSecrets := nsFilter.ResourceFilterMap["secrets."]
+	assert.True(t, hasSecrets, "ns-a filter should list secrets GR")
+
+	// When a global excludeAllIE is set, item_collector would return nil before consulting the map.
+	// This is verified by the Stage 1 check: ShouldInclude("secrets.") == false → skip.
+	ie := &excludeAllIE{}
+	assert.False(t, ie.ShouldInclude("secrets."),
+		"global exclusion must reject secrets before the per-namespace filter is consulted")
+}
+
+// excludeAllIE is an IncludesExcludesInterface that excludes every resource kind.
+type excludeAllIE struct{}
+
+func (excludeAllIE) ShouldInclude(string) bool { return false }
+func (excludeAllIE) ShouldExclude(string) bool { return true }
+
+func TestGetResourceItems(t *testing.T) {
+	tests := []struct {
+		name                   string
+		namespaces             []string
+		clusterScopedFilterMap map[string]*ResolvedResourceFilter
+		namespacedFilterMap    map[string]*ResolvedNamespaceFilter
+		resource               metav1.APIResource
+		gr                     schema.GroupResource
+	}{
+		{
+			name:       "cluster scoped resource with filter",
+			namespaces: []string{""},
+			resource: metav1.APIResource{
+				Name:       "persistentvolumes",
+				Namespaced: false,
+			},
+			gr: schema.GroupResource{Resource: "persistentvolumes"},
+			clusterScopedFilterMap: map[string]*ResolvedResourceFilter{
+				"persistentvolumes": {
+					LabelSelector: labels.Set{"app": "foo"}.AsSelector(),
+				},
+			},
+		},
+		{
+			name:       "namespace scoped resource with filter",
+			namespaces: []string{"ns1"},
+			resource: metav1.APIResource{
+				Name:       "pods",
+				Namespaced: true,
+			},
+			gr: schema.GroupResource{Resource: "pods"},
+			namespacedFilterMap: map[string]*ResolvedNamespaceFilter{
+				"ns1": {
+					ResourceFilterMap: map[string]*ResolvedResourceFilter{
+						"pods": {
+							LabelSelector: labels.Set{"app": "bar"}.AsSelector(),
+						},
+					},
+				},
+			},
+		},
+		{
+			name:       "namespace scoped resource skipped due to no filter match",
+			namespaces: []string{"ns1"},
+			resource: metav1.APIResource{
+				Name:       "secrets",
+				Namespaced: true,
+			},
+			gr: schema.GroupResource{Resource: "secrets"},
+			namespacedFilterMap: map[string]*ResolvedNamespaceFilter{
+				"ns1": {
+					ResourceFilterMap: map[string]*ResolvedResourceFilter{
+						"pods": {
+							LabelSelector: labels.Set{"app": "bar"}.AsSelector(),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			dc := &test.FakeDynamicClient{}
+			dc.On("List", mock.Anything).Return(&unstructured.UnstructuredList{}, nil)
+
+			factory := &test.FakeDynamicFactory{}
+			factory.On("ClientForGroupVersionResource", mock.Anything, mock.Anything, mock.Anything).Return(dc, nil)
+
+			req := &Request{
+				Backup:                   builder.ForBackup("velero", "backup").Result(),
+				ClusterScopedFilterMap:   tc.clusterScopedFilterMap,
+				NamespacedFilterMap:      tc.namespacedFilterMap,
+				ResourceIncludesExcludes: includeAllIE{},
+			}
+			if len(tc.namespaces) > 0 && tc.namespaces[0] != "" {
+				req.NamespaceIncludesExcludes = collections.NewNamespaceIncludesExcludes().Includes(tc.namespaces...)
+			} else {
+				req.NamespaceIncludesExcludes = collections.NewNamespaceIncludesExcludes().Includes("*")
+			}
+
+			r := &itemCollector{
+				backupRequest:   req,
+				dynamicFactory:  factory,
+				discoveryHelper: test.NewFakeDiscoveryHelper(true, nil),
+				log:             test.NewLogger(),
+			}
+
+			_, err := r.getResourceItems(test.NewLogger(), schema.GroupVersion{}, tc.resource, nil)
+			assert.NoError(t, err)
+		})
+	}
+}