Support to set data mover for the uploader from volume policy.

Refactor function ShouldPerformCustomAction and GetActionParameters:
extract shared code to a new function getPVAndMatchAction.

Signed-off-by: Xun Jiang <xun.jiang@broadcom.com>

changelogs/unreleased/10176-blackpiglet

@@ -0,0 +1 @@
+Support to set data mover for the uploader from volume policy.

internal/volumehelper/volume_policy_helper.go

@@ -8,6 +8,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/sirupsen/logrus"
 	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -21,6 +22,8 @@ import (
 	vhutil "github.com/vmware-tanzu/velero/pkg/util/volumehelper"
 )
 
+var errGetPVForPVC = errors.New("fail to get PV for PVC")
+
 type volumeHelperImpl struct {
 	volumePolicy             *resourcepolicies.Policies
 	snapshotVolumes          *bool
@@ -123,6 +126,44 @@ func NewVolumeHelperImplWithCache(
 	}, nil
 }
 
+func (v *volumeHelperImpl) getPVAndMatchAction(obj runtime.Unstructured, groupResource schema.GroupResource) (*resourcepolicies.Action, *corev1api.PersistentVolume, error) {
+	pvc := new(corev1api.PersistentVolumeClaim)
+	pv := new(corev1api.PersistentVolume)
+	var err error
+	var getPVErr error
+
+	if groupResource == kuberesource.PersistentVolumeClaims {
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &pvc); err != nil {
+			v.logger.WithError(err).Warn("fail to convert unstructured into PVC")
+			return nil, 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)
+			getPVErr = fmt.Errorf("fail to get PV for PVC %s: %w", pvc.Namespace+"/"+pvc.Name, errGetPVForPVC)
+		}
+	} else if groupResource == kuberesource.PersistentVolumes {
+		if err = runtime.DefaultUnstructuredConverter.FromUnstructured(obj.UnstructuredContent(), &pv); err != nil {
+			v.logger.WithError(err).Warn("fail to convert unstructured into PV")
+			return nil, 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).Warnf("fail to get VolumePolicy match action for %+v", vfd)
+			return nil, nil, err
+		}
+
+		return action, pv, getPVErr
+	}
+
+	return nil, pv, getPVErr
+}
+
 func (v *volumeHelperImpl) ShouldPerformSnapshot(obj runtime.Unstructured, groupResource schema.GroupResource) (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 snapshot
 	// if it is not snapshot then skip the code path for snapshotting the PV/PVC
@@ -316,117 +357,73 @@ 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
-		}
+	action, pv, err := v.getPVAndMatchAction(obj, groupResource)
+	if err != nil && !errors.Is(err, errGetPVForPVC) {
+		return false, err
 	}
 
-	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
-		}
+	metadata, metaErr := meta.Accessor(obj)
+	if metaErr != nil {
+		return false, metaErr
 	}
 
-	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
-					}
+	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 %s: %s as value for parameter %s is %s rather than the required %s",
+						groupResource.String(),
+						metadata.GetNamespace()+"/"+metadata.GetName(),
+						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
 			}
+			v.logger.Infof("performing custom action for %s: %s", groupResource.String(), metadata.GetNamespace()+"/"+metadata.GetName())
+			return true, nil
+		} else {
+			v.logger.Infof("Skipping custom action for %s: %s as the action type is %s",
+				groupResource.String(),
+				metadata.GetNamespace()+"/"+metadata.GetName(),
+				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)
+
+	if (groupResource == kuberesource.PersistentVolumeClaims) && (pv == nil) && errors.Is(err, errGetPVForPVC) {
 		return false, nil
 	}
 
-	v.logger.Infof("skipping custom action for pv %s due to no matching volume policy", pv.Name)
+	v.logger.Infof("skipping custom action for %s: %s due to no matching volume policy",
+		groupResource.String(), metadata.GetNamespace()+"/"+metadata.GetName())
 	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)
+	action, _, err := v.getPVAndMatchAction(obj, groupResource)
+	if err != nil {
+		if errors.Is(err, errGetPVForPVC) {
 			return false, "", nil, nil
 		}
+
+		return false, "", nil, err
 	}
 
