Add include/exclude policy to resources policy

fixes #8610

This commit extends the resources policy, such that user can define
resource include exclude filters in the policy and reuse it in different backups.

Signed-off-by: Daniel Jiang <daniel.jiang@broadcom.com>

.github/workflows/pr-codespell.yml

@@ -15,7 +15,7 @@ jobs:
       with:
         # ignore the config/.../crd.go file as it's generated binary data that is edited elsewhere.
         skip: .git,*.png,*.jpg,*.woff,*.ttf,*.gif,*.ico,./config/crd/v1beta1/crds/crds.go,./config/crd/v1/crds/crds.go,./config/crd/v2alpha1/crds/crds.go,./go.sum,./LICENSE
-        ignore_words_list: iam,aks,ist,bridget,ue,shouldnot,atleast,notin,sme,optin
+        ignore_words_list: iam,aks,ist,bridget,ue,shouldnot,atleast,notin,sme,optin,sie
         check_filenames: true
         check_hidden: true
 

changelogs/unreleased/9145-reasonerjt

@@ -0,0 +1 @@
+Add include/exclude policy to resources policy

internal/resourcepolicies/resource_policies.go

@@ -13,6 +13,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 */
+
 package resourcepolicies
 
 import (
@@ -20,6 +21,8 @@ import (
 	"fmt"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/util/sets"
+
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
@@ -49,24 +52,58 @@ type Action struct {
 	Parameters map[string]any `yaml:"parameters,omitempty"`
 }
 
-// volumePolicy defined policy to conditions to match Volumes and related action to handle matched Volumes
+// IncludeExcludePolicy defined policy to include or exclude resources based on the names
+type IncludeExcludePolicy struct {
+	// The following fields have the same semantics as those from the spec of backup.
+	// Refer to the comment in the velerov1api.BackupSpec for more details.
+	IncludedClusterScopedResources   []string `yaml:"includedClusterScopedResources"`
+	ExcludedClusterScopedResources   []string `yaml:"excludedClusterScopedResources"`
+	IncludedNamespaceScopedResources []string `yaml:"includedNamespaceScopedResources"`
+	ExcludedNamespaceScopedResources []string `yaml:"excludedNamespaceScopedResources"`
+}
+
+func (p *IncludeExcludePolicy) Validate() error {
+	if err := p.validateIncludeExclude(p.IncludedClusterScopedResources, p.ExcludedClusterScopedResources); err != nil {
+		return err
+	}
+	return p.validateIncludeExclude(p.IncludedNamespaceScopedResources, p.ExcludedNamespaceScopedResources)
+}
+
+func (p *IncludeExcludePolicy) validateIncludeExclude(includesList, excludesList []string) error {
+	includes := sets.NewString(includesList...)
+	excludes := sets.NewString(excludesList...)
+
+	if includes.Has("*") || excludes.Has("*") {
+		return fmt.Errorf("cannot use '*' in includes or excludes filters in the policy")
+	}
+	for _, itm := range excludes.List() {
+		if includes.Has(itm) {
+			return fmt.Errorf("excludes list cannot contain an item in the includes list: %s", itm)
+		}
+	}
+	return nil
+}
+
+// VolumePolicy defined policy to conditions to match Volumes and related action to handle matched Volumes
 type VolumePolicy struct {
 	// Conditions defined list of conditions to match Volumes
 	Conditions map[string]any `yaml:"conditions"`
 	Action     Action         `yaml:"action"`
 }
 
-// resourcePolicies currently defined slice of volume policies to handle backup
+// ResourcePolicies currently defined slice of volume policies to handle backup
 type ResourcePolicies struct {
-	Version        string         `yaml:"version"`
+	Version              string                `yaml:"version"`
-	VolumePolicies []VolumePolicy `yaml:"volumePolicies"`
+	VolumePolicies       []VolumePolicy        `yaml:"volumePolicies"`
+	IncludeExcludePolicy *IncludeExcludePolicy `yaml:"includeExcludePolicy"`
 	// we may support other resource policies in the future, and they could be added separately
 	// OtherResourcePolicies []OtherResourcePolicy
 }
 
 type Policies struct {
-	version        string
+	version              string
-	volumePolicies []volPolicy
+	volumePolicies       []volPolicy
+	includeExcludePolicy *IncludeExcludePolicy
 	// OtherPolicies
 }
 
@@ -115,6 +152,7 @@ func (p *Policies) BuildPolicy(resPolicies *ResourcePolicies) error {
 	// Other resource policies
 
 	p.version = resPolicies.Version
+	p.includeExcludePolicy = resPolicies.IncludeExcludePolicy
 	return nil
 }
 
@@ -175,9 +213,20 @@ func (p *Policies) Validate() error {
 			}
 		}
 	}
