Merge pull request #9678 from sseago/custom-volume-policy-1.18

[release-1.18] Add custom action type to volume policies

changelogs/unreleased/9678-sseago

@@ -0,0 +1 @@
+Add custom action type to volume policies

internal/resourcepolicies/resource_policies.go

@@ -42,6 +42,8 @@ const (
 	FSBackup VolumeActionType = "fs-backup"
 	// snapshot action can have 3 different meaning based on velero configuration and backup spec - cloud provider based snapshots, local csi snapshots and datamover snapshots
 	Snapshot VolumeActionType = "snapshot"
+	// custom action is used to identify a volume that will be handled by an external plugin. Velero will not snapshot or use fs-backup if action=="custom"
+	Custom VolumeActionType = "custom"
 )
 
 // Action defined as one action for a specific way of backup

internal/resourcepolicies/volume_resources_validator.go

@@ -90,7 +90,7 @@ func decodeStruct(r io.Reader, s any) error {
 func (a *Action) validate() error {
 	// validate Type
 	valid := false
-	if a.Type == Skip || a.Type == Snapshot || a.Type == FSBackup {
+	if a.Type == Skip || a.Type == Snapshot || a.Type == FSBackup || a.Type == Custom {
 		valid = true
 	}
 	if !valid {

internal/volumehelper/volume_policy_helper.go

@@ -18,13 +18,9 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	kubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
 	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
+	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
 )
 
-type VolumeHelper interface {
-	ShouldPerformSnapshot(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, error)
-	ShouldPerformFSBackup(volume corev1api.Volume, pod corev1api.Pod) (bool, error)
-}
-
 type volumeHelperImpl struct {
 	volumePolicy             *resourcepolicies.Policies
 	snapshotVolumes          *bool
@@ -53,7 +49,7 @@ func NewVolumeHelperImpl(
 	client crclient.Client,
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
-) VolumeHelper {
+) vhutil.VolumeHelper {
 	// Pass nil namespaces - no cache will be built, so this never fails.
 	// This is used by plugins that don't need the cache optimization.
 	vh, _ := NewVolumeHelperImplWithNamespaces(
@@ -81,7 +77,7 @@ func NewVolumeHelperImplWithNamespaces(
 	defaultVolumesToFSBackup bool,
 	backupExcludePVC bool,
 	namespaces []string,
-) (VolumeHelper, error) {
+) (vhutil.VolumeHelper, error) {
 	var pvcPodCache *podvolumeutil.PVCPodCache
 	if len(namespaces) > 0 {
 		pvcPodCache = podvolumeutil.NewPVCPodCache()
@@ -110,7 +106,7 @@ func NewVolumeHelperImplWithCache(
 	client crclient.Client,
 	logger logrus.FieldLogger,
 	pvcPodCache *podvolumeutil.PVCPodCache,
-) (VolumeHelper, error) {
+) (vhutil.VolumeHelper, error) {
 	resourcePolicies, err := resourcepolicies.GetResourcePoliciesFromBackup(backup, client, logger)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get volume policies from backup")
@@ -319,6 +315,121 @@ func (v volumeHelperImpl) shouldPerformFSBackupLegacy(
 	}
 }
 
+func (v *volumeHelperImpl) ShouldPerformCustomAction(obj runtime.Unstructured, groupResource schema.GroupResource, matchParams map[string]any) (bool, error) {
+	// check if volume policy exists and also check if the object(pv/pvc) fits a volume policy criteria and see if the associated action is custom with the provided param values
+	pvc := new(corev1api.PersistentVolumeClaim)
+	pv := new(corev1api.PersistentVolume)
+	var err error
+
+	var pvNotFoundErr error
+	if groupResource == kuberesource.PersistentVolumeClaims {
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &pvc); err != nil {
+			v.logger.WithError(err).Error("fail to convert unstructured into PVC")
+			return false, err
+		}
+
+		pv, err = kubeutil.GetPVForPVC(pvc, v.client)
+		if err != nil {
+			// Any error means PV not available - save to return later if no policy matches
+			v.logger.Debugf("PV not found for PVC %s: %v", pvc.Namespace+"/"+pvc.Name, err)
+			pvNotFoundErr = err
+			pv = nil
+		}
+	}
+
+	if groupResource == kuberesource.PersistentVolumes {
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &pv); err != nil {
+			v.logger.WithError(err).Error("fail to convert unstructured into PV")
+			return false, err
+		}
+	}
+
+	if v.volumePolicy != nil {
+		vfd := resourcepolicies.NewVolumeFilterData(pv, nil, pvc)
+		action, err := v.volumePolicy.GetMatchAction(vfd)
+		if err != nil {
+			v.logger.WithError(err).Errorf("fail to get VolumePolicy match action for %+v", vfd)
+			return false, err
+		}
+
+		// If there is a match action, and the action type is custom, return true
+		// if the provided parameters match as well, else return false.
+		// If there is no match action, also return false
+		if action != nil {
+			if action.Type == resourcepolicies.Custom {
+				for k, requiredValue := range matchParams {
+					if actionValue, ok := action.Parameters[k]; !ok || actionValue != requiredValue {
+						v.logger.Infof("Skipping custom action for %+v as value for parameter %s is %s rather than the required %s", vfd, k, actionValue, requiredValue)
+						return false, nil
+					}
+				}
+				v.logger.Infof("performing custom action for %+v", vfd)
+				return true, nil
+			} else {
+				v.logger.Infof("Skipping custom action for %+v as the action type is %s", vfd, action.Type)
+				return false, nil
+			}
+		}
+	}
+	// If resource is PVC, and PV is nil (e.g., Pending/Lost PVC with no matching policy), return the original error
+	// Don't error out on no PV, just return false
+	if groupResource == kuberesource.PersistentVolumeClaims && pv == nil && pvNotFoundErr != nil {
+		v.logger.WithError(pvNotFoundErr).Warnf("fail to get PV for PVC %s", pvc.Namespace+"/"+pvc.Name)
+		return false, nil
+	}
+
+	v.logger.Infof("skipping custom action for pv %s due to no matching volume policy", pv.Name)
+	return false, nil
+}
+
+// returns false if no matching action found. Returns true with the action name and Parameters map if there is a matching policy
+func (v *volumeHelperImpl) GetActionParameters(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, string, map[string]any, error) {
+	// if volume policy exists, return action parameters.
+	pvc := new(corev1api.PersistentVolumeClaim)
+	pv := new(corev1api.PersistentVolume)
+	var err error
+
+	if groupResource == kuberesource.PersistentVolumeClaims {
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &pvc); err != nil {
+			v.logger.WithError(err).Error("fail to convert unstructured into PVC")
+			return false, "", nil, err
+		}
+
+		pv, err = kubeutil.GetPVForPVC(pvc, v.client)
+		if err != nil {
+			v.logger.WithError(err).Warnf("failed to get PV for PVC %s", pvc.Namespace+"/"+pvc.Name)
+			return false, "", nil, nil
+		}
+	}
+
+	if groupResource == kuberesource.PersistentVolumes {
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &pv); err != nil {
+			v.logger.WithError(err).Error("fail to convert unstructured into PV")
+			return false, "", nil, err
+		}
+	}
+
+	if v.volumePolicy != nil {
+		vfd := resourcepolicies.NewVolumeFilterData(pv, nil, pvc)
+		action, err := v.volumePolicy.GetMatchAction(vfd)
+		if err != nil {
+			v.logger.WithError(err).Errorf("fail to get VolumePolicy match action for PV %s", pv.Name)
+			return false, "", nil, err
+		}
+
+		// If there is a match action, and the action type is custom, return true
+		// if the provided parameters match as well, else return false.
+		// If there is no match action, also return false
+		if action != nil {
+			v.logger.Infof("found matching action for pv %s, returning parameters", pv.Name)
+			return true, string(action.Type), action.Parameters, nil
+		}
+	}
+
+	v.logger.Infof("no matching volume policy found for pv %s, no parameters to return", pv.Name)
+	return false, "", nil, nil
+}
+
 func (v *volumeHelperImpl) shouldIncludeVolumeInBackup(vol corev1api.Volume) bool {
 	includeVolumeInBackup := true
 	// cannot backup hostpath volumes as they are not mounted into /var/lib/kubelet/pods

pkg/backup/actions/csi/pvc_action.go

@@ -44,7 +44,6 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/resource"
 
-	internalvolumehelper "github.com/vmware-tanzu/velero/internal/volumehelper"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	veleroclient "github.com/vmware-tanzu/velero/pkg/client"
@@ -59,6 +58,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/util/csi"
 	kubeutil "github.com/vmware-tanzu/velero/pkg/util/kube"
 	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
+	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
 )
 
 // TODO: Replace hardcoded VolumeSnapshot finalizer strings with constants from
@@ -128,9 +128,9 @@ func (p *pvcBackupItemAction) ensurePVCPodCacheForNamespace(ctx context.Context,
 
 // getVolumeHelperWithCache creates a VolumeHelper using the pre-built PVC-to-Pod cache.
 // The cache should be ensured for the relevant namespace(s) before calling this.
-func (p *pvcBackupItemAction) getVolumeHelperWithCache(backup *velerov1api.Backup) (internalvolumehelper.VolumeHelper, error) {
+func (p *pvcBackupItemAction) getVolumeHelperWithCache(backup *velerov1api.Backup) (vhutil.VolumeHelper, error) {
 	// Create VolumeHelper with our lazy-built cache
-	vh, err := internalvolumehelper.NewVolumeHelperImplWithCache(
+	vh, err := volumehelper.NewVolumeHelperWithCache(
 		*backup,
 		p.crClient,
 		p.log,
@@ -149,7 +149,7 @@ func (p *pvcBackupItemAction) getVolumeHelperWithCache(backup *velerov1api.Backu
 // Since plugin instances are unique per backup (created via newPluginManager and
 // cleaned up via CleanupClients at backup completion), we can safely cache this.
 // See issue #9179 and PR #9226 for details.
-func (p *pvcBackupItemAction) getOrCreateVolumeHelper(backup *velerov1api.Backup) (internalvolumehelper.VolumeHelper, error) {
+func (p *pvcBackupItemAction) getOrCreateVolumeHelper(backup *velerov1api.Backup) (vhutil.VolumeHelper, error) {
 	// Initialize the PVC-to-Pod cache if needed
 	if p.pvcPodCache == nil {
 		p.pvcPodCache = podvolumeutil.NewPVCPodCache()
@@ -322,13 +322,9 @@ func (p *pvcBackupItemAction) Execute(
 		return nil, nil, "", nil, err
 	}
 
-	shouldSnapshot, err := volumehelper.ShouldPerformSnapshotWithVolumeHelper(
+	shouldSnapshot, err := vh.ShouldPerformSnapshot(
 		item,
 		kuberesource.PersistentVolumeClaims,
-		*backup,
-		p.crClient,
-		p.log,
-		vh,
 	)
 	if err != nil {
 		return nil, nil, "", nil, err
@@ -708,7 +704,7 @@ func (p *pvcBackupItemAction) getVolumeSnapshotReference(
 		}
 
 		// Filter PVCs by volume policy
-		filteredPVCs, err := p.filterPVCsByVolumePolicy(groupedPVCs, backup, vh)
+		filteredPVCs, err := p.filterPVCsByVolumePolicy(groupedPVCs, vh)
 		if err != nil {
 			return nil, errors.Wrapf(err, "failed to filter PVCs by volume policy for VolumeGroupSnapshot group %q", group)
 		}
@@ -844,8 +840,7 @@ func (p *pvcBackupItemAction) listGroupedPVCs(ctx context.Context, namespace, la
 
 func (p *pvcBackupItemAction) filterPVCsByVolumePolicy(
 	pvcs []corev1api.PersistentVolumeClaim,
-	backup *velerov1api.Backup,
-	vh internalvolumehelper.VolumeHelper,
+	vh vhutil.VolumeHelper,
 ) ([]corev1api.PersistentVolumeClaim, error) {
 	var filteredPVCs []corev1api.PersistentVolumeClaim
 
@@ -859,13 +854,9 @@ func (p *pvcBackupItemAction) filterPVCsByVolumePolicy(
 
 		// Check if this PVC should be snapshotted according to volume policies
 		// Uses the cached VolumeHelper for better performance with many PVCs/pods
-		shouldSnapshot, err := volumehelper.ShouldPerformSnapshotWithVolumeHelper(
+		shouldSnapshot, err := vh.ShouldPerformSnapshot(
 			unstructuredPVC,
 			kuberesource.PersistentVolumeClaims,
-			*backup,
-			p.crClient,
-			p.log,
-			vh,
 		)
 		if err != nil {
 			return nil, errors.Wrapf(err, "failed to check volume policy for PVC %s/%s", pvc.Namespace, pvc.Name)

pkg/backup/actions/csi/pvc_action_test.go

@@ -842,9 +842,13 @@ volumePolicies:
 				crClient: client,
 			}
 
-			// Pass nil for VolumeHelper in tests - it will fall back to creating a new one per call
-			// This is the expected behavior for testing and third-party plugins
-			result, err := action.filterPVCsByVolumePolicy(tt.pvcs, backup, nil)
+			// Create a VolumeHelper using the same method the plugin would use
+			vh, err := action.getOrCreateVolumeHelper(backup)
+			require.NoError(t, err)
+			require.NotNil(t, vh)
+
+			// Test with the pre-created VolumeHelper
+			result, err := action.filterPVCsByVolumePolicy(tt.pvcs, vh)
 			if tt.expectError {
 				require.Error(t, err)
 			} else {
@@ -959,7 +963,7 @@ volumePolicies:
 	require.NotNil(t, vh)
 
 	// Test with the pre-created VolumeHelper (non-nil path)
-	result, err := action.filterPVCsByVolumePolicy(pvcs, backup, vh)
+	result, err := action.filterPVCsByVolumePolicy(pvcs, vh)
 	require.NoError(t, err)
 
 	// Should filter out the NFS PVC, leaving only the CSI PVC

pkg/backup/item_backupper.go

@@ -40,7 +40,6 @@ import (
 	"github.com/vmware-tanzu/velero/internal/hook"
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
 	"github.com/vmware-tanzu/velero/internal/volume"
-	"github.com/vmware-tanzu/velero/internal/volumehelper"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/archive"
 	"github.com/vmware-tanzu/velero/pkg/client"
@@ -54,6 +53,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/podvolume"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
 	csiutil "github.com/vmware-tanzu/velero/pkg/util/csi"
+	"github.com/vmware-tanzu/velero/pkg/util/volumehelper"
 )
 
 const (

pkg/plugin/utils/volumehelper/volume_policy_helper.go

@@ -26,6 +26,8 @@ import (
 	"github.com/vmware-tanzu/velero/internal/volumehelper"
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/util/boolptr"
+	podvolumeutil "github.com/vmware-tanzu/velero/pkg/util/podvolume"
+	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
 )
 
 // ShouldPerformSnapshotWithBackup is used for third-party plugins.
@@ -66,7 +68,7 @@ func ShouldPerformSnapshotWithVolumeHelper(
 	backup velerov1api.Backup,
 	crClient crclient.Client,
 	logger logrus.FieldLogger,
-	vh volumehelper.VolumeHelper,
+	vh vhutil.VolumeHelper,
 ) (bool, error) {
 	// If a VolumeHelper is provided, use it directly
 	if vh != nil {
@@ -95,3 +97,45 @@ func ShouldPerformSnapshotWithVolumeHelper(
 
 	return volumeHelperImpl.ShouldPerformSnapshot(unstructured, groupResource)
 }
+
+// NewVolumeHelperWithNamespaces creates a VolumeHelper with a PVC-to-Pod cache for improved performance.
+// The cache is built internally from the provided namespaces list.
+// This avoids O(N*M) complexity when there are many PVCs and pods.
+// See issue #9179 for details.
+// Returns an error if cache building fails - callers should not proceed with backup in this case.
+func NewVolumeHelperWithNamespaces(
+	volumePolicy *resourcepolicies.Policies,
+	snapshotVolumes *bool,
+	logger logrus.FieldLogger,
+	client crclient.Client,
+	defaultVolumesToFSBackup bool,
+	backupExcludePVC bool,
+	namespaces []string,
+) (vhutil.VolumeHelper, error) {
+	return volumehelper.NewVolumeHelperImplWithNamespaces(
+		volumePolicy,
+		snapshotVolumes,
+		logger,
+		client,
+		defaultVolumesToFSBackup,
+		backupExcludePVC,
+		namespaces,
+	)
+}
+
+// NewVolumeHelperWithCache creates a VolumeHelper using an externally managed PVC-to-Pod cache.
+// This is used by plugins that build the cache lazily per-namespace (following the pattern from PR #9226).
+// The cache can be nil, in which case PVC-to-Pod lookups will fall back to direct API calls.
+func NewVolumeHelperWithCache(
+	backup velerov1api.Backup,
+	client crclient.Client,
+	logger logrus.FieldLogger,
+	pvcPodCache *podvolumeutil.PVCPodCache,
+) (vhutil.VolumeHelper, error) {
+	return volumehelper.NewVolumeHelperImplWithCache(
+		backup,
+		client,
+		logger,
+		pvcPodCache,
+	)
+}

pkg/util/volumehelper/volume_policy_helper.go

@@ -0,0 +1,30 @@
+/*
+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 volumehelper
+
+import (
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+type VolumeHelper interface {
+	ShouldPerformSnapshot(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, error)
+	ShouldPerformFSBackup(volume corev1api.Volume, pod corev1api.Pod) (bool, error)
+	ShouldPerformCustomAction(obj runtime.Unstructured, groupResource schema.GroupResource, matchParams map[string]any) (bool, error)
+	GetActionParameters(obj runtime.Unstructured, groupResource schema.GroupResource) (bool, string, map[string]any, error)
+}