restore filters via resource policy

restore filters via resource policy, support ClusterScopedFilterPolicy
and NamespaceFilterPolicies.

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

changelogs/unreleased/9946-adam-jian-zhang

@@ -0,0 +1 @@
+Fix issue #9936, restore filters via resource policy implementation

pkg/controller/restore_controller.go

@@ -44,6 +44,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/internal/hook"
 	"github.com/vmware-tanzu/velero/internal/resourcemodifiers"
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/volume"
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/constant"
@@ -232,7 +233,7 @@ func (r *restoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	original := restore.DeepCopy()
 
 	// Validate the restore and fetch the backup
-	info, resourceModifiers := r.validateAndComplete(restore)
+	info, resourceModifiers, restoreResPolicies := r.validateAndComplete(ctx, restore)
 
 	// Register attempts after validation so we don't have to fetch the backup multiple times
 	backupScheduleName := restore.Spec.ScheduleName
@@ -267,7 +268,7 @@ func (r *restoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, nil
 	}
 
-	if err := r.runValidatedRestore(restore, info, resourceModifiers); err != nil {
+	if err := r.runValidatedRestore(restore, info, resourceModifiers, restoreResPolicies); err != nil {
 		log.WithError(err).Debug("Restore failed")
 		restore.Status.Phase = api.RestorePhaseFailed
 		restore.Status.FailureReason = err.Error()
@@ -303,7 +304,7 @@ func (r *restoreReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Complete(r)
 }
 
-func (r *restoreReconciler) validateAndComplete(restore *api.Restore) (backupInfo, *resourcemodifiers.ResourceModifiers) {
+func (r *restoreReconciler) validateAndComplete(ctx context.Context, restore *api.Restore) (backupInfo, *resourcemodifiers.ResourceModifiers, *resourcepolicies.Policies) {
 	// add non-restorable resources to restore's excluded resources
 	excludedResources := sets.NewString(restore.Spec.ExcludedResources...)
 	for _, nonrestorable := range nonRestorableResources {
@@ -338,7 +339,7 @@ func (r *restoreReconciler) validateAndComplete(restore *api.Restore) (backupInf
 	// validate that exactly one of BackupName and ScheduleName have been specified
 	if !backupXorScheduleProvided(restore) {
 		restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, "Either a backup or schedule must be specified as a source for the restore, but not both")
-		return backupInfo{}, nil
+		return backupInfo{}, nil, nil
 	}
 
 	// validate Restore Init Hook's InitContainers
@@ -372,9 +373,9 @@ func (r *restoreReconciler) validateAndComplete(restore *api.Restore) (backupInf
 		}))
 
 		backupList := &api.BackupList{}
-		if err := r.kbClient.List(context.Background(), backupList, &client.ListOptions{LabelSelector: selector}); err != nil {
+		if err := r.kbClient.List(ctx, backupList, &client.ListOptions{LabelSelector: selector}); err != nil {
 			restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, "Unable to list backups for schedule")
-			return backupInfo{}, nil
+			return backupInfo{}, nil, nil
 		}
 		if len(backupList.Items) == 0 {
 			restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, "No backups found for schedule")
@@ -384,19 +385,19 @@ func (r *restoreReconciler) validateAndComplete(restore *api.Restore) (backupInf
 			restore.Spec.BackupName = backup.Name
 		} else {
 			restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, "No completed backups found for schedule")
-			return backupInfo{}, nil
+			return backupInfo{}, nil, nil
 		}
 	}
 
 	info, err := r.fetchBackupInfo(restore.Spec.BackupName)
 	if err != nil {
 		restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, fmt.Sprintf("Error retrieving backup: %v", err))
-		return backupInfo{}, nil
+		return backupInfo{}, nil, nil
 	}
 
 	if !veleroutil.BSLIsAvailable(*info.location) {
 		restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, fmt.Sprintf("The BSL %s is unavailable, cannot retrieve the backup", info.location.Name))
-		return backupInfo{}, nil
+		return backupInfo{}, nil, nil
 	}
 
 	// reject restores from backups that are not in a usable phase
@@ -407,7 +408,7 @@ func (r *restoreReconciler) validateAndComplete(restore *api.Restore) (backupInf
 		restore.Status.ValidationErrors = append(restore.Status.ValidationErrors,
 			fmt.Sprintf("backup %q is in phase %q and cannot be used as a restore source",
 				info.backup.Name, info.backup.Status.Phase))
-		return backupInfo{}, nil
+		return backupInfo{}, nil, nil
 	}
 
 	// Fill in the ScheduleName so it's easier to consume for metrics.