+
+	if p.GetIncludeExcludePolicy() != nil {
+		if err := p.GetIncludeExcludePolicy().Validate(); err != nil {
+			return errors.WithStack(err)
+		}
+	}
+
 	return nil
 }
 
+func (p *Policies) GetIncludeExcludePolicy() *IncludeExcludePolicy {
+	return p.includeExcludePolicy
+}
+
 func GetResourcePoliciesFromBackup(
 	backup velerov1api.Backup,
 	client crclient.Client,

internal/resourcepolicies/volume_resources_validator_test.go

@@ -453,6 +453,102 @@ func TestValidate(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: " '*' in the filters of include exclude policy - 1",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{Type: "skip"},
+						Conditions: map[string]any{
+							"pvcLabels": map[string]string{
+								"environment": "production",
+								"app":         "database",
+							},
+						},
+					},
+				},
+				IncludeExcludePolicy: &IncludeExcludePolicy{
+					IncludedClusterScopedResources:   []string{"*"},
+					ExcludedClusterScopedResources:   []string{"crds"},
+					IncludedNamespaceScopedResources: []string{"pods"},
+					ExcludedNamespaceScopedResources: []string{"secrets"},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: " '*' in the filters of include exclude policy - 2",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{Type: "skip"},
+						Conditions: map[string]any{
+							"pvcLabels": map[string]string{
+								"environment": "production",
+								"app":         "database",
+							},
+						},
+					},
+				},
+				IncludeExcludePolicy: &IncludeExcludePolicy{
+					IncludedClusterScopedResources:   []string{"persistentvolumes"},
+					ExcludedClusterScopedResources:   []string{"crds"},
+					IncludedNamespaceScopedResources: []string{"pods"},
+					ExcludedNamespaceScopedResources: []string{"*"},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: " dup item in both the include and exclude filters of include exclude policy",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{Type: "skip"},
+						Conditions: map[string]any{
+							"pvcLabels": map[string]string{
+								"environment": "production",
+								"app":         "database",
+							},
+						},
+					},
+				},
+				IncludeExcludePolicy: &IncludeExcludePolicy{
+					IncludedClusterScopedResources:   []string{"persistentvolumes"},
+					ExcludedClusterScopedResources:   []string{"crds"},
+					IncludedNamespaceScopedResources: []string{"pods", "configmaps"},
+					ExcludedNamespaceScopedResources: []string{"secrets", "pods"},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: " valid volume policies and valid include/exclude policy",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{Type: "skip"},
+						Conditions: map[string]any{
+							"pvcLabels": map[string]string{
+								"environment": "production",
+								"app":         "database",
+							},
+						},
+					},
+				},
+				IncludeExcludePolicy: &IncludeExcludePolicy{
+					IncludedClusterScopedResources:   []string{"persistentvolumes"},
+					ExcludedClusterScopedResources:   []string{"crds"},
+					IncludedNamespaceScopedResources: []string{"pods", "configmaps"},
+					ExcludedNamespaceScopedResources: []string{"secrets"},
+				},
+			},
+			wantErr: false,
+		},
 	}
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {

pkg/backup/backup.go

@@ -43,7 +43,6 @@ 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"
@@ -270,13 +269,17 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 			backupRequest.Spec.IncludeClusterResources,
 			*backupRequest.NamespaceIncludesExcludes)
 	} else {
-		backupRequest.ResourceIncludesExcludes = collections.GetScopeResourceIncludesExcludes(kb.discoveryHelper, log,
+		srie := collections.GetScopeResourceIncludesExcludes(kb.discoveryHelper, log,
 			backupRequest.Spec.IncludedNamespaceScopedResources,
 			backupRequest.Spec.ExcludedNamespaceScopedResources,
 			backupRequest.Spec.IncludedClusterScopedResources,
 			backupRequest.Spec.ExcludedClusterScopedResources,
 			*backupRequest.NamespaceIncludesExcludes,
 		)
+		if backupRequest.ResPolicies != nil {
+			srie.CombineWithPolicy(backupRequest.ResPolicies.GetIncludeExcludePolicy())
+		}
+		backupRequest.ResourceIncludesExcludes = srie
 	}
 
 	log.Infof("Backing up all volumes using pod volume backup: %t", boolptr.IsSetToTrue(backupRequest.Backup.Spec.DefaultVolumesToFsBackup))
@@ -355,11 +358,6 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 	}
 	backupRequest.Status.Progress = &velerov1api.BackupProgress{TotalItems: len(items)}
 
