Merge pull request #9928 from chlins/feat/global-policy

feat(backup): add global backup volume policies

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

internal/resourcepolicies/resource_policies.go

@@ -351,6 +351,80 @@ func GetResourcePoliciesFromBackup(
 	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
+}
+
 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

@@ -18,11 +18,15 @@ package resourcepolicies
 import (
 	"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"
+
+	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 {
@@ -2172,3 +2176,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())
+}

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/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.