@@ -415,26 +416,40 @@ func (r *restoreReconciler) validateAndComplete(restore *api.Restore) (backupInf
 		restore.Spec.ScheduleName = info.backup.GetLabels()[api.ScheduleNameLabel]
 	}
 
+	var restoreResPolicies *resourcepolicies.Policies
+	if restore.Spec.ResourcePolicy != nil {
+		var err error
+		restoreResPolicies, err = resourcepolicies.GetResourcePoliciesFromRestore(
+			ctx, restore, r.kbClient, r.logger,
+		)
+		if err != nil {
+			restore.Status.ValidationErrors = append(
+				restore.Status.ValidationErrors, err.Error(),
+			)
+			return backupInfo{}, nil, nil
+		}
+	}
+
 	var resourceModifiers *resourcemodifiers.ResourceModifiers
 	if restore.Spec.ResourceModifier != nil && strings.EqualFold(restore.Spec.ResourceModifier.Kind, resourcemodifiers.ConfigmapRefType) {
 		ResourceModifierConfigMap := &corev1api.ConfigMap{}
-		err := r.kbClient.Get(context.Background(), client.ObjectKey{Namespace: restore.Namespace, Name: restore.Spec.ResourceModifier.Name}, ResourceModifierConfigMap)
+		err := r.kbClient.Get(ctx, client.ObjectKey{Namespace: restore.Namespace, Name: restore.Spec.ResourceModifier.Name}, ResourceModifierConfigMap)
 		if err != nil {
 			restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, fmt.Sprintf("failed to get resource modifiers configmap %s/%s", restore.Namespace, restore.Spec.ResourceModifier.Name))
-			return backupInfo{}, nil
+			return backupInfo{}, nil, nil
 		}
 		resourceModifiers, err = resourcemodifiers.GetResourceModifiersFromConfig(ResourceModifierConfigMap)
 		if err != nil {
 			restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, errors.Wrapf(err, "Error in parsing resource modifiers provided in configmap %s/%s", restore.Namespace, restore.Spec.ResourceModifier.Name).Error())
-			return backupInfo{}, nil
+			return backupInfo{}, nil, nil
 		} else if err = resourceModifiers.Validate(); err != nil {
 			restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, errors.Wrapf(err, "Validation error in resource modifiers provided in configmap %s/%s", restore.Namespace, restore.Spec.ResourceModifier.Name).Error())
-			return backupInfo{}, nil
+			return backupInfo{}, nil, nil
 		}
 		r.logger.Infof("Retrieved Resource modifiers provided in configmap %s/%s", restore.Namespace, restore.Spec.ResourceModifier.Name)
 	}
 
-	return info, resourceModifiers
+	return info, resourceModifiers, restoreResPolicies
 }
 
 // backupXorScheduleProvided returns true if exactly one of BackupName and
@@ -507,7 +522,7 @@ func fetchBackupInfoInternal(kbClient client.Client, namespace, backupName strin
 // The log and results files are uploaded to backup storage. Any error returned from this function
 // means that the restore failed. This function updates the restore API object with warning and error
 // counts, but *does not* update its phase or patch it via the API.
-func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backupInfo, resourceModifiers *resourcemodifiers.ResourceModifiers) error {
+func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backupInfo, resourceModifiers *resourcemodifiers.ResourceModifiers, restoreResPolicies *resourcepolicies.Policies) error {
 	// instantiate the per-restore logger that will output both to a temp file
 	// (for upload to object storage) and to stdout.
 	restoreLog, err := logging.NewTempFileLogger(r.restoreLogLevel, r.logFormat, nil, logrus.Fields{"restore": kubeutil.NamespaceAndName(restore)})
@@ -586,6 +601,7 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 		VolumeSnapshots:               volumeSnapshots,
 		BackupReader:                  backupFile,
 		ResourceModifiers:             resourceModifiers,
+		ResPolicies:                   restoreResPolicies,
 		DisableInformerCache:          r.disableInformerCache,
 		CSIVolumeSnapshots:            csiVolumeSnapshots,
 		BackupVolumeInfoMap:           backupVolumeInfoMap,

pkg/controller/restore_controller_test.go

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"bytes"
+	"context"
 	"io"
 	"testing"
 	"time"
@@ -785,7 +786,7 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 		Phase(velerov1api.BackupPhaseCompleted).
 		Result()))
 
-	r.validateAndComplete(restore)
+	r.validateAndComplete(context.Background(), restore)
 	assert.Contains(t, restore.Status.ValidationErrors, "No backups found for schedule")
 	assert.Empty(t, restore.Spec.BackupName)
 
@@ -801,7 +802,7 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 			Result(),
 	))
 
