Merge branch 'main' into block-uploader-snapshot-operations

changelogs/unreleased/9928-chlins

@@ -0,0 +1 @@
+Add `--global-backup-volume-policies-configmap` server flag to configure cluster-wide global backup volume policies that are merged into every backup

changelogs/unreleased/9939-adam-jian-zhang

@@ -0,0 +1 @@
+Fix issue #9935, add resource policy on restore CRD

config/crd/v1/bases/velero.io_restores.yaml

@@ -404,6 +404,33 @@ spec:
                 - name
                 type: object
                 x-kubernetes-map-type: atomic
+              resourcePolicy:
+                description: |-
+                  ResourcePolicy specifies the reference to a ConfigMap containing resource
+                  filter policies for this restore. The ConfigMap can contain a
+                  namespacedFilterPolicies section that specifies per-namespace resource type
+                  filters, label selectors, and resource name patterns, and a
+                  clusterScopedFilterPolicy section for per-kind filtering of cluster-scoped
+                  resources. The ConfigMap format is the same as for BackupSpec.ResourcePolicy.
+                nullable: true
+                properties:
+                  apiGroup:
+                    description: |-
+                      APIGroup is the group for the resource being referenced.
+                      If APIGroup is not specified, the specified Kind must be in the core API group.
+                      For any other third-party types, APIGroup is required.
+                    type: string
+                  kind:
+                    description: Kind is the type of resource being referenced
+                    type: string
+                  name:
+                    description: Name is the name of resource being referenced
+                    type: string
+                required:
+                - kind
+                - name
+                type: object
+                x-kubernetes-map-type: atomic
               restorePVs:
                 description: |-
                   RestorePVs specifies whether to restore all included

internal/resourcepolicies/resource_policies.go

@@ -303,6 +303,30 @@ func (p *Policies) Validate() error {
 	return nil
 }
 
+func (p *Policies) ValidateForRestore() error {
+	if p.version != currentSupportDataVersion {
+		return fmt.Errorf("incompatible version number %s with supported version %s", p.version, currentSupportDataVersion)
+	}
+
+	if len(p.volumePolicies) > 0 {
+		return fmt.Errorf("volumePolicies are not supported for restore")
+	}
+
+	if p.GetIncludeExcludePolicy() != nil {
+		return fmt.Errorf("includeExcludePolicy is not supported for restore")
+	}
+
+	if err := p.validateClusterScopedFilterPolicy(); err != nil {
+		return errors.WithStack(err)
+	}
+
+	if err := p.validateNamespacedFilterPolicies(); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
 func (p *Policies) GetIncludeExcludePolicy() *IncludeExcludePolicy {
 	return p.includeExcludePolicy
 }
@@ -331,26 +355,143 @@ func GetResourcePoliciesFromBackup(
 		if err != nil {
 			logger.Errorf("Fail to get ResourcePolicies %s ConfigMap with error %s.",
 				backup.Namespace+"/"+backup.Spec.ResourcePolicy.Name, err.Error())
-			return nil, fmt.Errorf("fail to get ResourcePolicies %s ConfigMap with error %s",
-				backup.Namespace+"/"+backup.Spec.ResourcePolicy.Name, err.Error())
+			return nil, fmt.Errorf("fail to get ResourcePolicies %s ConfigMap: %w",
+				backup.Namespace+"/"+backup.Spec.ResourcePolicy.Name, err)
 		}
 		resourcePolicies, err = getResourcePoliciesFromConfig(policiesConfigMap)
 		if err != nil {
 			logger.Errorf("Fail to read ResourcePolicies from ConfigMap %s with error %s.",
 				backup.Namespace+"/"+backup.Name, err.Error())
-			return nil, fmt.Errorf("fail to read the ResourcePolicies from ConfigMap %s with error %s",
-				backup.Namespace+"/"+backup.Name, err.Error())
+			return nil, fmt.Errorf("fail to read the ResourcePolicies from ConfigMap %s: %w",
+				backup.Namespace+"/"+backup.Name, err)
 		} else if err = resourcePolicies.Validate(); err != nil {
 			logger.Errorf("Fail to validate ResourcePolicies in ConfigMap %s with error %s.",
 				backup.Namespace+"/"+backup.Name, err.Error())
-			return nil, fmt.Errorf("fail to validate ResourcePolicies in ConfigMap %s with error %s",
-				backup.Namespace+"/"+backup.Name, err.Error())
+			return nil, fmt.Errorf("fail to validate ResourcePolicies in ConfigMap %s: %w",
+				backup.Namespace+"/"+backup.Name, err)
 		}
 	}
 
 	return resourcePolicies, nil
 }
 