-	var resourcePolicy *resourcepolicies.Policies
-	if backupRequest.ResPolicies != nil {
-		resourcePolicy = backupRequest.ResPolicies
-	}
-
 	itemBackupper := &itemBackupper{
 		backupRequest:            backupRequest,
 		tarWriter:                tw,
@@ -374,7 +372,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		},
 		hookTracker: hook.NewHookTracker(),
 		volumeHelperImpl: volumehelper.NewVolumeHelperImpl(
-			resourcePolicy,
+			backupRequest.ResPolicies,
 			backupRequest.Spec.SnapshotVolumes,
 			log,
 			kb.kbClient,

pkg/cmd/util/output/backup_describer_test.go

@@ -235,9 +235,10 @@ Hooks:  <none>
   Excluded:  <none>
 
 Resources:
-  Included:        *
+  Included cluster-scoped:    <none>
-  Excluded:        <none>
+  Excluded cluster-scoped:    <none>
-  Cluster-scoped:  auto
+  Included namespace-scoped:  *
+  Excluded namespace-scoped:  <none>
 
 Label selector:  <none>
 
@@ -292,9 +293,10 @@ OrderedResources:
   Excluded:  <none>
 
 Resources:
-  Included:        *
+  Included cluster-scoped:    <none>
-  Excluded:        <none>
+  Excluded cluster-scoped:    <none>
-  Cluster-scoped:  auto
+  Included namespace-scoped:  *
+  Excluded namespace-scoped:  <none>
 
 Label selector:  <none>
 
@@ -325,9 +327,10 @@ Hooks:  <none>
   Excluded:  <none>
 
 Resources:
-  Included:        *
+  Included cluster-scoped:    <none>
-  Excluded:        <none>
+  Excluded cluster-scoped:    <none>
-  Cluster-scoped:  auto
+  Included namespace-scoped:  *
+  Excluded namespace-scoped:  <none>
 
 Label selector:  <none>
 

pkg/cmd/util/output/schedule_describe_test.go

@@ -32,9 +32,10 @@ Backup Template:
     Excluded:  <none>
   
   Resources:
-    Included:        *
+    Included cluster-scoped:    <none>
-    Excluded:        <none>
+    Excluded cluster-scoped:    <none>
-    Cluster-scoped:  auto
+    Included namespace-scoped:  *
+    Excluded namespace-scoped:  <none>
   
   Label selector:  <none>
   
@@ -81,9 +82,10 @@ Backup Template:
     Excluded:  <none>
   
   Resources:
-    Included:        *
+    Included cluster-scoped:    <none>
-    Excluded:        <none>
+    Excluded cluster-scoped:    <none>
-    Cluster-scoped:  auto
+    Included namespace-scoped:  *
+    Excluded namespace-scoped:  <none>
   
   Label selector:  <none>
   
@@ -127,9 +129,10 @@ Backup Template:
     Excluded:  <none>
   
   Resources:
-    Included:        *
+    Included cluster-scoped:    <none>
-    Excluded:        <none>
+    Excluded cluster-scoped:    <none>
-    Cluster-scoped:  auto
+    Included namespace-scoped:  *
+    Excluded namespace-scoped:  <none>
   
   Label selector:  <none>
   
@@ -174,9 +177,10 @@ Backup Template:
     Excluded:  <none>
   
   Resources:
-    Included:        *
+    Included cluster-scoped:    <none>
-    Excluded:        <none>
+    Excluded cluster-scoped:    <none>
-    Cluster-scoped:  auto
+    Included namespace-scoped:  *
+    Excluded namespace-scoped:  <none>
   
   Label selector:  <none>
   

pkg/controller/backup_controller.go

@@ -558,8 +558,11 @@ func (b *backupReconciler) prepareBackupRequest(backup *velerov1api.Backup, logg
 	if err != nil {
 		request.Status.ValidationErrors = append(request.Status.ValidationErrors, err.Error())
 	}
+	if resourcePolicies != nil && resourcePolicies.GetIncludeExcludePolicy() != nil && collections.UseOldResourceFilters(request.Spec) {
+		request.Status.ValidationErrors = append(request.Status.ValidationErrors, "include-resources, exclude-resources and include-cluster-resources are old filter parameters.\n"+
+			"They cannot be used with include-exclude policies.")
+	}
 	request.ResPolicies = resourcePolicies
-
 	return request
 }
 
@@ -812,7 +815,6 @@ func (b *backupReconciler) runBackup(backup *pkgbackup.Request) error {
 			fatalErrs = append(fatalErrs, errs...)
 		}
 	}
-
 	b.logger.WithField(constant.ControllerBackup, kubeutil.NamespaceAndName(backup)).Infof("Initial backup processing complete, moving to %s", backup.Status.Phase)
 
 	// if we return a non-nil error, the calling function will update

pkg/controller/backup_controller_test.go