-	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
-		}
+	metadata, metaErr := meta.Accessor(obj)
+	if metaErr != nil {
+		return false, "", nil, metaErr
 	}
 
-	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 action != nil {
+		v.logger.Infof("found matching action for %s: %s, returning parameters",
+			groupResource.String(), metadata.GetNamespace()+"/"+metadata.GetName())
 
-		// 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
-		}
+		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)
+	v.logger.Infof("no matching volume policy found for %s: %s, no parameters to return",
+		groupResource.String(), metadata.GetNamespace()+"/"+metadata.GetName())
+
 	return false, "", nil, nil
 }
 
@@ -486,3 +483,30 @@ func (v *volumeHelperImpl) getVolumeFromResource(resource any) (*corev1api.Persi
 	}
 	return nil, nil, fmt.Errorf("resource is not a PersistentVolume or Volume")
 }
+
+func (v *volumeHelperImpl) GetDataMoverFromActionParameters(obj runtime.Unstructured, groupResource schema.GroupResource) string {
+	action, _, err := v.getPVAndMatchAction(obj, groupResource)
+	if err != nil {
+		return ""
+	}
+
+	metadata, metaErr := meta.Accessor(obj)
+	if metaErr != nil {
+		return ""
+	}
+
+	if action != nil {
+		dataMover, err := action.GetDataMover()
+		if err != nil {
+			v.logger.WithError(err).Warn("fail to get data mover.")
+			return ""
+		}
+		v.logger.Infof("found matching action for %s: %s, returning data mover %s",
+			groupResource.String(), metadata.GetNamespace()+"/"+metadata.GetName(), dataMover)
+		return dataMover
+	}
+
+	v.logger.Debugf("no matching volume policy found for %s: %s, no data mover parameter to return",
+		groupResource.String(), metadata.GetNamespace()+"/"+metadata.GetName())
+	return ""
+}

internal/volumehelper/volume_policy_helper_test.go

@@ -1543,3 +1543,586 @@ func TestVolumeHelperImpl_ShouldPerformFSBackup_UnboundPVC(t *testing.T) {
 		})
 	}
 }