-	r.validateAndComplete(restore)
+	r.validateAndComplete(context.Background(), restore)
 	assert.Contains(t, restore.Status.ValidationErrors, "No completed backups found for schedule")
 	assert.Empty(t, restore.Spec.BackupName)
 
@@ -832,11 +833,140 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 			ScheduleName: "schedule-1",
 		},
 	}
-	r.validateAndComplete(restore)
+	r.validateAndComplete(context.Background(), restore)
 	assert.Nil(t, restore.Status.ValidationErrors)
 	assert.Equal(t, "foo", restore.Spec.BackupName)
 }
 
+func TestValidateAndCompleteWithResourcePolicySpecified(t *testing.T) {
+	formatFlag := logging.FormatText
+
+	var (
+		logger           = velerotest.NewLogger()
+		pluginManager    = &pluginmocks.Manager{}
+		fakeClient       = velerotest.NewFakeControllerRuntimeClient(t)
+		fakeGlobalClient = velerotest.NewFakeControllerRuntimeClient(t)
+		backupStore      = &persistencemocks.BackupStore{}
+	)
+
+	r := NewRestoreReconciler(
+		t.Context(),
+		velerov1api.DefaultNamespace,
+		nil,
+		fakeClient,
+		logger,
+		logrus.DebugLevel,
+		func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
+		NewFakeSingleObjectBackupStoreGetter(backupStore),
+		metrics.NewServerMetrics(),
+		formatFlag,
+		60*time.Minute,
+		false,
+		fakeGlobalClient,
+		10*time.Minute,
+	)
+
+	restore := &velerov1api.Restore{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: velerov1api.DefaultNamespace,
+			Name:      "restore-1",
+		},
+		Spec: velerov1api.RestoreSpec{
+			BackupName: "backup-1",
+			ResourcePolicy: &corev1api.TypedLocalObjectReference{
+				Kind: "configmap",
+				Name: "test-configmap",
+			},
+		},
+	}
+
+	location := builder.ForBackupStorageLocation("velero", "default").Provider("myCloud").Bucket("bucket").Phase(velerov1api.BackupStorageLocationPhaseAvailable).Result()
+	require.NoError(t, r.kbClient.Create(t.Context(), location))
+
+	require.NoError(t, r.kbClient.Create(
+		t.Context(),
+		defaultBackup().
+			ObjectMeta(
+				builder.WithName("backup-1"),
+			).StorageLocation("default").
+			Phase(velerov1api.BackupPhaseCompleted).
+			Result(),
+	))
+
+	r.validateAndComplete(context.Background(), restore)
+	assert.Contains(t, restore.Status.ValidationErrors[0], "fail to get ResourcePolicies velero/test-configmap ConfigMap")
+
+	restore1 := &velerov1api.Restore{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: velerov1api.DefaultNamespace,
+			Name:      "restore-1",
+		},
+		Spec: velerov1api.RestoreSpec{
+			BackupName: "backup-1",
+			ResourcePolicy: &corev1api.TypedLocalObjectReference{
+				Kind: "configmap",
+				Name: "test-configmap",
+			},
+		},
+	}
+
+	cm1 := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-configmap",
+			Namespace: velerov1api.DefaultNamespace,
+		},
+		Data: map[string]string{
+			"policy.yaml": `version: v1
+clusterScopedFilterPolicy:
+  resourceFilters:
+    - kinds:
+      - pods
+`,
+		},
+	}
+	require.NoError(t, r.kbClient.Create(t.Context(), cm1))
+
+	r.validateAndComplete(context.Background(), restore1)
+	assert.Nil(t, restore1.Status.ValidationErrors)
+
+	restore2 := &velerov1api.Restore{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: velerov1api.DefaultNamespace,
+			Name:      "restore-1",
+		},
+		Spec: velerov1api.RestoreSpec{
+			BackupName: "backup-1",
+			ResourcePolicy: &corev1api.TypedLocalObjectReference{
+				// intentional to ensure case insensitivity works as expected
+				Kind: "confIGMaP",
+				Name: "test-configmap-invalid",
+			},
+		},
+	}
+
+	cm2 := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-configmap-invalid",
+			Namespace: velerov1api.DefaultNamespace,
+		},
+		Data: map[string]string{
+			"policy.yaml": `version: v1
+volumePolicies:
+  - conditions:
+      capacity: '0,10Gi'
+      csi:
+        driver: disks.csi.driver
+    action:
+      type: invalid_action
+`,
+		},
+	}
+	require.NoError(t, r.kbClient.Create(t.Context(), cm2))
+
+	r.validateAndComplete(context.Background(), restore2)
+	assert.Contains(t, restore2.Status.ValidationErrors[0], "fail to validate ResourcePolicies in ConfigMap velero/test-configmap-invalid")
+}
+
 func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 	formatFlag := logging.FormatText
 