@@ -702,10 +702,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.True(),
+					DefaultVolumesToFsBackup:         boolptr.True(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -740,10 +741,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          "alt-loc",
+					StorageLocation:                  "alt-loc",
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -782,10 +784,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          "read-write",
+					StorageLocation:                  "read-write",
-					DefaultVolumesToFsBackup: boolptr.True(),
+					DefaultVolumesToFsBackup:         boolptr.True(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -820,11 +823,12 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					TTL:                      metav1.Duration{Duration: 10 * time.Minute},
+					TTL:                              metav1.Duration{Duration: 10 * time.Minute},
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -860,10 +864,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.True(),
+					DefaultVolumesToFsBackup:         boolptr.True(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -900,10 +905,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -940,10 +946,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.True(),
+					DefaultVolumesToFsBackup:         boolptr.True(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -980,10 +987,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.True(),
+					DefaultVolumesToFsBackup:         boolptr.True(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -1020,10 +1028,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:          velerov1api.BackupPhaseFinalizing,
@@ -1061,10 +1070,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.True(),
+					DefaultVolumesToFsBackup:         boolptr.True(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:               velerov1api.BackupPhaseFailed,
@@ -1102,10 +1112,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.True(),
+					DefaultVolumesToFsBackup:         boolptr.True(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:               velerov1api.BackupPhaseFailed,
@@ -1143,10 +1154,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.True(),
+					SnapshotMoveData:                 boolptr.True(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:                       velerov1api.BackupPhaseFinalizing,
@@ -1185,10 +1197,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:                       velerov1api.BackupPhaseFinalizing,
@@ -1227,10 +1240,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:                       velerov1api.BackupPhaseFinalizing,
@@ -1269,10 +1283,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.True(),
+					SnapshotMoveData:                 boolptr.True(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:                       velerov1api.BackupPhaseFinalizing,
@@ -1312,10 +1327,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.False(),
+					SnapshotMoveData:                 boolptr.False(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:                       velerov1api.BackupPhaseFinalizing,
@@ -1354,10 +1370,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					},
 				},
 				Spec: velerov1api.BackupSpec{
-					StorageLocation:          defaultBackupLocation.Name,
+					StorageLocation:                  defaultBackupLocation.Name,
-					DefaultVolumesToFsBackup: boolptr.False(),
+					DefaultVolumesToFsBackup:         boolptr.False(),
-					SnapshotMoveData:         boolptr.True(),
+					SnapshotMoveData:                 boolptr.True(),
-					ExcludedResources:        append(autoExcludeNamespaceScopedResources, autoExcludeClusterScopedResources...),
+					ExcludedClusterScopedResources:   autoExcludeClusterScopedResources,
+					ExcludedNamespaceScopedResources: autoExcludeNamespaceScopedResources,
 				},
 				Status: velerov1api.BackupStatus{
 					Phase:                       velerov1api.BackupPhaseFinalizing,

pkg/util/collections/includes_excludes.go

@@ -19,6 +19,8 @@ package collections
 import (
 	"strings"
 
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
+
 	"github.com/gobwas/glob"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -105,14 +107,63 @@ func (ie *IncludesExcludes) ShouldInclude(s string) bool {
 	return ie.includes.Len() == 0 || ie.includes.Has("*") || ie.includes.match(s)
 }
 
+// IncludesString returns a string containing all of the includes, separated by commas, or * if the
+// list is empty.
+func (ie *IncludesExcludes) IncludesString() string {
+	return asString(ie.GetIncludes(), "*")
+}
+
+// ExcludesString returns a string containing all of the excludes, separated by commas, or <none> if the
+// list is empty.
+func (ie *IncludesExcludes) ExcludesString() string {
+	return asString(ie.GetExcludes(), "<none>")
+}
+
+// IncludeEverything returns true if the includes list is empty or '*'
+// and the excludes list is empty, or false otherwise.
+func (ie *IncludesExcludes) IncludeEverything() bool {
+	return ie.excludes.Len() == 0 && (ie.includes.Len() == 0 || (ie.includes.Len() == 1 && ie.includes.Has("*")))
+}
+
+// GetResourceIncludesExcludes takes the lists of resources to include and exclude, uses the
+// discovery helper to resolve them to fully-qualified group-resource names, and returns an
+// IncludesExcludes list.
+func GetResourceIncludesExcludes(helper discovery.Helper, includes, excludes []string) *IncludesExcludes {
+	resources := generateIncludesExcludes(
+		includes,
+		excludes,
+		func(item string) string {
+			gvr, _, err := helper.ResourceFor(schema.ParseGroupResource(item).WithVersion(""))
+			if err != nil {
+				// If we can't resolve it, return it as-is. This prevents the generated
+				// includes-excludes list from including *everything*, if none of the includes
+				// can be resolved. ref. https://github.com/vmware-tanzu/velero/issues/2461
+				return item
+			}
+
+			gr := gvr.GroupResource()
+			return gr.String()
+		},
+	)
+
+	return resources
+}
+
+func asString(in []string, empty string) string {
+	if len(in) == 0 {
+		return empty
+	}
+	return strings.Join(in, ", ")
+}
+
 // IncludesExcludesInterface is used as polymorphic IncludesExcludes for Global and scope
 // resources Include/Exclude.
 type IncludesExcludesInterface interface {
-	// Check whether the type name passed in by parameter should be included.
+	// ShouldInclude checks whether the type name passed in by parameter should be included.
 	// typeName should be k8s.io/apimachinery/pkg/runtime/schema GroupResource's String() result.
 	ShouldInclude(typeName string) bool
 
-	// Check whether the type name passed in by parameter should be excluded.
+	// ShouldExclude checks whether the type name passed in by parameter should be excluded.
 	// typeName should be k8s.io/apimachinery/pkg/runtime/schema GroupResource's String() result.
 	ShouldExclude(typeName string) bool
 }
@@ -188,6 +239,20 @@ func (ie *GlobalIncludesExcludes) ShouldExclude(typeName string) bool {
 	return false
 }
 
+func GetGlobalResourceIncludesExcludes(helper discovery.Helper, logger logrus.FieldLogger, includes, excludes []string, includeClusterResources *bool, nsIncludesExcludes IncludesExcludes) *GlobalIncludesExcludes {
+	ret := &GlobalIncludesExcludes{
+		resourceFilter:          *GetResourceIncludesExcludes(helper, includes, excludes),
+		includeClusterResources: includeClusterResources,
+		namespaceFilter:         nsIncludesExcludes,
+		helper:                  helper,
+		logger:                  logger,
+	}
+
+	logger.Infof("Including resources: %s", ret.resourceFilter.IncludesString())
+	logger.Infof("Excluding resources: %s", ret.resourceFilter.ExcludesString())
+	return ret
+}
+
 type ScopeIncludesExcludes struct {
 	namespaceScopedResourceFilter IncludesExcludes // namespace-scoped resource filter
 	clusterScopedResourceFilter   IncludesExcludes // cluster-scoped resource filter
@@ -259,29 +324,64 @@ func (ie *ScopeIncludesExcludes) ShouldExclude(typeName string) bool {
 	return false
 }
 
-// IncludesString returns a string containing all of the includes, separated by commas, or * if the
+func (ie *ScopeIncludesExcludes) CombineWithPolicy(policy *resourcepolicies.IncludeExcludePolicy) {
-// list is empty.
+	if policy == nil {
-func (ie *IncludesExcludes) IncludesString() string {
+		return
-	return asString(ie.GetIncludes(), "*")
-}
-
-// ExcludesString returns a string containing all of the excludes, separated by commas, or <none> if the
-// list is empty.
-func (ie *IncludesExcludes) ExcludesString() string {
-	return asString(ie.GetExcludes(), "<none>")
-}
-
-func asString(in []string, empty string) string {
-	if len(in) == 0 {
-		return empty
 	}
-	return strings.Join(in, ", ")
+	mapFunc := scopeResourceMapFunc(ie.helper)
-}
+	for _, item := range policy.ExcludedNamespaceScopedResources {
-
+		resolvedItem := mapFunc(item, true)
-// IncludeEverything returns true if the includes list is empty or '*'
+		if resolvedItem == "" {
-// and the excludes list is empty, or false otherwise.
+			continue
-func (ie *IncludesExcludes) IncludeEverything() bool {
+		}
-	return ie.excludes.Len() == 0 && (ie.includes.Len() == 0 || (ie.includes.Len() == 1 && ie.includes.Has("*")))
+		// The existing includeExcludes in the struct has higher priority, therefore, we should only add the item to the filter
+		// when the struct does not include this item and this item is not yet in the excludes filter.
+		if !ie.namespaceScopedResourceFilter.includes.match(resolvedItem) &&
+			!ie.namespaceScopedResourceFilter.excludes.match(resolvedItem) {
+			ie.namespaceScopedResourceFilter.Excludes(resolvedItem)
+		}
+	}
+	for _, item := range policy.IncludedNamespaceScopedResources {
+		resolvedItem := mapFunc(item, true)
+		if resolvedItem == "" {
+			continue
+		}
+		// The existing includeExcludes in the struct has higher priority, therefore, we should only add the item to the filter
+		// when the struct does not exclude this item and this item is not yet in the includes filter.
+		if !ie.namespaceScopedResourceFilter.includes.match(resolvedItem) &&
+			!ie.namespaceScopedResourceFilter.excludes.match(resolvedItem) {
+			ie.namespaceScopedResourceFilter.Includes(resolvedItem)
+		}
+	}
+	for _, item := range policy.ExcludedClusterScopedResources {
+		resolvedItem := mapFunc(item, false)
+		if resolvedItem == "" {
+			continue
+		}
+		if !ie.clusterScopedResourceFilter.includes.match(resolvedItem) &&
+			!ie.clusterScopedResourceFilter.excludes.match(resolvedItem) {
+			// The existing includeExcludes in the struct has higher priority, therefore, we should only add the item to the filter
+			// when the struct does not exclude this item and this item is not yet in the includes filter.
+			ie.clusterScopedResourceFilter.Excludes(resolvedItem)
+		}
+	}
+	for _, item := range policy.IncludedClusterScopedResources {
+		resolvedItem := mapFunc(item, false)
+		if resolvedItem == "" {
+			continue
+		}
+		if !ie.clusterScopedResourceFilter.includes.match(resolvedItem) &&
+			!ie.clusterScopedResourceFilter.excludes.match(resolvedItem) {
+			// The existing includeExcludes in the struct has higher priority, therefore, we should only add the item to the filter
+			// when the struct does not exclude this item and this item is not yet in the includes filter.
+			ie.clusterScopedResourceFilter.Includes(resolvedItem)
+		}
+	}
+	ie.logger.Infof("Scoped resource includes/excludes after combining with resource policy")
+	ie.logger.Infof("Including namespace-scoped resources: %s", ie.namespaceScopedResourceFilter.IncludesString())
+	ie.logger.Infof("Excluding namespace-scoped resources: %s", ie.namespaceScopedResourceFilter.ExcludesString())
+	ie.logger.Infof("Including cluster-scoped resources: %s", ie.clusterScopedResourceFilter.GetIncludes())
+	ie.logger.Infof("Excluding cluster-scoped resources: %s", ie.clusterScopedResourceFilter.ExcludesString())
 }
 
 func newScopeIncludesExcludes(nsIncludesExcludes IncludesExcludes, helper discovery.Helper, logger logrus.FieldLogger) *ScopeIncludesExcludes {
@@ -302,6 +402,43 @@ func newScopeIncludesExcludes(nsIncludesExcludes IncludesExcludes, helper discov
 	return ret
 }
 
+// GetScopeResourceIncludesExcludes function is similar with GetResourceIncludesExcludes,
+// but it's used for scoped Includes/Excludes, and can handle both cluster-scoped and namespace-scoped resources.
+func GetScopeResourceIncludesExcludes(helper discovery.Helper, logger logrus.FieldLogger, namespaceIncludes, namespaceExcludes, clusterIncludes, clusterExcludes []string, nsIncludesExcludes IncludesExcludes) *ScopeIncludesExcludes {
+	ret := generateScopedIncludesExcludes(
+		namespaceIncludes,
+		namespaceExcludes,
+		clusterIncludes,
+		clusterExcludes,
+		scopeResourceMapFunc(helper),
+		nsIncludesExcludes,
+		helper,
+		logger,
+	)
+	logger.Infof("Scoped resource includes/excludes after initialization")
+	logger.Infof("Including namespace-scoped resources: %s", ret.namespaceScopedResourceFilter.IncludesString())
+	logger.Infof("Excluding namespace-scoped resources: %s", ret.namespaceScopedResourceFilter.ExcludesString())
+	logger.Infof("Including cluster-scoped resources: %s", ret.clusterScopedResourceFilter.GetIncludes())
+	logger.Infof("Excluding cluster-scoped resources: %s", ret.clusterScopedResourceFilter.ExcludesString())
+
+	return ret
+}
+
+func scopeResourceMapFunc(helper discovery.Helper) func(string, bool) string {
+	return func(item string, namespaced bool) string {
+		gvr, resource, err := helper.ResourceFor(schema.ParseGroupResource(item).WithVersion(""))
+		if err != nil {
+			return item
+		}
+		if resource.Namespaced != namespaced {
+			return ""
+		}
+
+		gr := gvr.GroupResource()
+		return gr.String()
+	}
+}
+
 // ValidateIncludesExcludes checks provided lists of included and excluded
 // items to ensure they are a valid set of IncludesExcludes data.
 func ValidateIncludesExcludes(includesList, excludesList []string) []error {
@@ -470,97 +607,16 @@ func generateFilter(filter globStringSet, resources []string, mapFunc func(strin
 	}
 }
 
-// GetResourceIncludesExcludes takes the lists of resources to include and exclude, uses the
-// discovery helper to resolve them to fully-qualified group-resource names, and returns an
-// IncludesExcludes list.
-func GetResourceIncludesExcludes(helper discovery.Helper, includes, excludes []string) *IncludesExcludes {
-	resources := generateIncludesExcludes(
-		includes,
-		excludes,
-		func(item string) string {
-			gvr, _, err := helper.ResourceFor(schema.ParseGroupResource(item).WithVersion(""))
-			if err != nil {
-				// If we can't resolve it, return it as-is. This prevents the generated
-				// includes-excludes list from including *everything*, if none of the includes
-				// can be resolved. ref. https://github.com/vmware-tanzu/velero/issues/2461
-				return item
-			}
-
-			gr := gvr.GroupResource()
-			return gr.String()
-		},
-	)
-
-	return resources
-}
-
-func GetGlobalResourceIncludesExcludes(helper discovery.Helper, logger logrus.FieldLogger, includes, excludes []string, includeClusterResources *bool, nsIncludesExcludes IncludesExcludes) *GlobalIncludesExcludes {
-	ret := &GlobalIncludesExcludes{
-		resourceFilter:          *GetResourceIncludesExcludes(helper, includes, excludes),
-		includeClusterResources: includeClusterResources,
-		namespaceFilter:         nsIncludesExcludes,
-		helper:                  helper,
-		logger:                  logger,
-	}
-
-	logger.Infof("Including resources: %s", ret.resourceFilter.IncludesString())
-	logger.Infof("Excluding resources: %s", ret.resourceFilter.ExcludesString())
-	return ret
-}
-
-// GetScopeResourceIncludesExcludes function is similar with GetResourceIncludesExcludes,
-// but it's used for scoped Includes/Excludes, and can handle both cluster-scoped and namespace-scoped resources.
-func GetScopeResourceIncludesExcludes(helper discovery.Helper, logger logrus.FieldLogger, namespaceIncludes, namespaceExcludes, clusterIncludes, clusterExcludes []string, nsIncludesExcludes IncludesExcludes) *ScopeIncludesExcludes {
-	ret := generateScopedIncludesExcludes(
-		namespaceIncludes,
-		namespaceExcludes,
-		clusterIncludes,
-		clusterExcludes,
-		func(item string, namespaced bool) string {
-			gvr, resource, err := helper.ResourceFor(schema.ParseGroupResource(item).WithVersion(""))
-			if err != nil {
-				return item
-			}
-			if resource.Namespaced != namespaced {
-				return ""
-			}
-
-			gr := gvr.GroupResource()
-			return gr.String()
-		},
-		nsIncludesExcludes,
-		helper,
-		logger,
-	)
-	logger.Infof("Including namespace-scoped resources: %s", ret.namespaceScopedResourceFilter.IncludesString())
-	logger.Infof("Excluding namespace-scoped resources: %s", ret.namespaceScopedResourceFilter.ExcludesString())
-	logger.Infof("Including cluster-scoped resources: %s", ret.clusterScopedResourceFilter.GetIncludes())
-	logger.Infof("Excluding cluster-scoped resources: %s", ret.clusterScopedResourceFilter.ExcludesString())
-
-	return ret
-}
-
 // UseOldResourceFilters checks whether to use old resource filters (IncludeClusterResources,
 // IncludedResources and ExcludedResources), depending the backup's filters setting.
 // New filters are IncludedClusterScopedResources, ExcludedClusterScopedResources,
 // IncludedNamespaceScopedResources and ExcludedNamespaceScopedResources.
+// If all resource filters are none, it is treated as using new parameter filters.
 func UseOldResourceFilters(backupSpec velerov1api.BackupSpec) bool {
-	// If all resource filters are none, it is treated as using old parameter filters.
-	if backupSpec.IncludeClusterResources == nil &&
-		len(backupSpec.IncludedResources) == 0 &&
-		len(backupSpec.ExcludedResources) == 0 &&
-		len(backupSpec.IncludedClusterScopedResources) == 0 &&
-		len(backupSpec.ExcludedClusterScopedResources) == 0 &&
-		len(backupSpec.IncludedNamespaceScopedResources) == 0 &&
-		len(backupSpec.ExcludedNamespaceScopedResources) == 0 {
-		return true
-	}
-
 	if backupSpec.IncludeClusterResources != nil ||
 		len(backupSpec.IncludedResources) > 0 ||
 		len(backupSpec.ExcludedResources) > 0 {
 		return true
 	}
-
 	return false
 }

pkg/util/collections/includes_excludes_test.go

@@ -19,6 +19,8 @@ package collections
 import (
 	"testing"
 
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
+
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -741,6 +743,100 @@ func TestGetScopedResourceIncludesExcludes(t *testing.T) {
 	}
 }
 
+func TestScopeIncludesExcludes_CombineWithPolicy(t *testing.T) {
+	apiResources := []*test.APIResource{test.Deployments(), test.Pods(), test.ConfigMaps(), test.Secrets(), test.PVs(), test.CRDs(), test.ServiceAccounts()}
+	tests := []struct {
+		name                    string
+		namespaceScopedIncludes []string
+		namespaceScopedExcludes []string
+		clusterScopedIncludes   []string
+		clusterScopedExcludes   []string
+		policy                  *resourcepolicies.IncludeExcludePolicy
+		verify                  func(sie ScopeIncludesExcludes) bool
+	}{
+		{
+			name:                    "When policy is nil, the original includes excludes filters should not change",
+			namespaceScopedIncludes: []string{"deployments", "pods"},
+			namespaceScopedExcludes: []string{"configmaps"},
+			clusterScopedIncludes:   []string{"persistentvolumes"},
+			clusterScopedExcludes:   []string{"crds"},
+			policy:                  nil,
+			verify: func(sie ScopeIncludesExcludes) bool {
+				return sie.clusterScopedResourceFilter.ShouldInclude("persistentvolumes") &&
+					!sie.clusterScopedResourceFilter.ShouldInclude("crds") &&
+					sie.namespaceScopedResourceFilter.ShouldInclude("deployments") &&
+					!sie.namespaceScopedResourceFilter.ShouldInclude("configmaps")
+			},
+		},
+		{
+			name:                    "policy includes excludes should be merged to the original includes excludes when there's no conflict",
+			namespaceScopedIncludes: []string{"pods"},
+			namespaceScopedExcludes: []string{"configmaps"},
+			clusterScopedIncludes:   []string{},
+			clusterScopedExcludes:   []string{"crds"},
+			policy: &resourcepolicies.IncludeExcludePolicy{
+				IncludedNamespaceScopedResources: []string{"deployments"},
+				ExcludedNamespaceScopedResources: []string{"secrets"},
+				IncludedClusterScopedResources:   []string{"persistentvolumes"},
+				ExcludedClusterScopedResources:   []string{},
+			},
+			verify: func(sie ScopeIncludesExcludes) bool {
+				return sie.clusterScopedResourceFilter.ShouldInclude("persistentvolumes") &&
+					!sie.clusterScopedResourceFilter.ShouldInclude("crds") &&
+					sie.namespaceScopedResourceFilter.ShouldInclude("deployments") &&
+					!sie.namespaceScopedResourceFilter.ShouldInclude("configmaps") &&
+					!sie.namespaceScopedResourceFilter.ShouldInclude("secrets")
+			},
+		},
+		{
+			name:                    "when there are conflicts, the existing includes excludes filters have higher priorities",
+			namespaceScopedIncludes: []string{"pods", "deployments"},
+			namespaceScopedExcludes: []string{"configmaps"},
+			clusterScopedIncludes:   []string{"crds"},
+			clusterScopedExcludes:   []string{"persistentvolumes"},
+			policy: &resourcepolicies.IncludeExcludePolicy{
+				IncludedNamespaceScopedResources: []string{"configmaps"},
+				ExcludedNamespaceScopedResources: []string{"pods", "secrets"},
+				IncludedClusterScopedResources:   []string{"persistentvolumes"},
+				ExcludedClusterScopedResources:   []string{"crds"},
+			},
+			verify: func(sie ScopeIncludesExcludes) bool {
+				return sie.clusterScopedResourceFilter.ShouldInclude("crds") &&
+					!sie.clusterScopedResourceFilter.ShouldInclude("persistentvolumes") &&
+					sie.namespaceScopedResourceFilter.ShouldInclude("pods") &&
+					!sie.namespaceScopedResourceFilter.ShouldInclude("configmaps") &&
+					!sie.namespaceScopedResourceFilter.ShouldInclude("secrets")
+			},
+		},
+		{
+			name:                    "verify the case when there's '*' in the original include filter",
+			namespaceScopedIncludes: []string{"*"},
+			namespaceScopedExcludes: []string{},
+			clusterScopedIncludes:   []string{},
+			clusterScopedExcludes:   []string{},
+			policy: &resourcepolicies.IncludeExcludePolicy{
+				IncludedNamespaceScopedResources: []string{"deployments", "pods"},
+				ExcludedNamespaceScopedResources: []string{"configmaps", "secrets"},
+				IncludedClusterScopedResources:   []string{},
+				ExcludedClusterScopedResources:   []string{},
+			},
+			verify: func(sie ScopeIncludesExcludes) bool {
+				return sie.namespaceScopedResourceFilter.ShouldInclude("configmaps") &&
+					sie.namespaceScopedResourceFilter.ShouldInclude("secrets")
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			logger := logrus.StandardLogger()
+			discoveryHelper := setupDiscoveryClientWithResources(apiResources)
+			sie := GetScopeResourceIncludesExcludes(discoveryHelper, logger, tc.namespaceScopedIncludes, tc.namespaceScopedExcludes, tc.clusterScopedIncludes, tc.clusterScopedExcludes, *NewIncludesExcludes())
+			sie.CombineWithPolicy(tc.policy)
+			assert.True(t, tc.verify(*sie))
+		})
+	}
+}
+
 func TestUseOldResourceFilters(t *testing.T) {
 	tests := []struct {
 		name                  string
@@ -748,9 +844,9 @@ func TestUseOldResourceFilters(t *testing.T) {
 		useOldResourceFilters bool
 	}{
 		{
-			name:                  "backup with no filters should use old filters",
+			name:                  "backup with no filters should use new filters",
 			backup:                *defaultBackup().Result(),
-			useOldResourceFilters: true,
+			useOldResourceFilters: false,
 		},
 		{
 			name:                  "backup with only old filters should use old filters",