+// GetGlobalResourcePolicies loads and validates the cluster-wide global backup volume
+// policies from a ConfigMap in the Velero install namespace. Only the volumePolicies
+// section is honored globally; any include/exclude or fine-grained filter policies are
+// ignored (a warning is logged), as those are tied to a specific backup use case.
+func GetGlobalResourcePolicies(
+	client crclient.Client,
+	namespace string,
+	configMapName string,
+	logger logrus.FieldLogger,
+) (*Policies, error) {
+	cm := &corev1api.ConfigMap{}
+	if err := client.Get(context.Background(), crclient.ObjectKey{Namespace: namespace, Name: configMapName}, cm); err != nil {
+		return nil, fmt.Errorf("fail to get global backup volume policies ConfigMap %s/%s: %w", namespace, configMapName, err)
+	}
+
+	policies, err := getResourcePoliciesFromConfig(cm)
+	if err != nil {
+		return nil, fmt.Errorf("fail to read global backup volume policies from ConfigMap %s/%s: %w", namespace, configMapName, err)
+	}
+	if err := policies.Validate(); err != nil {
+		return nil, fmt.Errorf("fail to validate global backup volume policies in ConfigMap %s/%s: %w", namespace, configMapName, err)
+	}
+
+	// Only volumePolicies apply globally; warn about any other filter policies that will be ignored.
+	if policies.includeExcludePolicy != nil ||
+		policies.clusterScopedFilterPolicy != nil ||
+		len(policies.namespacedFilterPolicies) > 0 {
+		logger.Warnf("Global backup volume policies ConfigMap %s/%s contains include/exclude or fine-grained "+
+			"filter policies; these are ignored, only volumePolicies apply globally.", namespace, configMapName)
+	}
+
+	// Return a fresh Policies carrying only the globally-applicable fields. Using an allowlist here
+	// (rather than nil-ing out the ignored fields) means any filter field added to Policies in the
+	// future is excluded from the global policies by default, without needing to update this code.
+	return &Policies{
+		version:        policies.version,
+		volumePolicies: policies.volumePolicies,
+	}, nil
+}
+
+// GetResourcePoliciesFromBackupWithGlobal builds the effective resource policies for a backup
+// by merging the backup-referenced resource policies with the global backup volume policies
+// (when globalConfigMapName is set). The merged volumePolicies list is the backup-level
+// policies followed by the global ones, so the first match wins and a backup can override the
+// global baseline for a specific volume while still inheriting the rest of the global rules.
+func GetResourcePoliciesFromBackupWithGlobal(
+	backup velerov1api.Backup,
+	client crclient.Client,
+	globalConfigMapName string,
+	installNamespace string,
+	logger logrus.FieldLogger,
+) (*Policies, error) {
+	backupPolicies, err := GetResourcePoliciesFromBackup(backup, client, logger)
+	if err != nil {
+		return nil, err
+	}
+
+	if globalConfigMapName == "" {
+		return backupPolicies, nil
+	}
+
+	globalPolicies, err := GetGlobalResourcePolicies(client, installNamespace, globalConfigMapName, logger)
+	if err != nil {
+		return nil, err
+	}
+
+	if backupPolicies == nil {
+		return globalPolicies, nil
+	}
+	// Backup-level policies first, then global, so backups can override the global baseline.
+	backupPolicies.volumePolicies = append(backupPolicies.volumePolicies, globalPolicies.volumePolicies...)
+	return backupPolicies, nil
+}
+
+// GetResourcePoliciesFromRestore retrieves the resource policies from the ConfigMap referenced in the Restore spec.
+func GetResourcePoliciesFromRestore(
+	ctx context.Context,
+	restore *velerov1api.Restore,
+	client crclient.Client,
+	logger logrus.FieldLogger,
+) (resourcePolicies *Policies, err error) {
+	if restore.Spec.ResourcePolicy != nil {
+		if !strings.EqualFold(restore.Spec.ResourcePolicy.Kind, ConfigmapRefType) {
+			return nil, fmt.Errorf("invalid ResourcePolicy kind %q, only %q is supported",
+				restore.Spec.ResourcePolicy.Kind, ConfigmapRefType)
+		}
+		policiesConfigMap := &corev1api.ConfigMap{}
+		err = client.Get(
+			ctx,
+			crclient.ObjectKey{
+				Namespace: restore.Namespace,
+				Name:      restore.Spec.ResourcePolicy.Name,
+			},
+			policiesConfigMap,
+		)
+		if err != nil {
+			logger.Errorf("Fail to get ResourcePolicies %s ConfigMap with error %s.",
+				restore.Namespace+"/"+restore.Spec.ResourcePolicy.Name, err.Error())
+			return nil, fmt.Errorf("fail to get ResourcePolicies %s ConfigMap: %w",
+				restore.Namespace+"/"+restore.Spec.ResourcePolicy.Name, err)
+		}
+		resourcePolicies, err = getResourcePoliciesFromConfig(policiesConfigMap)
+		if err != nil {
+			logger.Errorf("Fail to read ResourcePolicies from ConfigMap %s with error %s.",
+				restore.Namespace+"/"+restore.Spec.ResourcePolicy.Name, err.Error())
+			return nil, fmt.Errorf("fail to read the ResourcePolicies from ConfigMap %s: %w",
+				restore.Namespace+"/"+restore.Spec.ResourcePolicy.Name, err)
+		} else if err = resourcePolicies.ValidateForRestore(); err != nil {
+			logger.Errorf("Fail to validate ResourcePolicies in ConfigMap %s with error %s.",
+				restore.Namespace+"/"+restore.Spec.ResourcePolicy.Name, err.Error())
+			return nil, fmt.Errorf("fail to validate ResourcePolicies in ConfigMap %s: %w",
+				restore.Namespace+"/"+restore.Spec.ResourcePolicy.Name, err)
+		}
+	}
+	return resourcePolicies, nil
+}
+
 func getResourcePoliciesFromConfig(cm *corev1api.ConfigMap) (*Policies, error) {
 	if cm == nil {
 		return nil, fmt.Errorf("could not parse config from nil configmap")

internal/resourcepolicies/resource_policies_test.go

@@ -16,13 +16,20 @@ limitations under the License.
 package resourcepolicies
 
 import (
+	"context"
 	"testing"
 
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1api "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 )
 
 func pvcVolumeMode(mode corev1api.PersistentVolumeMode) *corev1api.PersistentVolumeMode {
@@ -205,6 +212,18 @@ volumePolicies:
       pvcAccessModes: ReadWriteOnce
     action:
       type: skip
+`,
+			wantErr: true,
+		},
+		{
+			name: "error format of pvcAccessModes (list with non-string)",
+			yamlData: `version: v1
+volumePolicies:
+  - conditions:
+      pvcAccessModes:
+        - 123
+    action:
+      type: skip
 `,
 			wantErr: true,
 		},
@@ -403,14 +422,20 @@ func TestGetResourceMatchedAction(t *testing.T) {
 }
 
 func TestGetResourcePoliciesFromConfig(t *testing.T) {
-	// Create a test ConfigMap
-	cm := &corev1api.ConfigMap{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "test-configmap",
-			Namespace: "test-namespace",
-		},
-		Data: map[string]string{
-			"test-data": `version: v1
+	testCases := []struct {
+		name        string
+		cm          *corev1api.ConfigMap
+		expectedErr string
+	}{
+		{
+			name: "valid configmap",
+			cm: &corev1api.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-configmap",
+					Namespace: "test-namespace",
+				},
+				Data: map[string]string{
+					"test-data": `version: v1
 volumePolicies:
   - conditions:
       capacity: '0,10Gi'
@@ -431,63 +456,457 @@ volumePolicies:
     action:
       type: skip
 `,
+				},
+			},
+			expectedErr: "",
+		},
+		{
+			name:        "nil configmap",
+			cm:          nil,
+			expectedErr: "could not parse config from nil configmap",
+		},
+		{
+			name: "empty data configmap",
+			cm: &corev1api.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-configmap",
+					Namespace: "test-namespace",
+				},
+				Data: map[string]string{},
+			},
+			expectedErr: "illegal resource policies test-namespace/test-configmap configmap",
+		},
+		{
+			name: "multiple data configmap",
+			cm: &corev1api.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-configmap",
+					Namespace: "test-namespace",
+				},
+				Data: map[string]string{
+					"data1": "value1",
+					"data2": "value2",
+				},
+			},
+			expectedErr: "illegal resource policies test-namespace/test-configmap configmap",
+		},
+		{
+			name: "invalid yaml data",
+			cm: &corev1api.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-configmap",
+					Namespace: "test-namespace",
+				},
+				Data: map[string]string{
+					"test-data": `version: v1
+volumePolicies:
+  - conditions:
+      capacity: '0,10Gi'
+      csi:
+        driver: disks.csi.driver
+    action:
+      type: skip
+    invalid-key: value
+`,
+				},
+			},
+			expectedErr: "failed to decode yaml data into resource policies",
+		},
+		{
+			name: "build policy error",
+			cm: &corev1api.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-configmap",
+					Namespace: "test-namespace",
+				},
+				Data: map[string]string{
+					"test-data": `version: v1
+volumePolicies:
+  - conditions:
+      capacity: 'invalid-capacity'
+      csi:
+        driver: disks.csi.driver
+    action:
+      type: skip
+`,
+				},
+			},
+			expectedErr: "wrong format of Capacity invalid-capacity",
 		},
 	}
 
-	// Call the function and check for errors
-	resPolicies, err := getResourcePoliciesFromConfig(cm)
-	require.NoError(t, err)
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			resPolicies, err := getResourcePoliciesFromConfig(tc.cm)
+			if tc.expectedErr == "" {
+				require.NoError(t, err)
+				assert.Equal(t, "v1", resPolicies.version)
+				assert.Len(t, resPolicies.volumePolicies, 3)
+			} else {
+				require.ErrorContains(t, err, tc.expectedErr)
+				assert.Nil(t, resPolicies)
+			}
+		})
+	}
+}
 