+
+func TestGetDataMoverFromActionParameters(t *testing.T) {
+	testCases := []struct {
+		name             string
+		inputObj         runtime.Object
+		groupResource    schema.GroupResource
+		resourcePolicies *resourcepolicies.ResourcePolicies
+		expected         string
+	}{
+		{
+			name:          "VolumePolicy match with dataMover parameter, returns dataMover string",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+							Parameters: map[string]any{
+								resourcepolicies.DataMoverParameter: "velero-block",
+							},
+						},
+					},
+				},
+			},
+			expected: "velero-block",
+		},
+		{
+			name:          "VolumePolicy match without dataMover parameter, returns default dataMover",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+							Parameters: map[string]any{
+								"otherParam": "value",
+							},
+						},
+					},
+				},
+			},
+			expected: "velero-fs",
+		},
+		{
+			name:          "VolumePolicy match with non-string dataMover parameter, returns empty string",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+							Parameters: map[string]any{
+								resourcepolicies.DataMoverParameter: 123,
+							},
+						},
+					},
+				},
+			},
+			expected: "",
+		},
+		{
+			name:          "VolumePolicy not match, returns empty string",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp3-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+							Parameters: map[string]any{
+								resourcepolicies.DataMoverParameter: "velero",
+							},
+						},
+					},
+				},
+			},
+			expected: "",
+		},
+		{
+			name:          "Error converting unstructured, returns empty string",
+			inputObj:      builder.ForPod("ns", "pod-1").Result(), // wrong type for PersistentVolumes
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+			},
+			expected: "",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t)
+
+			var p *resourcepolicies.Policies
+			if tc.resourcePolicies != nil {
+				p = &resourcepolicies.Policies{}
+				err := p.BuildPolicy(tc.resourcePolicies)
+				require.NoError(t, err)
+			}
+
+			vh := NewVolumeHelperImpl(
+				p,
+				ptr.To(true),
+				logrus.StandardLogger(),
+				fakeClient,
+				false,
+				false,
+			)
+
+			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
+			require.NoError(t, err)
+
+			actual := vh.GetDataMoverFromActionParameters(&unstructured.Unstructured{Object: obj}, tc.groupResource)
+			assert.Equal(t, tc.expected, actual)
+		})
+	}
+}
+
+func TestGetActionParameters(t *testing.T) {
+	testCases := []struct {
+		name             string
+		inputObj         runtime.Object
+		groupResource    schema.GroupResource
+		resourcePolicies *resourcepolicies.ResourcePolicies
+		expectedMatched  bool
+		expectedAction   string
+		expectedParams   map[string]any
+		expectedErr      bool
+	}{
+		{
+			name:          "VolumePolicy match with parameters, returns true, action type, parameters",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Custom,
+							Parameters: map[string]any{
+								"param1": "value1",
+							},
+						},
+					},
+				},
+			},
+			expectedMatched: true,
+			expectedAction:  string(resourcepolicies.Custom),
+			expectedParams: map[string]any{
+				"param1": "value1",
+			},
+			expectedErr: false,
+		},
+		{
+			name:          "VolumePolicy match without parameters, returns true, action type, nil parameters",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+						},
+					},
+				},
+			},
+			expectedMatched: true,
+			expectedAction:  string(resourcepolicies.Snapshot),
+			expectedParams:  nil,
+			expectedErr:     false,
+		},
+		{
+			name:          "VolumePolicy not match, returns false",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp3-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+						},
+					},
+				},
+			},
+			expectedMatched: false,
+			expectedAction:  "",
+			expectedParams:  nil,
+			expectedErr:     false,
+		},
+		{
+			name:          "PVC not having PV, returns false and no error",
+			inputObj:      builder.ForPersistentVolumeClaim("ns", "pvc-1").StorageClass("gp2-csi").Result(),
+			groupResource: kuberesource.PersistentVolumeClaims,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+			},
+			expectedMatched: false,
+			expectedAction:  "",
+			expectedParams:  nil,
+			expectedErr:     false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t)
+
+			var p *resourcepolicies.Policies
+			if tc.resourcePolicies != nil {
+				p = &resourcepolicies.Policies{}
+				err := p.BuildPolicy(tc.resourcePolicies)
+				require.NoError(t, err)
+			}
+
+			vh := NewVolumeHelperImpl(
+				p,
+				ptr.To(true),
+				logrus.StandardLogger(),
+				fakeClient,
+				false,
+				false,
+			)
+
+			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
+			require.NoError(t, err)
+
+			matched, actionType, params, err := vh.GetActionParameters(&unstructured.Unstructured{Object: obj}, tc.groupResource)
+			if tc.expectedErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			assert.Equal(t, tc.expectedMatched, matched)
+			assert.Equal(t, tc.expectedAction, actionType)
+			assert.Equal(t, tc.expectedParams, params)
+		})
+	}
+}
+
+func TestShouldPerformCustomAction(t *testing.T) {
+	testCases := []struct {
+		name             string
+		inputObj         runtime.Object
+		groupResource    schema.GroupResource
+		resourcePolicies *resourcepolicies.ResourcePolicies
+		matchParams      map[string]any
+		expected         bool
+		expectedErr      bool
+	}{
+		{
+			name:          "VolumePolicy match, action type is Custom, matchParams match exactly, returns true",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Custom,
+							Parameters: map[string]any{
+								"param1": "value1",
+								"param2": "value2",
+							},
+						},
+					},
+				},
+			},
+			matchParams: map[string]any{
+				"param1": "value1",
+			},
+			expected:    true,
+			expectedErr: false,
+		},
+		{
+			name:          "VolumePolicy match, action type is Custom, matchParams don't match (missing key), returns false",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Custom,
+							Parameters: map[string]any{
+								"param1": "value1",
+							},
+						},
+					},
+				},
+			},
+			matchParams: map[string]any{
+				"param2": "value2",
+			},
+			expected:    false,
+			expectedErr: false,
+		},
+		{
+			name:          "VolumePolicy match, action type is Custom, matchParams don't match (different value), returns false",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Custom,
+							Parameters: map[string]any{
+								"param1": "value1",
+							},
+						},
+					},
+				},
+			},
+			matchParams: map[string]any{
+				"param1": "value2",
+			},
+			expected:    false,
+			expectedErr: false,
+		},
+		{
+			name:          "VolumePolicy match, action type is not Custom, returns false",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp2-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+						},
+					},
+				},
+			},
+			matchParams: map[string]any{
+				"param1": "value1",
+			},
+			expected:    false,
+			expectedErr: false,
+		},
+		{
+			name:          "VolumePolicy not match, returns false",
+			inputObj:      builder.ForPersistentVolume("example-pv").StorageClass("gp3-csi").ClaimRef("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Custom,
+						},
+					},
+				},
+			},
+			matchParams: map[string]any{
+				"param1": "value1",
+			},
+			expected:    false,
+			expectedErr: false,
+		},
+		{
+			name:          "PVC not having PV, returns false and no error",
+			inputObj:      builder.ForPersistentVolumeClaim("ns", "pvc-1").StorageClass("gp2-csi").Result(),
+			groupResource: kuberesource.PersistentVolumeClaims,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+			},
+			matchParams: map[string]any{
+				"param1": "value1",
+			},
+			expected:    false,
+			expectedErr: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t)
+
+			var p *resourcepolicies.Policies
+			if tc.resourcePolicies != nil {
+				p = &resourcepolicies.Policies{}
+				err := p.BuildPolicy(tc.resourcePolicies)
+				require.NoError(t, err)
+			}
+
+			vh := NewVolumeHelperImpl(
+				p,
+				ptr.To(true),
+				logrus.StandardLogger(),
+				fakeClient,
+				false,
+				false,
+			)
+
+			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
+			require.NoError(t, err)
+
+			actual, err := vh.ShouldPerformCustomAction(&unstructured.Unstructured{Object: obj}, tc.groupResource, tc.matchParams)
+			if tc.expectedErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+			assert.Equal(t, tc.expected, actual)
+		})
+	}
+}
+
+func TestGetPVAndMatchAction(t *testing.T) {
+	testCases := []struct {
+		name             string
+		inputObj         runtime.Object
+		groupResource    schema.GroupResource
+		resourcePolicies *resourcepolicies.ResourcePolicies
+		expectedAction   *resourcepolicies.Action
+		expectedPVName   string
+		expectedErr      bool
+		expectedErrStr   string
+	}{
+		{
+			name:          "PVC with matching PV and VolumePolicy, returns action and PV",
+			inputObj:      builder.ForPersistentVolumeClaim("ns", "pvc-1").VolumeName("pv-1").Phase(corev1api.ClaimBound).Result(),
+			groupResource: kuberesource.PersistentVolumeClaims,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+						},
+					},
+				},
+			},
+			expectedAction: &resourcepolicies.Action{
+				Type: resourcepolicies.Snapshot,
+			},
+			expectedPVName: "pv-1",
+			expectedErr:    false,
+		},
+		{
+			name:          "PVC without matching PV, returns errGetPVForPVC",
+			inputObj:      builder.ForPersistentVolumeClaim("ns", "pvc-1").Result(),
+			groupResource: kuberesource.PersistentVolumeClaims,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+			},
+			expectedAction: nil,
+			expectedPVName: "",
+			expectedErr:    true,
+			expectedErrStr: "fail to get PV for PVC ns/pvc-1: fail to get PV for PVC",
+		},
+		{
+			name:          "PV with matching VolumePolicy, returns action and PV",
+			inputObj:      builder.ForPersistentVolume("pv-1").StorageClass("gp2-csi").Result(),
+			groupResource: kuberesource.PersistentVolumes,
+			resourcePolicies: &resourcepolicies.ResourcePolicies{
+				Version: "v1",
+				VolumePolicies: []resourcepolicies.VolumePolicy{
+					{
+						Conditions: map[string]any{
+							"storageClass": []string{"gp2-csi"},
+						},
+						Action: resourcepolicies.Action{
+							Type: resourcepolicies.Snapshot,
+						},
+					},
+				},
+			},
+			expectedAction: &resourcepolicies.Action{
+				Type: resourcepolicies.Snapshot,
+			},
+			expectedPVName: "pv-1",
+			expectedErr:    false,
+		},
+		{
+			name:           "PV without VolumePolicy, returns nil action and PV",
+			inputObj:       builder.ForPersistentVolume("pv-1").Result(),
+			groupResource:  kuberesource.PersistentVolumes,
+			expectedAction: nil,
+			expectedPVName: "pv-1",
+			expectedErr:    false,
+		},
+		{
+			name:           "Invalid object for PVC, returns error",
+			inputObj:       builder.ForPod("ns", "pod-1").Result(),
+			groupResource:  kuberesource.PersistentVolumeClaims,
+			expectedAction: nil,
+			expectedPVName: "",
+			expectedErr:    true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			pv := builder.ForPersistentVolume("pv-1").StorageClass("gp2-csi").Result()
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, pv)
+
+			var p *resourcepolicies.Policies
+			if tc.resourcePolicies != nil {
+				p = &resourcepolicies.Policies{}
+				err := p.BuildPolicy(tc.resourcePolicies)
+				require.NoError(t, err)
+			}
+
+			vh := NewVolumeHelperImpl(
+				p,
+				ptr.To(true),
+				logrus.StandardLogger(),
+				fakeClient,
+				false,
+				false,
+			)
+
+			obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.inputObj)
+			require.NoError(t, err)
+
+			action, outPV, err := vh.(*volumeHelperImpl).getPVAndMatchAction(&unstructured.Unstructured{Object: obj}, tc.groupResource)
+			if tc.expectedErr {
+				require.Error(t, err)
+				if tc.expectedErrStr != "" {
+					assert.Contains(t, err.Error(), tc.expectedErrStr)
+				}
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tc.expectedAction, action)
+				if tc.expectedPVName == "" {
+					assert.Nil(t, outPV)
+				} else {
+					require.NotNil(t, outPV)
+					assert.Equal(t, tc.expectedPVName, outPV.Name)
+				}
+			}
+		})
+	}
+}