@@ -892,7 +1022,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 			Result(),
 	))
 
-	r.validateAndComplete(restore)
+	r.validateAndComplete(context.Background(), restore)
 	assert.Contains(t, restore.Status.ValidationErrors[0], "failed to get resource modifiers configmap")
 
 	restore1 := &velerov1api.Restore{
@@ -920,7 +1050,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), cm1))
 
-	r.validateAndComplete(restore1)
+	r.validateAndComplete(context.Background(), restore1)
 	assert.Nil(t, restore1.Status.ValidationErrors)
 
 	restore2 := &velerov1api.Restore{
@@ -949,7 +1079,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), invalidVersionCm))
 
-	r.validateAndComplete(restore2)
+	r.validateAndComplete(context.Background(), restore2)
 	assert.Contains(t, restore2.Status.ValidationErrors[0], "Error in parsing resource modifiers provided in configmap")
 
 	restore3 := &velerov1api.Restore{
@@ -977,7 +1107,7 @@ func TestValidateAndCompleteWithResourceModifierSpecified(t *testing.T) {
 	}
 	require.NoError(t, r.kbClient.Create(t.Context(), invalidOperatorCm))
 
-	r.validateAndComplete(restore3)
+	r.validateAndComplete(context.Background(), restore3)
 	assert.Contains(t, restore3.Status.ValidationErrors[0], "Validation error in resource modifiers provided in configmap")
 }
 