-	// Check that the returned resourcePolicies object contains the expected data
-	assert.Equal(t, "v1", resPolicies.version)
-
-	assert.Len(t, resPolicies.volumePolicies, 3)
-
-	policies := ResourcePolicies{
-		Version: "v1",
-		VolumePolicies: []VolumePolicy{
-			{
-				Conditions: map[string]any{
-					"capacity": "0,10Gi",
-					"csi": map[string]any{
-						"driver": "disks.csi.driver",
-					},
-				},
-				Action: Action{
-					Type: Skip,
-				},
-			},
-			{
-				Conditions: map[string]any{
-					"csi": map[string]any{
-						"driver":           "files.csi.driver",
-						"volumeAttributes": map[string]string{"protocol": "nfs"},
-					},
-				},
-				Action: Action{
-					Type: Skip,
-				},
-			},
-			{
-				Conditions: map[string]any{
-					"pvcLabels": map[string]string{
-						"environment": "production",
-					},
-				},
-				Action: Action{
-					Type: Skip,
-				},
-			},
+func TestGetResourcePoliciesFromBackup(t *testing.T) {
+	validCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"test-data": `version: v1
+volumePolicies:
+  - conditions:
+      capacity: '0,10Gi'
+      csi:
+        driver: disks.csi.driver
+    action:
+      type: skip
+`,
 		},
 	}
 
-	p := &Policies{}
-	err = p.BuildPolicy(&policies)
-	if err != nil {
-		t.Fatalf("failed to build policy: %v", err)
+	invalidActionCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "invalid-action-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"test-data": `version: v1
+volumePolicies:
+  - conditions:
+      capacity: '0,10Gi'
+      csi:
+        driver: disks.csi.driver
+    action:
+      type: invalid-action
+`,
+		},
 	}
 
-	assert.Equal(t, p, resPolicies)
+	invalidVersionCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "invalid-version-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"test-data": `version: v2
+volumePolicies:
+  - conditions:
+      capacity: '0,10Gi'
+      csi:
+        driver: disks.csi.driver
+    action:
+      type: skip
+`,
+		},
+	}
+
+	emptyCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "empty-configmap",
+			Namespace: "test-namespace",
+		},
+	}
+
+	client := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(validCM, invalidActionCM, invalidVersionCM, emptyCM).Build()
+	logger := logrus.New()
+
+	testCases := []struct {
+		name        string
+		backup      velerov1api.Backup
+		expectedErr string
+	}{
+		{
+			name: "valid configmap",
+			backup: velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-backup",
+				},
+				Spec: velerov1api.BackupSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "test-configmap",
+					},
+				},
+			},
+			expectedErr: "",
+		},
+		{
+			name: "invalid kind",
+			backup: velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-backup",
+				},
+				Spec: velerov1api.BackupSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: "Secret",
+						Name: "test-configmap",
+					},
+				},
+			},
+			expectedErr: "",
+		},
+		{
+			name: "configmap not found",
+			backup: velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-backup",
+				},
+				Spec: velerov1api.BackupSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "non-existent-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to get ResourcePolicies test-namespace/non-existent-configmap ConfigMap",
+		},
+		{
+			name: "invalid action configmap",
+			backup: velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-backup",
+				},
+				Spec: velerov1api.BackupSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "invalid-action-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to validate ResourcePolicies in ConfigMap test-namespace/test-backup",
+		},
+		{
+			name: "invalid version configmap",
+			backup: velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-backup",
+				},
+				Spec: velerov1api.BackupSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "invalid-version-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to validate ResourcePolicies in ConfigMap test-namespace/test-backup",
+		},
+		{
+			name: "empty configmap",
+			backup: velerov1api.Backup{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-backup",
+				},
+				Spec: velerov1api.BackupSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "empty-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to read the ResourcePolicies from ConfigMap test-namespace/test-backup",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			resPolicies, err := GetResourcePoliciesFromBackup(tc.backup, client, logger)
+			if tc.expectedErr == "" {
+				require.NoError(t, err)
+				if tc.backup.Spec.ResourcePolicy != nil && tc.backup.Spec.ResourcePolicy.Kind == ConfigmapRefType {
+					assert.NotNil(t, resPolicies)
+				} else {
+					assert.Nil(t, resPolicies)
+				}
+			} else {
+				require.ErrorContains(t, err, tc.expectedErr)
+				assert.Nil(t, resPolicies)
+			}
+		})
+	}
+}
+
+func TestGetResourcePoliciesFromRestore(t *testing.T) {
+	validCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"test-data": `version: v1
+namespacedFilterPolicies:
+  - namespaces: ["default"]
+    resourceFilters:
+      - kinds: ["Pod"]
+`,
+		},
+	}
+
+	invalidNfpCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "invalid-action-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"test-data": `version: v1
+namespacedFilterPolicies:
+  - namespaces: []
+    resourceFilters:
+      - kinds: ["Pod"]
+`,
+		},
+	}
+
+	invalidVersionCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "invalid-version-configmap",
+			Namespace: "test-namespace",
+		},
+		Data: map[string]string{
+			"test-data": `version: v2
+namespacedFilterPolicies:
+  - namespaces: ["default"]
+    resourceFilters:
+      - kinds: ["Pod"]
+`,
+		},
+	}
+
+	emptyCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "empty-configmap",
+			Namespace: "test-namespace",
+		},
+	}
+
+	client := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithObjects(validCM, invalidNfpCM, invalidVersionCM, emptyCM).Build()
+	logger := logrus.New()
+
+	testCases := []struct {
+		name        string
+		restore     *velerov1api.Restore
+		expectedErr string
+	}{
+		{
+			name: "valid configmap",
+			restore: &velerov1api.Restore{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-restore",
+				},
+				Spec: velerov1api.RestoreSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "test-configmap",
+					},
+				},
+			},
+			expectedErr: "",
+		},
+		{
+			name: "invalid kind",
+			restore: &velerov1api.Restore{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-restore",
+				},
+				Spec: velerov1api.RestoreSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: "Secret",
+						Name: "test-configmap",
+					},
+				},
+			},
+			expectedErr: "invalid ResourcePolicy kind \"Secret\", only \"configmap\" is supported",
+		},
+		{
+			name: "configmap not found",
+			restore: &velerov1api.Restore{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-restore",
+				},
+				Spec: velerov1api.RestoreSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "non-existent-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to get ResourcePolicies test-namespace/non-existent-configmap ConfigMap",
+		},
+		{
+			name: "invalid action configmap",
+			restore: &velerov1api.Restore{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-restore",
+				},
+				Spec: velerov1api.RestoreSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "invalid-action-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to validate ResourcePolicies in ConfigMap test-namespace/invalid-action-configmap",
+		},
+		{
+			name: "invalid version configmap",
+			restore: &velerov1api.Restore{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-restore",
+				},
+				Spec: velerov1api.RestoreSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "invalid-version-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to validate ResourcePolicies in ConfigMap test-namespace/invalid-version-configmap",
+		},
+		{
+			name: "empty configmap",
+			restore: &velerov1api.Restore{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "test-namespace",
+					Name:      "test-restore",
+				},
+				Spec: velerov1api.RestoreSpec{
+					ResourcePolicy: &corev1api.TypedLocalObjectReference{
+						Kind: ConfigmapRefType,
+						Name: "empty-configmap",
+					},
+				},
+			},
+			expectedErr: "fail to read the ResourcePolicies from ConfigMap test-namespace/empty-configmap",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			resPolicies, err := GetResourcePoliciesFromRestore(context.Background(), tc.restore, client, logger)
+			if tc.expectedErr == "" {
+				require.NoError(t, err)
+				assert.NotNil(t, resPolicies)
+			} else {
+				require.ErrorContains(t, err, tc.expectedErr)
+				assert.Nil(t, resPolicies)
+			}
+		})
+	}
 }
 
 func TestGetMatchAction(t *testing.T) {
@@ -1784,6 +2203,17 @@ namespacedFilterPolicies:
 			wantErr: true,
 			errMsg:  "invalid glob pattern",
 		},
+		{
+			name: "invalid - bad glob pattern in excludedNames",
+			yamlData: `version: v1
+namespacedFilterPolicies:
+- namespaces: ["test"]
+  resourceFilters:
+  - kinds: ["Pod"]
+    excludedNames: ["[invalid"]`,
+			wantErr: true,
+			errMsg:  "invalid glob pattern",
+		},
 		{
 			name: "invalid - duplicate namespace pattern",
 			yamlData: `version: v1
@@ -1797,6 +2227,16 @@ namespacedFilterPolicies:
 			wantErr: true,
 			errMsg:  "duplicate namespace pattern",
 		},
+		{
+			name: "invalid - bad namespace pattern",
+			yamlData: `version: v1
+namespacedFilterPolicies:
+- namespaces: ["prod**uction"]
+  resourceFilters:
+  - kinds: ["Pod"]`,
+			wantErr: true,
+			errMsg:  "wildcard pattern contains consecutive asterisks",
+		},
 	}
 
 	for _, tc := range testCases {
@@ -1853,6 +2293,50 @@ namespacedFilterPolicies:
 	assert.Equal(t, map[string]string{"app": "web"}, rf.LabelSelector)
 }
 