pkg/backup/actions/csi/pvc_action.go

@@ -407,6 +407,8 @@ func (p *pvcBackupItemAction) Execute(
 			"Backup":         backup.Name,
 		})
 
+		dataMoverFromVolumePolicy := vh.GetDataMoverFromActionParameters(item, kuberesource.PersistentVolumeClaims)
+
 		dataUploadLog.Info("Starting data upload of backup")
 
 		dataUpload, err := createDataUpload(
@@ -418,6 +420,7 @@ func (p *pvcBackupItemAction) Execute(
 			operationID,
 			vsc,
 			fsType,
+			dataMoverFromVolumePolicy,
 		)
 		if err != nil {
 			dataUploadLog.WithError(err).Error("failed to submit DataUpload")
@@ -557,6 +560,7 @@ func newDataUpload(
 	operationID string,
 	vsc *snapshotv1api.VolumeSnapshotContent,
 	fsType string,
+	dataMoverFromVolumePolicy string,
 ) *velerov2alpha1.DataUpload {
 	parentSnapshot := ""
 
@@ -564,6 +568,11 @@ func newDataUpload(
 		parentSnapshot = veleroshared.DataUploadParentSnapshotNone
 	}
 
+	dataMover := backup.Spec.DataMover
+	if dataMoverFromVolumePolicy != "" {
+		dataMover = dataMoverFromVolumePolicy
+	}
+
 	dataUpload := &velerov2alpha1.DataUpload{
 		TypeMeta: metav1.TypeMeta{
 			APIVersion: velerov2alpha1.SchemeGroupVersion.String(),
@@ -596,7 +605,7 @@ func newDataUpload(
 				Driver:         vsc.Spec.Driver,
 			},
 			SourcePVC:             pvc.Name,
-			DataMover:             backup.Spec.DataMover,
+			DataMover:             dataMover,
 			BackupStorageLocation: backup.Spec.StorageLocation,
 			SourceNamespace:       pvc.Namespace,
 			OperationTimeout:      backup.Spec.CSISnapshotTimeout,
@@ -627,8 +636,9 @@ func createDataUpload(
 	operationID string,
 	vsc *snapshotv1api.VolumeSnapshotContent,
 	fsType string,
+	dataMoverFromVolumePolicy string,
 ) (*velerov2alpha1.DataUpload, error) {
-	dataUpload := newDataUpload(backup, vs, pvc, operationID, vsc, fsType)
+	dataUpload := newDataUpload(backup, vs, pvc, operationID, vsc, fsType, dataMoverFromVolumePolicy)
 
 	err := crClient.Create(ctx, dataUpload)
 	if err != nil {

pkg/backup/actions/csi/pvc_action_test.go

@@ -2229,12 +2229,13 @@ func TestGetOrCreateVolumeHelper(t *testing.T) {
 
 func TestNewDataUpload(t *testing.T) {
 	tests := []struct {
-		name                 string
-		backupType           velerov1api.BackupType
-		vsClassName          *string
-		uploaderConfig       *velerov1api.UploaderConfigForBackup
-		expectedParentSnap   string
-		expectedDataMoverCfg map[string]string
+		name                      string
+		backupType                velerov1api.BackupType
+		vsClassName               *string
+		uploaderConfig            *velerov1api.UploaderConfigForBackup
+		dataMoverFromVolumePolicy string
+		expectedParentSnap        string
+		expectedDataMoverCfg      map[string]string
 	}{
 		{
 			name:                 "Full backup type, no uploader config, no vs class name",
@@ -2262,6 +2263,15 @@ func TestNewDataUpload(t *testing.T) {
 			expectedParentSnap:   "",
 			expectedDataMoverCfg: nil,
 		},
+		{
+			name:                      "Default backup type, uploader config with 0 parallel files",
+			backupType:                "",
+			vsClassName:               ptr.To("test-vs-class"),
+			uploaderConfig:            &velerov1api.UploaderConfigForBackup{ParallelFilesUpload: 0},
+			dataMoverFromVolumePolicy: "velero-block",
+			expectedParentSnap:        "",
+			expectedDataMoverCfg:      nil,
+		},
 	}
 
 	for _, tc := range tests {
@@ -2310,7 +2320,7 @@ func TestNewDataUpload(t *testing.T) {
 			operationID := "test-op-id"
 			fsType := "ext4"
 
-			du := newDataUpload(backup, vs, pvc, operationID, vsc, fsType)
+			du := newDataUpload(backup, vs, pvc, operationID, vsc, fsType, tc.dataMoverFromVolumePolicy)
 
 			require.NotNil(t, du)
 			assert.Equal(t, velerov2alpha1.SchemeGroupVersion.String(), du.APIVersion)
@@ -2344,7 +2354,12 @@ func TestNewDataUpload(t *testing.T) {
 			}
 
 			assert.Equal(t, pvc.Name, du.Spec.SourcePVC)
-			assert.Equal(t, backup.Spec.DataMover, du.Spec.DataMover)
+			if tc.dataMoverFromVolumePolicy != "" {
+				assert.Equal(t, tc.dataMoverFromVolumePolicy, du.Spec.DataMover)
+			} else {
+				assert.Equal(t, backup.Spec.DataMover, du.Spec.DataMover)
+			}
+
 			assert.Equal(t, backup.Spec.StorageLocation, du.Spec.BackupStorageLocation)
 			assert.Equal(t, pvc.Namespace, du.Spec.SourceNamespace)
 			assert.Equal(t, backup.Spec.CSISnapshotTimeout, du.Spec.OperationTimeout)

pkg/util/volumehelper/volume_policy_helper.go

@@ -28,4 +28,5 @@ type VolumeHelper interface {
 	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)
 	GetSnapshotClass(obj runtime.Unstructured, groupResource schema.GroupResource) (string, error)
+	GetDataMoverFromActionParameters(obj runtime.Unstructured, groupResource schema.GroupResource) string
 }