pkg/restore/request.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	"github.com/vmware-tanzu/velero/internal/resourcemodifiers"
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/volume"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
@@ -61,6 +62,7 @@ type Request struct {
 	RestoredItems                 map[itemKey]restoredItemStatus
 	itemOperationsList            *[]*itemoperation.RestoreOperation
 	ResourceModifiers             *resourcemodifiers.ResourceModifiers
+	ResPolicies                   *resourcepolicies.Policies
 	DisableInformerCache          bool
 	CSIVolumeSnapshots            []*snapshotv1api.VolumeSnapshot
 	BackupVolumeInfoMap           map[string]volume.BackupVolumeInfo

pkg/restore/restore.go

@@ -32,6 +32,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/gobwas/glob"
 	"github.com/google/uuid"
 	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	"github.com/sirupsen/logrus"
@@ -55,6 +56,7 @@ import (
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	"github.com/vmware-tanzu/velero/internal/hook"
 	"github.com/vmware-tanzu/velero/internal/resourcemodifiers"
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/volume"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/archive"
@@ -237,6 +239,43 @@ func (kr *kubernetesRestorer) RestoreWithResolvers(
 		Includes(req.Restore.Spec.IncludedNamespaces...).
 		Excludes(req.Restore.Spec.ExcludedNamespaces...)
 
+	var clusterScopedFilterMap map[string]*resolvedResourceFilter
+	var namespacedFilterMap map[string]*resolvedNamespaceFilter
+	var namespacedFilterPatterns []namespacedFilterPattern
+
+	if req.ResPolicies != nil {
+		if kr.discoveryHelper == nil {
+			return results.Result{}, results.Result{Velero: []string{"failed to resolve namespace filter policies: discovery client unavailable"}}
+		}
+
+		// Resolve clusterScopedFilterPolicy
+		csPolicy := req.ResPolicies.GetClusterScopedFilterPolicy()
+		if csPolicy != nil {
+			clusterScopedFilterMap, err = resolveRestoreClusterScopedFilterPolicy(
+				csPolicy,
+				kr.discoveryHelper,
+				req.Log,
+			)
+			if err != nil {
+				return results.Result{}, results.Result{Velero: []string{err.Error()}}
+			}
+		}
+
+		// Resolve namespacedFilterPolicies
+		nfPolicies := req.ResPolicies.GetNamespacedFilterPolicies()
+		if len(nfPolicies) > 0 {
+			namespacedFilterMap, namespacedFilterPatterns, err = resolveRestoreNamespacedFilterPolicies(
+				nfPolicies,
+				req.Restore.Spec.ExcludedResources,
+				kr.discoveryHelper,
+				req.Log,
+			)
+			if err != nil {
+				return results.Result{}, results.Result{Velero: []string{err.Error()}}
+			}
+		}
+	}
+
 	resolvedActions, err := restoreItemActionResolver.ResolveActions(kr.discoveryHelper, kr.logger)
 	if err != nil {
 		return results.Result{}, results.Result{Velero: []string{err.Error()}}
@@ -333,6 +372,10 @@ func (kr *kubernetesRestorer) RestoreWithResolvers(
 		restoreVolumeInfoTracker:       req.RestoreVolumeInfoTracker,
 		hooksWaitExecutor:              hooksWaitExecutor,
 		resourceDeletionStatusTracker:  req.ResourceDeletionStatusTracker,
+		clusterScopedFilterMap:         clusterScopedFilterMap,
+		namespacedFilterMap:            namespacedFilterMap,
+		namespacedFilterPatterns:       namespacedFilterPatterns,
+		namespaceFilterCache:           make(map[string]*resolvedNamespaceFilter),
 	}
 
 	return restoreCtx.execute()
@@ -382,6 +425,216 @@ type restoreContext struct {
 	restoreVolumeInfoTracker       *volume.RestoreVolumeInfoTracker
 	hooksWaitExecutor              *hooksWaitExecutor
 	resourceDeletionStatusTracker  kube.ResourceDeletionStatusTracker
+
+	// clusterScopedFilterMap holds resolved per-kind filters for cluster-scoped resources.
+	// Key is the resolved group-resource string.
+	clusterScopedFilterMap map[string]*resolvedResourceFilter
+
+	// namespacedFilterMap holds resolved per-namespace filters.
+	// Key is either an exact namespace name or a glob pattern string.
+	namespacedFilterMap map[string]*resolvedNamespaceFilter
+
+	// namespacedFilterPatterns preserves the order of patterns for first-match
+	// semantics and caches pre-compiled globs to avoid repeated compilation.
+	namespacedFilterPatterns []namespacedFilterPattern
+
+	// namespaceFilterCache memoizes the resolved filter for a given namespace
+	// to avoid re-evaluating glob patterns on every call.
+	namespaceFilterCache map[string]*resolvedNamespaceFilter
+}
+
+type resolvedResourceFilter struct {
+	labelSelector    labels.Selector
+	orLabelSelectors []labels.Selector
+	nameIE           *collections.IncludesExcludes
+}
+
+type resolvedNamespaceFilter struct {
+	// resourceFilterMap is keyed by the resolved group-resource string
+	resourceFilterMap map[string]*resolvedResourceFilter
+	// catchAllFilter holds the resolved filter for a catch-all entry (empty kinds or ["*"]).
+	// nil when no catch-all entry is defined.
+	catchAllFilter *resolvedResourceFilter
+}
+
+// namespacedFilterPattern pairs a namespace pattern string with its pre-compiled
+// glob so that getNamespaceFilter does not recompile on every call.
+type namespacedFilterPattern struct {
+	pattern  string
+	compiled glob.Glob // compiled once at restore start; nil for exact-match patterns
+}
+
+func (ctx *restoreContext) getNamespaceFilter(namespace string) *resolvedNamespaceFilter {
+	if ctx.namespacedFilterMap == nil {
+		return nil
+	}
+
+	// 1. Check the cache first
+	if filter, ok := ctx.namespaceFilterCache[namespace]; ok {
+		return filter
+	}
+
+	// 2. Walk patterns in definition order (first-match semantics)
+	for _, p := range ctx.namespacedFilterPatterns {
+		if p.compiled != nil {
+			if p.compiled.Match(namespace) {
+				filter := ctx.namespacedFilterMap[p.pattern]
+				ctx.namespaceFilterCache[namespace] = filter
+				return filter
+			}
+		} else if p.pattern == namespace {
+			filter := ctx.namespacedFilterMap[p.pattern]
+			ctx.namespaceFilterCache[namespace] = filter
+			return filter
+		}
+	}
+
+	// 3. Cache the miss so we don't re-evaluate failed matches
+	ctx.namespaceFilterCache[namespace] = nil
+	return nil
+}
+
+// resolveRestoreClusterScopedFilterPolicy resolves the cluster-scoped filter policy
+// into a map keyed by group-resource string. Note: catch-all entries (empty or ["*"] kinds)
+// are NOT supported in clusterScopedFilterPolicy — validation rejects them earlier.
+// Cluster-scoped filtering is a refinement overlay; unlisted kinds fall back to global
+// filters via the existing pipeline, so there is no catchAllFilter field on this map.
+func resolveRestoreClusterScopedFilterPolicy(
+	policy *resourcepolicies.ClusterScopedFilterPolicy,
+	helper discovery.Helper,
+	log logrus.FieldLogger,
+) (map[string]*resolvedResourceFilter, error) {
+	result := 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, resource, err := helper.ResourceFor(schema.GroupVersionResource{Resource: kind})
+			if err != nil {
+				log.WithField("kind", kind).Warnf("Cannot resolve kind via discovery, using as-is")
+				result[kind] = resolved
+				continue
+			}
+			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)
+			}
+			result[gr.GroupResource().String()] = resolved
+		}
+	}
+	return result, nil
+}
+
+func resolveRestoreNamespacedFilterPolicies(
+	policies []resourcepolicies.NamespacedFilterPolicy,
+	excludedResources []string,
+	helper discovery.Helper,
+	log logrus.FieldLogger,
+) (map[string]*resolvedNamespaceFilter, []namespacedFilterPattern, error) {
+	result := make(map[string]*resolvedNamespaceFilter)
+	var patternOrder []namespacedFilterPattern
+
+	// Build a quick lookup map for globally excluded resources
+	globalExcludes := make(map[string]bool)
+	for _, ex := range excludedResources {
+		globalExcludes[ex] = true
+	}
+
+	for _, policy := range policies {
+		rfMap := make(map[string]*resolvedResourceFilter)
+		var catchAll *resolvedResourceFilter
+
+		for _, rf := range policy.ResourceFilters {
+			resolved, err := resolveResourceFilter(rf)
+			if err != nil {
+				return nil, nil, err
+			}
+
+			if rf.IsCatchAll() {
+				catchAll = resolved
+				continue
+			}
+
+			for _, kind := range rf.Kinds {
+				gr, resource, err := helper.ResourceFor(
+					schema.GroupVersionResource{Resource: kind},
+				)
+				if err != nil {
+					log.WithField("kind", kind).Warnf(
+						"Cannot resolve kind via discovery, using as-is")
+					rfMap[kind] = resolved
+					continue
+				}
+
+				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")
+				}
+
+				rfMap[gr.GroupResource().String()] = resolved
+			}
+		}
+
+		nsFilter := &resolvedNamespaceFilter{
+			resourceFilterMap: rfMap,
+			catchAllFilter:    catchAll,
+		}
+		for _, nsPattern := range policy.Namespaces {
+			result[nsPattern] = nsFilter
+			var compiled glob.Glob
+			if strings.ContainsAny(nsPattern, "*?[") {
+				var err error
+				compiled, err = glob.Compile(nsPattern)
+				if err != nil {
+					log.WithError(err).Warnf("Failed to compile namespace glob pattern %q, falling back to exact match", nsPattern)
+				}
+			}
+			patternOrder = append(patternOrder, namespacedFilterPattern{
+				pattern:  nsPattern,
+				compiled: compiled,
+			})
+		}
+	}
+	return result, patternOrder, nil
+}
+
+// resolveResourceFilter converts a ResourceFilter's label selectors and name patterns
+// into their runtime representations.
+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().Includes(rf.Names...).Excludes(rf.ExcludedNames...)
+	}
+	return &resolvedResourceFilter{
+		labelSelector:    selector,
+		orLabelSelectors: orSelectors,
+		nameIE:           nameIE,
+	}, nil
 }
 
 type resourceClientKey struct {
@@ -1128,6 +1381,10 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 	// and should be excluded. Note that we're checking the object's namespace (
 	// via obj.GetNamespace()) instead of the namespace parameter, because we want
 	// to check the *original* namespace, not the remapped one if it's been remapped.
+	//
+	// Note: Additional items intentionally bypass fine-grained resource filter policies
+	// (like per-namespace label/name selectors) to avoid breaking semantic dependencies,
+	// but they must still pass the global exclusions enforced below.
 	if namespace != "" {
 		if !ctx.namespaceIncludesExcludes.ShouldInclude(obj.GetNamespace()) && !ctx.resourceMustHave.Has(groupResource.String()) {
 			restoreLogger.Info("Not restoring item because namespace is excluded")
@@ -2277,6 +2534,18 @@ func (ctx *restoreContext) getOrderedResourceCollection(
 				continue
 			}
 
+			// Per-namespace resource type check from restore filter policy
+			if namespace != "" && !ctx.resourceMustHave.Has(groupResource.String()) {
+				if nsFilter := ctx.getNamespaceFilter(namespace); nsFilter != nil {
+					_, kindListed := nsFilter.resourceFilterMap[groupResource.String()]
+					if !kindListed && nsFilter.catchAllFilter == nil {
+						ctx.log.Infof("Skipping resource %s in namespace %s: not in resourceFilters",
+							resource, namespace)
+						continue
+					}
+				}
+			}
+
 			res, w, e := ctx.getSelectedRestoreableItems(groupResource.String(), namespace, items)
 			warnings.Merge(&w)
 			errs.Merge(&e)
@@ -2330,6 +2599,30 @@ func (ctx *restoreContext) getSelectedRestoreableItems(resource string, original
 		resourceForPath = filepath.Join(resource, cgv.Dir)
 	}
 
+	var rf *resolvedResourceFilter
+	var useFilterPolicy bool
+
+	if !ctx.resourceMustHave.Has(resource) {
+		if originalNamespace != "" {
+			// Namespace-scoped path
+			if nsFilter := ctx.getNamespaceFilter(originalNamespace); nsFilter != nil {
+				// Resolve effective filter: kind-specific takes precedence over catch-all
+				rf = nsFilter.resourceFilterMap[resource]
+				if rf == nil {
+					rf = nsFilter.catchAllFilter // may be nil if no catch-all
+				}
+				useFilterPolicy = true
+			}
+		} else if ctx.clusterScopedFilterMap != nil {
+			// Cluster-scoped path: only applies if kind is listed (refinement overlay)
+			if listedRF, ok := ctx.clusterScopedFilterMap[resource]; ok {
+				rf = listedRF
+				useFilterPolicy = true
+			}
+			// If kind not listed, fall through to global selectors below
+		}
+	}
+
 	for _, item := range items {
 		itemPath := archive.GetItemFilePath(ctx.restoreDir, resourceForPath, originalNamespace, item)
 
@@ -2347,29 +2640,58 @@ func (ctx *restoreContext) getSelectedRestoreableItems(resource string, original
 		}
 
 		if !ctx.resourceMustHave.Has(resource) {
-			if !ctx.selector.Matches(labels.Set(obj.GetLabels())) {
-				continue
-			}
-
-			// Processing OrLabelSelectors when specified in the restore request. LabelSelectors as well as OrLabelSelectors
-			// cannot co-exist, only one of them can be specified
-			var skipItem = false
-			var skip = 0
-			ctx.log.Debugf("orSelectors specified: %s for item: %s", ctx.OrSelectors, item)
-			for _, s := range ctx.OrSelectors {
-				if !s.Matches(labels.Set(obj.GetLabels())) {
-					skip++
+			if useFilterPolicy {
+				if rf != nil {
+					// Per-kind label selector
+					if rf.labelSelector != nil && !rf.labelSelector.Matches(labels.Set(obj.GetLabels())) {
+						continue
+					}
+					// Per-kind OR label selectors
+					if len(rf.orLabelSelectors) > 0 {
+						matched := false
+						for _, s := range rf.orLabelSelectors {
+							if s.Matches(labels.Set(obj.GetLabels())) {
+								matched = true
+								break
+							}
+						}
+						if !matched {
+							ctx.log.Infof("Excluding item %s: no OR label selector matched (restore filter policy)", item)
+							continue
+						}
+					}
+					// Per-kind name filter
+					if rf.nameIE != nil && !rf.nameIE.ShouldInclude(obj.GetName()) {
+						ctx.log.Infof("Excluding item %s: name does not match restore filter policy", obj.GetName())
+						continue
+					}
+				}
+			} else {
+				// Existing global selector logic
+				if !ctx.selector.Matches(labels.Set(obj.GetLabels())) {
+					continue
 				}
 
-				if len(ctx.OrSelectors) == skip && skip > 0 {
-					ctx.log.Infof("setting skip flag to true for item: %s", item)
-					skipItem = true
-				}
-			}
+				// Processing OrLabelSelectors when specified in the restore request. LabelSelectors as well as OrLabelSelectors
+				// cannot co-exist, only one of them can be specified
+				var skipItem = false
+				var skip = 0
+				ctx.log.Debugf("orSelectors specified: %s for item: %s", ctx.OrSelectors, item)
+				for _, s := range ctx.OrSelectors {
+					if !s.Matches(labels.Set(obj.GetLabels())) {
+						skip++
+					}
 
-			if skipItem {
-				ctx.log.Infof("restore orSelector labels did not match, skipping restore of item: %s", skipItem, item)
-				continue
+					if len(ctx.OrSelectors) == skip && skip > 0 {
+						ctx.log.Infof("setting skip flag to true for item: %s", item)
+						skipItem = true
+					}
+				}
+
+				if skipItem {
+					ctx.log.Infof("restore orSelector labels did not match, skipping restore of item: %s", skipItem, item)
+					continue
+				}
 			}
 		}
 

pkg/restore/restore_policies_test.go

@@ -0,0 +1,206 @@
+package restore
+
+import (
+	"context"
+	"io"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	"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/test"
+)
+
+func TestRestoreResourcePoliciesFiltering(t *testing.T) {
+	tests := []struct {
+		name         string
+		restore      *velerov1api.Restore
+		backup       *velerov1api.Backup
+		policyYAML   string
+		apiResources []*test.APIResource
+		tarball      io.Reader
+		want         map[*test.APIResource][]string
+	}{
+		{
+			name:    "namespaced filter policy with exact namespace match",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			policyYAML: `version: v1
+namespacedFilterPolicies:
+  - namespaces:
+      - ns-1
+    resourceFilters:
+      - kinds:
+          - pods
+        names:
+          - pod-1
+`,
+			tarball: test.NewTarWriter(t).
+				AddItems("pods",
+					builder.ForPod("ns-1", "pod-1").Result(),
+					builder.ForPod("ns-1", "pod-2").Result(),
+					builder.ForPod("ns-2", "pod-1").Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.Pods(),
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods(): {"ns-1/pod-1", "ns-2/pod-1"}, // ns-2 is not filtered, ns-1 only includes pod-1
+			},
+		},
+		{
+			name:    "namespaced filter policy with glob namespace match and first-match semantics",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			policyYAML: `version: v1
+namespacedFilterPolicies:
+  - namespaces:
+      - ns-*
+    resourceFilters:
+      - kinds:
+          - pods
+        names:
+          - pod-1
+  - namespaces:
+      - ns-1
+    resourceFilters:
+      - kinds:
+          - pods
+        names:
+          - pod-2
+`,
+			tarball: test.NewTarWriter(t).
+				AddItems("pods",
+					builder.ForPod("ns-1", "pod-1").Result(),
+					builder.ForPod("ns-1", "pod-2").Result(),
+					builder.ForPod("ns-2", "pod-1").Result(),
+					builder.ForPod("ns-2", "pod-2").Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.Pods(),
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods(): {"ns-1/pod-1", "ns-2/pod-1"},
+			},
+		},
+		{
+			name:    "cluster scoped filter policy",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			policyYAML: `version: v1
+clusterScopedFilterPolicy:
+  resourceFilters:
+    - kinds:
+        - persistentvolumes
+      names:
+        - pv-1
+`,
+			tarball: test.NewTarWriter(t).
+				AddItems("persistentvolumes",
+					builder.ForPersistentVolume("pv-1").Result(),
+					builder.ForPersistentVolume("pv-2").Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.PVs(),
+			},
+			want: map[*test.APIResource][]string{
+				test.PVs(): {"/pv-1"},
+			},
+		},
+		{
+			name:    "catch-all filter",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			policyYAML: `version: v1
+namespacedFilterPolicies:
+  - namespaces:
+      - ns-1
+    resourceFilters:
+      - kinds:
+          - '*'
+        labelSelector:
+          app: test
+`,
+			tarball: test.NewTarWriter(t).
+				AddItems("pods",
+					builder.ForPod("ns-1", "pod-1").ObjectMeta(builder.WithLabels("app", "test")).Result(),
+					builder.ForPod("ns-1", "pod-2").Result(),
+				).
+				AddItems("deployments.apps",
+					builder.ForDeployment("ns-1", "deploy-1").ObjectMeta(builder.WithLabels("app", "test")).Result(),
+					builder.ForDeployment("ns-1", "deploy-2").Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.Pods(),
+				test.Deployments(),
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods():        {"ns-1/pod-1"},
+				test.Deployments(): {"ns-1/deploy-1"},
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t)
+
+			for _, r := range tc.apiResources {
+				h.DiscoveryClient.WithAPIResource(r)
+			}
+			require.NoError(t, h.restorer.discoveryHelper.Refresh())
+
+			var resPolicies *resourcepolicies.Policies
+			if tc.policyYAML != "" {
+				cm := &corev1api.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-policies",
+						Namespace: "velero",
+					},
+					Data: map[string]string{
+						"policy.yaml": tc.policyYAML,
+					},
+				}
+				client := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(cm).Build()
+				restore := tc.restore.DeepCopy()
+				restore.Namespace = "velero"
+				restore.Spec.ResourcePolicy = &corev1api.TypedLocalObjectReference{
+					Kind: "configmap",
+					Name: "test-policies",
+				}
+				var err error
+				resPolicies, err = resourcepolicies.GetResourcePoliciesFromRestore(context.Background(), restore, client, logrus.New())
+				require.NoError(t, err)
+			}
+
+			data := &Request{
+				Log:              h.log,
+				Restore:          tc.restore,
+				Backup:           tc.backup,
+				PodVolumeBackups: nil,
+				VolumeSnapshots:  nil,
+				BackupReader:     tc.tarball,
+				ResPolicies:      resPolicies,
+			}
+			warnings, errs := h.restorer.Restore(
+				data,
+				nil, // restoreItemActions
+				nil, // volume snapshotter getter
+			)
+
+			assertEmptyResults(t, warnings, errs)
+			assertAPIContents(t, h, tc.want)
+		})
+	}
+}