+func TestClusterScopedFilterPoliciesAccessor(t *testing.T) {
+	yamlData := `version: v1
+clusterScopedFilterPolicy:
+  resourceFilters:
+  - kinds: ["ClusterRole"]
+    names: ["my-app-*"]`
+
+	resPolicies, err := unmarshalResourcePolicies(&yamlData)
+	require.NoError(t, err)
+
+	policies := &Policies{}
+	err = policies.BuildPolicy(resPolicies)
+	require.NoError(t, err)
+
+	csfPolicy := policies.GetClusterScopedFilterPolicy()
+	require.NotNil(t, csfPolicy)
+	assert.Len(t, csfPolicy.ResourceFilters, 1)
+
+	rf := csfPolicy.ResourceFilters[0]
+	assert.Equal(t, []string{"ClusterRole"}, rf.Kinds)
+	assert.Equal(t, []string{"my-app-*"}, rf.Names)
+}
+
+func TestIncludeExcludePolicyAccessor(t *testing.T) {
+	yamlData := `version: v1
+includeExcludePolicy:
+  includedClusterScopedResources:
+  - ClusterRole
+  excludedClusterScopedResources:
+  - ClusterRoleBinding`
+
+	resPolicies, err := unmarshalResourcePolicies(&yamlData)
+	require.NoError(t, err)
+
+	policies := &Policies{}
+	err = policies.BuildPolicy(resPolicies)
+	require.NoError(t, err)
+
+	iePolicy := policies.GetIncludeExcludePolicy()
+	require.NotNil(t, iePolicy)
+	assert.Equal(t, []string{"ClusterRole"}, iePolicy.IncludedClusterScopedResources)
+	assert.Equal(t, []string{"ClusterRoleBinding"}, iePolicy.ExcludedClusterScopedResources)
+}
+
 func TestFirstMatchSemantics(t *testing.T) {
 	yamlData := `version: v1
 namespacedFilterPolicies:
@@ -2172,3 +2656,192 @@ func TestPVCAccessModesMatch(t *testing.T) {
 		})
 	}
 }
+
+// ---- Global backup volume policies ----
+
+func globalPolicyConfigMap(name, data string) *corev1api.ConfigMap {
+	return &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: name},
+		Data:       map[string]string{"policies.yaml": data},
+	}
+}
+
+func backupWithPolicy(ref string) velerov1api.Backup {
+	b := velerov1api.Backup{ObjectMeta: metav1.ObjectMeta{Namespace: "velero", Name: "backup"}}
+	if ref != "" {
+		b.Spec.ResourcePolicy = &corev1api.TypedLocalObjectReference{Kind: ConfigmapRefType, Name: ref}
+	}
+	return b
+}
+
+// firstActionFor returns the action type the policies select for a PV with the given storage
+// class, or "" when nothing matches. It exercises the compiled match logic so the tests verify
+// merge ordering rather than internal field layout.
+func firstActionFor(p *Policies, storageClass string) VolumeActionType {
+	pv := &corev1api.PersistentVolume{Spec: corev1api.PersistentVolumeSpec{StorageClassName: storageClass}}
+	vol := &structuredVolume{}
+	vol.parsePV(pv)
+	if a := p.match(vol); a != nil {
+		return a.Type
+	}
+	return ""
+}
+
+func TestGetResourcePoliciesFromBackupWithGlobal(t *testing.T) {
+	gp2Skip := `version: v1
+volumePolicies:
+  - conditions:
+      storageClass:
+        - gp2
+    action:
+      type: skip
+`
+	gp2Snapshot := `version: v1
+volumePolicies:
+  - conditions:
+      storageClass:
+        - gp2
+    action:
+      type: snapshot
+`
+	otherFsBackup := `version: v1
+volumePolicies:
+  - conditions:
+      storageClass:
+        - other
+    action:
+      type: fs-backup
+`
+
+	tests := []struct {
+		name                string
+		backupCM            *corev1api.ConfigMap
+		globalCMName        string
+		globalCM            *corev1api.ConfigMap
+		backupRef           string
+		expectErr           bool
+		expectedGp2Action   VolumeActionType
+		expectedNumPolicies int
+	}{
+		{
+			name:                "no global, backup only - unchanged behavior",
+			backupRef:           "backup01",
+			backupCM:            globalPolicyConfigMap("backup01", gp2Snapshot),
+			expectedGp2Action:   Snapshot,
+			expectedNumPolicies: 1,
+		},
+		{
+			name:                "global only, backup has no policy",
+			globalCMName:        "global",
+			globalCM:            globalPolicyConfigMap("global", gp2Skip),
+			expectedGp2Action:   Skip,
+			expectedNumPolicies: 1,
+		},
+		{
+			name:                "no global configured and no backup policy",
+			expectedGp2Action:   "",
+			expectedNumPolicies: 0,
+		},
+		{
+			name:                "merge - backup policy overrides global for gp2",
+			backupRef:           "backup01",
+			backupCM:            globalPolicyConfigMap("backup01", gp2Snapshot),
+			globalCMName:        "global",
+			globalCM:            globalPolicyConfigMap("global", gp2Skip),
+			expectedGp2Action:   Snapshot, // backup-level wins (evaluated first)
+			expectedNumPolicies: 2,
+		},
+		{
+			name:                "merge - backup inherits non-overlapping global rule",
+			backupRef:           "backup01",
+			backupCM:            globalPolicyConfigMap("backup01", otherFsBackup),
+			globalCMName:        "global",
+			globalCM:            globalPolicyConfigMap("global", gp2Skip),
+			expectedGp2Action:   Skip, // only global matches gp2
+			expectedNumPolicies: 2,
+		},
+		{
+			name:         "global configmap missing - error",
+			globalCMName: "global",
+			expectErr:    true,
+		},
+		{
+			name:         "global configmap invalid - error",
+			globalCMName: "global",
+			globalCM:     globalPolicyConfigMap("global", "not: [valid"),
+			expectErr:    true,
+		},
+		{
+			// Parses cleanly but fails Policies.Validate() due to the unsupported version.
+			name:         "global configmap fails validation - error",
+			globalCMName: "global",
+			globalCM:     globalPolicyConfigMap("global", "version: v2\nvolumePolicies: []\n"),
+			expectErr:    true,
+		},
+		{
+			// Backup references a ResourcePolicy ConfigMap that does not exist, so resolving the
+			// backup-level policies fails before the global ones are consulted.
+			name:         "backup configmap missing - error",
+			backupRef:    "missing-backup-cm",
+			globalCMName: "global",
+			globalCM:     globalPolicyConfigMap("global", gp2Skip),
+			expectErr:    true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			client := velerotest.NewFakeControllerRuntimeClient(t)
+			if tc.backupCM != nil {
+				require.NoError(t, client.Create(t.Context(), tc.backupCM))
+			}
+			if tc.globalCM != nil {
+				require.NoError(t, client.Create(t.Context(), tc.globalCM))
+			}
+
+			b := backupWithPolicy(tc.backupRef)
+
+			p, err := GetResourcePoliciesFromBackupWithGlobal(b, client, tc.globalCMName, "velero", logrus.New())
+			if tc.expectErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+
+			if tc.expectedNumPolicies == 0 {
+				assert.Nil(t, p)
+				return
+			}
+			require.NotNil(t, p)
+			assert.Len(t, p.volumePolicies, tc.expectedNumPolicies)
+			assert.Equal(t, tc.expectedGp2Action, firstActionFor(p, "gp2"))
+		})
+	}
+}
+
+func TestGetGlobalResourcePoliciesIgnoresNonVolumePolicies(t *testing.T) {
+	data := `version: v1
+volumePolicies:
+  - conditions:
+      storageClass:
+        - gp2
+    action:
+      type: skip
+namespacedFilterPolicies:
+- namespaces: ["frontend"]
+  resourceFilters:
+  - kinds: ["Pod"]
+`
+	client := velerotest.NewFakeControllerRuntimeClient(t)
+	require.NoError(t, client.Create(t.Context(), globalPolicyConfigMap("global", data)))
+
+	p, err := GetGlobalResourcePolicies(client, "velero", "global", logrus.New())
+	require.NoError(t, err)
+	require.NotNil(t, p)
+
+	// Only volumePolicies are kept; the namespaced filter policy is dropped.
+	assert.Len(t, p.volumePolicies, 1)
+	assert.Empty(t, p.GetNamespacedFilterPolicies())
+	assert.Nil(t, p.GetIncludeExcludePolicy())
+	assert.Nil(t, p.GetClusterScopedFilterPolicy())
+}

internal/resourcepolicies/volume_resources_validator_test.go

@@ -568,3 +568,85 @@ func TestValidate(t *testing.T) {
 		})
 	}
 }
+
+func TestValidateForRestore(t *testing.T) {
+	testCases := []struct {
+		name    string
+		res     *ResourcePolicies
+		wantErr bool
+	}{
+		{
+			name: "valid restore policies",
+			res: &ResourcePolicies{
+				Version: "v1",
+				ClusterScopedFilterPolicy: &ClusterScopedFilterPolicy{
+					ResourceFilters: []ResourceFilter{
+						{
+							Kinds: []string{"ClusterRole"},
+						},
+					},
+				},
+				NamespacedFilterPolicies: []NamespacedFilterPolicy{
+					{
+						Namespaces: []string{"default"},
+						ResourceFilters: []ResourceFilter{
+							{
+								Kinds: []string{"Pod"},
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "unsupported volumePolicies for restore",
+			res: &ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []VolumePolicy{
+					{
+						Action: Action{Type: "skip"},
+						Conditions: map[string]any{
+							"capacity": "10Gi",
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "unsupported includeExcludePolicy for restore",
+			res: &ResourcePolicies{
+				Version: "v1",
+				IncludeExcludePolicy: &IncludeExcludePolicy{
+					IncludedClusterScopedResources: []string{"persistentvolumes"},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "wrong version",
+			res: &ResourcePolicies{
+				Version: "v2",
+			},
+			wantErr: true,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			policies := &Policies{}
+			err1 := policies.BuildPolicy(tc.res)
+			err2 := policies.ValidateForRestore()
+
+			if tc.wantErr {
+				if err1 == nil && err2 == nil {
+					t.Fatalf("Expected error %v, but not get error", tc.wantErr)
+				}
+			} else {
+				if err1 != nil || err2 != nil {
+					t.Fatalf("Expected error %v, but got error %v %v", tc.wantErr, err1, err2)
+				}
+			}
+		})
+	}
+}

pkg/apis/velero/v1/labels_annotations.go

@@ -80,6 +80,11 @@ const (
 	// timeout value for backup to plugins.
 	ResourceTimeoutAnnotation = "velero.io/resource-timeout"
 
+	// GlobalBackupVolumePolicyConfigMapAnnotation is the annotation key used to record the
+	// name of the cluster-wide global backup volume policies ConfigMap that contributed to a
+	// backup, so that `velero backup describe` can surface it.
+	GlobalBackupVolumePolicyConfigMapAnnotation = "velero.io/global-backup-volume-policy-configmap"
+
 	// AsyncOperationIDLabel is the label key used to identify the async operation ID
 	AsyncOperationIDLabel = "velero.io/async-operation-id"
 

pkg/apis/velero/v1/restore_types.go

@@ -125,6 +125,16 @@ type RestoreSpec struct {
 	// +nullable
 	ResourceModifier *corev1api.TypedLocalObjectReference `json:"resourceModifier,omitempty"`
 
+	// ResourcePolicy specifies the reference to a ConfigMap containing resource
+	// filter policies for this restore. The ConfigMap can contain a
+	// namespacedFilterPolicies section that specifies per-namespace resource type
+	// filters, label selectors, and resource name patterns, and a
+	// clusterScopedFilterPolicy section for per-kind filtering of cluster-scoped
+	// resources. The ConfigMap format is the same as for BackupSpec.ResourcePolicy.
+	// +optional
+	// +nullable
+	ResourcePolicy *corev1api.TypedLocalObjectReference `json:"resourcePolicy,omitempty"`
+
 	// UploaderConfig specifies the configuration for the restore.
 	// +optional
 	// +nullable

pkg/apis/velero/v1/zz_generated.deepcopy.go

@@ -1415,6 +1415,11 @@ func (in *RestoreSpec) DeepCopyInto(out *RestoreSpec) {
 		*out = new(corev1.TypedLocalObjectReference)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.ResourcePolicy != nil {
+		in, out := &in.ResourcePolicy, &out.ResourcePolicy
+		*out = new(corev1.TypedLocalObjectReference)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.UploaderConfig != nil {
 		in, out := &in.UploaderConfig, &out.UploaderConfig
 		*out = new(UploaderConfigForRestore)

pkg/builder/restore_builder.go

@@ -19,6 +19,7 @@ package builder
 import (
 	"time"
 
+	corev1api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
@@ -171,3 +172,12 @@ func (b *RestoreBuilder) ItemOperationTimeout(timeout time.Duration) *RestoreBui
 	b.object.Spec.ItemOperationTimeout.Duration = timeout
 	return b
 }
+
+// ResourcePoliciesConfigmap sets the Restore's resource policies configmap.
+func (b *RestoreBuilder) ResourcePoliciesConfigmap(name string) *RestoreBuilder {
+	b.object.Spec.ResourcePolicy = &corev1api.TypedLocalObjectReference{
+		Kind: "configmap",
+		Name: name,
+	}
+	return b
+}

pkg/builder/restore_builder_test.go

@@ -0,0 +1,36 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+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 builder
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRestoreBuilder_ResourcePoliciesConfigmap(t *testing.T) {
+	restore := ForRestore("velero", "my-restore").
+		ResourcePoliciesConfigmap("my-policy-cm").
+		Result()
+
+	assert.Equal(t, "velero", restore.Namespace)
+	assert.Equal(t, "my-restore", restore.Name)
+	assert.NotNil(t, restore.Spec.ResourcePolicy)
+	assert.Equal(t, "configmap", restore.Spec.ResourcePolicy.Kind)
+	assert.Equal(t, "my-policy-cm", restore.Spec.ResourcePolicy.Name)
+	assert.Equal(t, (*string)(nil), restore.Spec.ResourcePolicy.APIGroup)
+}

pkg/cmd/server/config/config.go

@@ -145,42 +145,43 @@ var (
 )
 
 type Config struct {
-	PluginDir                      string
-	MetricsAddress                 string
-	DefaultBackupLocation          string // TODO(2.0) Deprecate defaultBackupLocation
-	BackupSyncPeriod               time.Duration
-	PodVolumeOperationTimeout      time.Duration
-	ResourceTerminatingTimeout     time.Duration
-	DefaultBackupTTL               time.Duration
-	DefaultVGSLabelKey             string
-	StoreValidationFrequency       time.Duration
-	DefaultCSISnapshotTimeout      time.Duration
-	DefaultItemOperationTimeout    time.Duration
-	ResourceTimeout                time.Duration
-	RestoreResourcePriorities      types.Priorities
-	DefaultVolumeSnapshotLocations flag.Map
-	RestoreOnly                    bool
-	DisabledControllers            []string
-	ClientQPS                      float32
-	ClientBurst                    int
-	ClientPageSize                 int
-	ProfilerAddress                string
-	LogLevel                       *logging.LevelFlag
-	LogFormat                      *logging.FormatFlag
-	RepoMaintenanceFrequency       time.Duration
-	GarbageCollectionFrequency     time.Duration
-	ItemOperationSyncFrequency     time.Duration
-	DefaultVolumesToFsBackup       bool
-	UploaderType                   string
-	MaxConcurrentK8SConnections    int
-	DefaultSnapshotMoveData        bool
-	DisableInformerCache           bool
-	ScheduleSkipImmediately        bool
-	CredentialsDirectory           string
-	BackupRepoConfig               string
-	RepoMaintenanceJobConfig       string
-	ItemBlockWorkerCount           int
-	ConcurrentBackups              int
+	PluginDir                           string
+	MetricsAddress                      string
+	DefaultBackupLocation               string // TODO(2.0) Deprecate defaultBackupLocation
+	BackupSyncPeriod                    time.Duration
+	PodVolumeOperationTimeout           time.Duration
+	ResourceTerminatingTimeout          time.Duration
+	DefaultBackupTTL                    time.Duration
+	DefaultVGSLabelKey                  string
+	StoreValidationFrequency            time.Duration
+	DefaultCSISnapshotTimeout           time.Duration
+	DefaultItemOperationTimeout         time.Duration
+	ResourceTimeout                     time.Duration
+	RestoreResourcePriorities           types.Priorities
+	DefaultVolumeSnapshotLocations      flag.Map
+	RestoreOnly                         bool
+	DisabledControllers                 []string
+	ClientQPS                           float32
+	ClientBurst                         int
+	ClientPageSize                      int
+	ProfilerAddress                     string
+	LogLevel                            *logging.LevelFlag
+	LogFormat                           *logging.FormatFlag
+	RepoMaintenanceFrequency            time.Duration
+	GarbageCollectionFrequency          time.Duration
+	ItemOperationSyncFrequency          time.Duration
+	DefaultVolumesToFsBackup            bool
+	UploaderType                        string
+	MaxConcurrentK8SConnections         int
+	DefaultSnapshotMoveData             bool
+	DisableInformerCache                bool
+	ScheduleSkipImmediately             bool
+	CredentialsDirectory                string
+	BackupRepoConfig                    string
+	RepoMaintenanceJobConfig            string
+	ItemBlockWorkerCount                int
+	ConcurrentBackups                   int
+	GlobalBackupVolumePoliciesConfigMap string
 }
 
 func GetDefaultConfig() *Config {
@@ -275,4 +276,10 @@ func (c *Config) BindFlags(flags *pflag.FlagSet) {
 		c.ConcurrentBackups,
 		"Number of backups to process concurrently. Default is one. Optional.",
 	)
+	flags.StringVar(
+		&c.GlobalBackupVolumePoliciesConfigMap,
+		"global-backup-volume-policies-configmap",
+		c.GlobalBackupVolumePoliciesConfigMap,
+		"The name of a ConfigMap in the Velero install namespace holding global backup volume policies that are merged into every backup. Optional.",
+	)
 }

pkg/cmd/server/config/config_test.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/spf13/pflag"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestGetDefaultConfig(t *testing.T) {
@@ -17,3 +18,14 @@ func TestBindFlags(t *testing.T) {
 	config.BindFlags(pflag.CommandLine)
 	assert.Equal(t, 1, config.ItemBlockWorkerCount)
 }
+
+func TestGlobalBackupVolumePoliciesConfigMapFlag(t *testing.T) {
+	config := GetDefaultConfig()
+	// Opt-in: defaults to empty.
+	assert.Empty(t, config.GlobalBackupVolumePoliciesConfigMap)
+
+	flags := pflag.NewFlagSet("test", pflag.ContinueOnError)
+	config.BindFlags(flags)
+	require.NoError(t, flags.Parse([]string{"--global-backup-volume-policies-configmap", "global-volume-policy"}))
+	assert.Equal(t, "global-volume-policy", config.GlobalBackupVolumePoliciesConfigMap)
+}

pkg/cmd/server/server.go

@@ -57,6 +57,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	"github.com/vmware-tanzu/velero/internal/hook"
+	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/storage"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
@@ -390,6 +391,14 @@ func (s *server) setupBeforeControllerRun() error {
 	if err := setDefaultBackupLocation(s.ctx, client, s.namespace, s.config.DefaultBackupLocation, s.logger); err != nil {
 		return err
 	}
+
+	// Validate the global backup volume policies ConfigMap early, so misconfigurations fail fast.
+	if s.config.GlobalBackupVolumePoliciesConfigMap != "" {
+		if _, err := resourcepolicies.GetGlobalResourcePolicies(client, s.namespace, s.config.GlobalBackupVolumePoliciesConfigMap, s.logger); err != nil {
+			return err
+		}
+		s.logger.WithField("configmap", s.config.GlobalBackupVolumePoliciesConfigMap).Info("Loaded global backup volume policies")
+	}
 	return nil
 }
 
@@ -671,6 +680,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			s.config.ItemBlockWorkerCount,
 			s.config.ConcurrentBackups,
 			s.crClient,
+			s.config.GlobalBackupVolumePoliciesConfigMap,
 		).SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", constant.ControllerBackup)
 		}

pkg/cmd/util/output/backup_describer.go

@@ -99,6 +99,8 @@ func DescribeBackup(
 			DescribeFineGrainedFilterPolicies(ctx, kbClient, d, backup)
 		}
 
+		DescribeGlobalVolumePolicy(d, backup)
+
 		if backup.Spec.UploaderConfig != nil && backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
 			d.Println()
 			DescribeUploaderConfigForBackup(d, backup.Spec)
@@ -136,6 +138,19 @@ func DescribeResourcePolicies(d *Describer, resPolicies *corev1api.TypedLocalObj
 	d.Printf("\tName:\t%s\n", resPolicies.Name)
 }
 
+// DescribeGlobalVolumePolicy describes the cluster-wide global backup volume policies
+// ConfigMap that contributed to the backup, if any.
+func DescribeGlobalVolumePolicy(d *Describer, backup *velerov1api.Backup) {
+	name := backup.Annotations[velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation]
+	if name == "" {
+		return
+	}
+	d.Println()
+	d.Printf("Global volume policies:\n")
+	d.Printf("\tType:\t%s\n", resourcepolicies.ConfigmapRefType)
+	d.Printf("\tName:\t%s\n", name)
+}
+
 // DescribeFineGrainedFilterPolicies describes cluster-scoped and namespace-scoped filter policies if present
 func DescribeFineGrainedFilterPolicies(ctx context.Context, kbClient kbclient.Client, d *Describer, backup *velerov1api.Backup) {
 	if backup.Spec.ResourcePolicy == nil {

pkg/cmd/util/output/backup_describer_test.go

@@ -72,6 +72,34 @@ func TestDescribeResourcePolicies(t *testing.T) {
 	assert.Equal(t, expect, d.buf.String())
 }
 
+func TestDescribeGlobalVolumePolicy(t *testing.T) {
+	newDescriber := func() *Describer {
+		d := &Describer{out: &tabwriter.Writer{}, buf: &bytes.Buffer{}}
+		d.out.Init(d.buf, 0, 8, 2, ' ', 0)
+		return d
+	}
+
+	// No annotation: nothing is printed.
+	d := newDescriber()
+	DescribeGlobalVolumePolicy(d, builder.ForBackup("velero", "b").Result())
+	d.out.Flush()
+	assert.Empty(t, d.buf.String())
+
+	// Annotation present: ConfigMap name is surfaced.
+	d = newDescriber()
+	backup := builder.ForBackup("velero", "b").
+		ObjectMeta(builder.WithAnnotations(velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation, "global-volume-policy")).
+		Result()
+	DescribeGlobalVolumePolicy(d, backup)
+	d.out.Flush()
+	expect := `
+Global volume policies:
+  Type:  configmap
+  Name:  global-volume-policy
+`
+	assert.Equal(t, expect, d.buf.String())
+}
+
 func TestDescribeBackupSpec(t *testing.T) {
 	input1 := builder.ForBackup("test-ns", "test-backup-1").
 		IncludedNamespaces("inc-ns-1", "inc-ns-2").

pkg/cmd/util/output/backup_structured_describer.go

@@ -60,6 +60,8 @@ func DescribeBackupInSF(
 			DescribeFineGrainedFilterPoliciesInSF(ctx, kbClient, d, backup)
 		}
 
+		DescribeGlobalVolumePolicyInSF(d, backup)
+
 		status := backup.Status
 		if len(status.ValidationErrors) > 0 {
 			d.Describe("validationErrors", status.ValidationErrors)
@@ -699,6 +701,19 @@ func DescribeResourcePoliciesInSF(d *StructuredDescriber, resPolicies *corev1api
 	d.Describe("resourcePolicies", policiesInfo)
 }
 
+// DescribeGlobalVolumePolicyInSF describes the global backup volume policies ConfigMap that
+// contributed to the backup, if any, in structured format.
+func DescribeGlobalVolumePolicyInSF(d *StructuredDescriber, backup *velerov1api.Backup) {
+	name := backup.Annotations[velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation]
+	if name == "" {
+		return
+	}
+	d.Describe("globalVolumePolicies", map[string]any{
+		"type": resourcepolicies.ConfigmapRefType,
+		"name": name,
+	})
+}
+
 func describeResultInSF(m map[string]any, result results.Result) {
 	m["velero"], m["cluster"], m["namespace"] = []string{}, []string{}, []string{}
 

pkg/cmd/util/output/backup_structured_describer_test.go

@@ -627,6 +627,27 @@ func TestDescribeResourcePoliciesInSF(t *testing.T) {
 	assert.True(t, reflect.DeepEqual(sd.output, expect))
 }
 
+func TestDescribeGlobalVolumePolicyInSF(t *testing.T) {
+	// No annotation: nothing is added to the output.
+	sd := &StructuredDescriber{output: make(map[string]any), format: ""}
+	DescribeGlobalVolumePolicyInSF(sd, builder.ForBackup("velero", "b").Result())
+	assert.Empty(t, sd.output)
+
+	// Annotation present: the ConfigMap name is surfaced.
+	sd = &StructuredDescriber{output: make(map[string]any), format: ""}
+	backup := builder.ForBackup("velero", "b").
+		ObjectMeta(builder.WithAnnotations(velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation, "global-volume-policy")).
+		Result()
+	DescribeGlobalVolumePolicyInSF(sd, backup)
+	expectGlobal := map[string]any{
+		"globalVolumePolicies": map[string]any{
+			"type": "configmap",
+			"name": "global-volume-policy",
+		},
+	}
+	assert.True(t, reflect.DeepEqual(sd.output, expectGlobal))
+}
+
 func TestDescribeBackupResultInSF(t *testing.T) {
 	input := results.Result{
 		Velero:  []string{"msg-1", "msg-2"},

pkg/controller/backup_controller.go

@@ -84,32 +84,33 @@ var autoExcludeClusterScopedResources = []string{
 }
 
 type backupReconciler struct {
-	ctx                         context.Context
-	logger                      logrus.FieldLogger
-	discoveryHelper             discovery.Helper
-	backupper                   pkgbackup.Backupper
-	kbClient                    kbclient.Client
-	clock                       clock.WithTickerAndDelayedExecution
-	backupLogLevel              logrus.Level
-	newPluginManager            func(logrus.FieldLogger) clientmgmt.Manager
-	backupTracker               BackupTracker
-	defaultBackupLocation       string
-	defaultVolumesToFsBackup    bool
-	defaultBackupTTL            time.Duration
-	defaultVGSLabelKey          string
-	defaultCSISnapshotTimeout   time.Duration
-	resourceTimeout             time.Duration
-	defaultItemOperationTimeout time.Duration
-	defaultSnapshotLocations    map[string]string
-	metrics                     *metrics.ServerMetrics
-	backupStoreGetter           persistence.ObjectBackupStoreGetter
-	formatFlag                  logging.Format
-	credentialFileStore         credentials.FileStore
-	maxConcurrentK8SConnections int
-	defaultSnapshotMoveData     bool
-	globalCRClient              kbclient.Client
-	itemBlockWorkerCount        int
-	concurrentBackups           int
+	ctx                           context.Context
+	logger                        logrus.FieldLogger
+	discoveryHelper               discovery.Helper
+	backupper                     pkgbackup.Backupper
+	kbClient                      kbclient.Client
+	clock                         clock.WithTickerAndDelayedExecution
+	backupLogLevel                logrus.Level
+	newPluginManager              func(logrus.FieldLogger) clientmgmt.Manager
+	backupTracker                 BackupTracker
+	defaultBackupLocation         string
+	defaultVolumesToFsBackup      bool
+	defaultBackupTTL              time.Duration
+	defaultVGSLabelKey            string
+	defaultCSISnapshotTimeout     time.Duration
+	resourceTimeout               time.Duration
+	defaultItemOperationTimeout   time.Duration
+	defaultSnapshotLocations      map[string]string
+	metrics                       *metrics.ServerMetrics
+	backupStoreGetter             persistence.ObjectBackupStoreGetter
+	formatFlag                    logging.Format
+	credentialFileStore           credentials.FileStore
+	maxConcurrentK8SConnections   int
+	defaultSnapshotMoveData       bool
+	globalCRClient                kbclient.Client
+	itemBlockWorkerCount          int
+	concurrentBackups             int
+	globalVolumePoliciesConfigMap string
 }
 
 func NewBackupReconciler(
@@ -138,34 +139,36 @@ func NewBackupReconciler(
 	itemBlockWorkerCount int,
 	concurrentBackups int,
 	globalCRClient kbclient.Client,
+	globalVolumePoliciesConfigMap string,
 ) *backupReconciler {
 	b := &backupReconciler{
-		ctx:                         ctx,
-		discoveryHelper:             discoveryHelper,
-		backupper:                   backupper,
-		clock:                       &clock.RealClock{},
-		logger:                      logger,
-		backupLogLevel:              backupLogLevel,
-		newPluginManager:            newPluginManager,
-		backupTracker:               backupTracker,
-		kbClient:                    kbClient,
-		defaultBackupLocation:       defaultBackupLocation,
-		defaultVolumesToFsBackup:    defaultVolumesToFsBackup,
-		defaultBackupTTL:            defaultBackupTTL,
-		defaultVGSLabelKey:          defaultVGSLabelKey,
-		defaultCSISnapshotTimeout:   defaultCSISnapshotTimeout,
-		resourceTimeout:             resourceTimeout,
-		defaultItemOperationTimeout: defaultItemOperationTimeout,
-		defaultSnapshotLocations:    defaultSnapshotLocations,
-		metrics:                     metrics,
-		backupStoreGetter:           backupStoreGetter,
-		formatFlag:                  formatFlag,
-		credentialFileStore:         credentialStore,
-		maxConcurrentK8SConnections: maxConcurrentK8SConnections,
-		defaultSnapshotMoveData:     defaultSnapshotMoveData,
-		itemBlockWorkerCount:        itemBlockWorkerCount,
-		concurrentBackups:           max(concurrentBackups, 1),
-		globalCRClient:              globalCRClient,
+		ctx:                           ctx,
+		discoveryHelper:               discoveryHelper,
+		backupper:                     backupper,
+		clock:                         &clock.RealClock{},
+		logger:                        logger,
+		backupLogLevel:                backupLogLevel,
+		newPluginManager:              newPluginManager,
+		backupTracker:                 backupTracker,
+		kbClient:                      kbClient,
+		defaultBackupLocation:         defaultBackupLocation,
+		defaultVolumesToFsBackup:      defaultVolumesToFsBackup,
+		defaultBackupTTL:              defaultBackupTTL,
+		defaultVGSLabelKey:            defaultVGSLabelKey,
+		defaultCSISnapshotTimeout:     defaultCSISnapshotTimeout,
+		resourceTimeout:               resourceTimeout,
+		defaultItemOperationTimeout:   defaultItemOperationTimeout,
+		defaultSnapshotLocations:      defaultSnapshotLocations,
+		metrics:                       metrics,
+		backupStoreGetter:             backupStoreGetter,
+		formatFlag:                    formatFlag,
+		credentialFileStore:           credentialStore,
+		maxConcurrentK8SConnections:   maxConcurrentK8SConnections,
+		defaultSnapshotMoveData:       defaultSnapshotMoveData,
+		itemBlockWorkerCount:          itemBlockWorkerCount,
+		concurrentBackups:             max(concurrentBackups, 1),
+		globalCRClient:                globalCRClient,
+		globalVolumePoliciesConfigMap: globalVolumePoliciesConfigMap,
 	}
 	b.updateTotalBackupMetric()
 	return b
@@ -587,9 +590,13 @@ func (b *backupReconciler) prepareBackupRequest(ctx context.Context, backup *vel
 		request.Status.ValidationErrors = append(request.Status.ValidationErrors, "encountered labelSelector as well as orLabelSelectors in backup spec, only one can be specified")
 	}
 
-	resourcePolicies, err := resourcepolicies.GetResourcePoliciesFromBackup(*request.Backup, b.kbClient, logger)
+	resourcePolicies, err := resourcepolicies.GetResourcePoliciesFromBackupWithGlobal(
+		*request.Backup, b.kbClient, b.globalVolumePoliciesConfigMap, request.Namespace, logger)
 	if err != nil {
 		request.Status.ValidationErrors = append(request.Status.ValidationErrors, err.Error())
+	} else if b.globalVolumePoliciesConfigMap != "" {
+		// Record the contributing global volume policies ConfigMap so `velero backup describe` can surface it.
+		request.Annotations[velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation] = b.globalVolumePoliciesConfigMap
 	}
 	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"+

pkg/controller/backup_controller_test.go

@@ -46,6 +46,7 @@ import (
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
 	fakeClient "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"
 	pkgbackup "github.com/vmware-tanzu/velero/pkg/backup"
 	"github.com/vmware-tanzu/velero/pkg/builder"
@@ -2076,6 +2077,100 @@ namespacedFilterPolicies:
 	assert.True(t, hasTargetError, "expected validation error about namespacedFilterPolicies incompatibility with old-style filters, got: %v", res.Status.ValidationErrors)
 }
 
+// TestPrepareBackupRequest_GlobalVolumePolicies verifies that the cluster-wide global backup
+// volume policies are merged into the request and that the contributing ConfigMap is recorded
+// on the backup so `velero backup describe` can surface it.
+func TestPrepareBackupRequest_GlobalVolumePolicies(t *testing.T) {
+	formatFlag := logging.FormatText
+	logger := logging.DefaultLogger(logrus.DebugLevel, formatFlag)
+
+	globalCM := &corev1api.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{Name: "global-volume-policy", Namespace: velerov1api.DefaultNamespace},
+		Data: map[string]string{"policies.yaml": `version: v1
+volumePolicies:
+  - conditions:
+      storageClass:
+        - gp2
+    action:
+      type: skip
+`},
+	}
+
+	fakeClient := velerotest.NewFakeControllerRuntimeClient(t, globalCM,
+		builder.ForBackupStorageLocation(velerov1api.DefaultNamespace, "loc-1").Result())
+	apiServer := velerotest.NewAPIServer(t)
+	discoveryHelper, err := discovery.NewHelper(apiServer.DiscoveryClient, logger)
+	require.NoError(t, err)
+
+	c := &backupReconciler{
+		logger:                        logger,
+		discoveryHelper:               discoveryHelper,
+		kbClient:                      fakeClient,
+		clock:                         &clock.RealClock{},
+		formatFlag:                    formatFlag,
+		defaultBackupLocation:         "loc-1",
+		globalVolumePoliciesConfigMap: "global-volume-policy",
+	}
+
+	backup := defaultBackup().StorageLocation("loc-1").Result()
+	res := c.prepareBackupRequest(ctx, backup, logger)
+	defer res.WorkerPool.Stop()
+
+	// The global volume policies must load cleanly (no policy-related validation error).
+	for _, e := range res.Status.ValidationErrors {
+		assert.NotContains(t, e, "global backup volume policies")
+	}
+	require.NotNil(t, res.ResPolicies)
+	assert.Equal(t, "global-volume-policy", res.Annotations[velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation])
+
+	action, err := res.ResPolicies.GetMatchAction(resourcepolicies.VolumeFilterData{
+		PersistentVolume: &corev1api.PersistentVolume{Spec: corev1api.PersistentVolumeSpec{StorageClassName: "gp2"}},
+	})
+	require.NoError(t, err)
+	require.NotNil(t, action)
+	assert.Equal(t, resourcepolicies.Skip, action.Type)
+}
+
+// TestPrepareBackupRequest_GlobalVolumePolicies_LoadError verifies that when the configured
+// global backup volume policies ConfigMap cannot be loaded, a validation error is recorded and
+// the contributing-ConfigMap annotation is not set on the backup.
+func TestPrepareBackupRequest_GlobalVolumePolicies_LoadError(t *testing.T) {
+	formatFlag := logging.FormatText
+	logger := logging.DefaultLogger(logrus.DebugLevel, formatFlag)
+
+	// No ConfigMap with this name exists, so loading the global policies fails.
+	fakeClient := velerotest.NewFakeControllerRuntimeClient(t,
+		builder.ForBackupStorageLocation(velerov1api.DefaultNamespace, "loc-1").Result())
+	apiServer := velerotest.NewAPIServer(t)
+	discoveryHelper, err := discovery.NewHelper(apiServer.DiscoveryClient, logger)
+	require.NoError(t, err)
+
+	c := &backupReconciler{
+		logger:                        logger,
+		discoveryHelper:               discoveryHelper,
+		kbClient:                      fakeClient,
+		clock:                         &clock.RealClock{},
+		formatFlag:                    formatFlag,
+		defaultBackupLocation:         "loc-1",
+		globalVolumePoliciesConfigMap: "missing-global-volume-policy",
+	}
+
+	backup := defaultBackup().StorageLocation("loc-1").Result()
+	res := c.prepareBackupRequest(ctx, backup, logger)
+	defer res.WorkerPool.Stop()
+
+	// The failure to load the global policies must surface as a validation error.
+	var hasGlobalPolicyError bool
+	for _, e := range res.Status.ValidationErrors {
+		if strings.Contains(e, "global backup volume policies") {
+			hasGlobalPolicyError = true
+		}
+	}
+	assert.True(t, hasGlobalPolicyError, "expected a validation error about global backup volume policies, got: %v", res.Status.ValidationErrors)
+	// The annotation is only set when the policies load successfully.
+	assert.Empty(t, res.Annotations[velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation])
+}
+
 // TestPrepareBackupRequest_ClusterScopedFilterPolicyIncompatibleWithOldFilters verifies
 // that a backup referencing a ResourcePolicy ConfigMap with clusterScopedFilterPolicy
 // produces a validation error when old-style resource filters are also set on the spec.

site/content/docs/main/resource-filtering.md

@@ -704,3 +704,85 @@ volumePolicies:
 3. The outcome would be that velero would perform `fs-backup` operation on both the volumes
    - `fs-backup` on `Volume 1` because `Volume 1` satisfies the criteria for `fs-backup` action. 
    - Also, for Volume 2 as no matching action was found so legacy approach will be used as a fallback option for this volume (`fs-backup` operation will be done as `defaultVolumesToFSBackup: true` is specified by the user).
+
+### Global backup volume policies
+
+Resource policies (volume policies) are normally opt-in per backup via `--resource-policies-configmap`. An administrator can instead configure a cluster-wide baseline that applies to **every** backup by starting the Velero server with the `--global-backup-volume-policies-configmap` flag, pointing at a ConfigMap in the Velero install namespace:
+
+```bash
+velero server --global-backup-volume-policies-configmap global-volume-policy
+```
+
+The ConfigMap uses the exact same format as a per-backup resource policies ConfigMap (a single data key holding a `ResourcePolicies` YAML document):
+
+```yaml
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: global-volume-policy
+  namespace: velero
+data:
+  policies.yaml: |
+    version: v1
+    volumePolicies:
+      - conditions:
+          storageClass:
+            - gp2
+        action:
+          type: skip
+```
+
+#### Behavior
+
+- **Only `volumePolicies` apply globally.** If the global ConfigMap contains `includeExcludePolicy`, `clusterScopedFilterPolicy`, or `namespacedFilterPolicies`, those sections are ignored and a warning is logged. Those filters are tied to a specific backup use case, so they remain per-backup only.
+- **Merge semantics.** When a backup runs, the effective `volumePolicies` list is the backup-level policies followed by the global policies:
+
+  ```
+  merged.volumePolicies = backup.volumePolicies ++ global.volumePolicies
+  ```
+
+  Because the first matching policy wins, a backup can override the global baseline for a specific volume while still inheriting every global rule it does not override. If a backup references no resource policy, the global policy applies on its own.
+- **Validation.** The global ConfigMap is validated at server startup (the server fails to start if it is missing or invalid) and again on each backup (a backup whose global policy has become missing or invalid is moved to the `FailedValidation` phase).
+
+#### Example
+
+Global policy (`--global-backup-volume-policies-configmap=global-volume-policy`): skip `gp2` volumes.
+
+```yaml
+version: v1
+volumePolicies:
+  - conditions:
+      storageClass:
+        - gp2
+    action:
+      type: skip
+```
+
+Backup-level policy (`--resource-policies-configmap backup01`): `fs-backup` NFS volumes.
+
+```yaml
+version: v1
+volumePolicies:
+  - conditions:
+      nfs: {}
+    action:
+      type: fs-backup
+```
+
+Effective (merged) policy used for the backup — backup rules first, then global:
+
+```yaml
+version: v1
+volumePolicies:
+  - conditions:
+      nfs: {}
+    action:
+      type: fs-backup
+  - conditions:
+      storageClass:
+        - gp2
+    action:
+      type: skip
+```
+
+When a global policy contributes to a backup, `velero backup describe` surfaces the contributing ConfigMap under a `Global volume policies` section.