From 79d9a5cde47980b61319b2cc6cd15c99c54ad2dc Mon Sep 17 00:00:00 2001 From: Xun Jiang Date: Thu, 6 Aug 2026 18:13:47 +0800 Subject: [PATCH] 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 --- changelogs/unreleased/10176-blackpiglet | 1 + internal/volumehelper/volume_policy_helper.go | 196 +++--- .../volumehelper/volume_policy_helper_test.go | 583 ++++++++++++++++++ pkg/backup/actions/csi/pvc_action.go | 14 +- pkg/backup/actions/csi/pvc_action_test.go | 31 +- pkg/util/volumehelper/volume_policy_helper.go | 1 + 6 files changed, 730 insertions(+), 96 deletions(-) create mode 100644 changelogs/unreleased/10176-blackpiglet diff --git a/changelogs/unreleased/10176-blackpiglet b/changelogs/unreleased/10176-blackpiglet new file mode 100644 index 000000000..301e7c8e1 --- /dev/null +++ b/changelogs/unreleased/10176-blackpiglet @@ -0,0 +1 @@ +Support to set data mover for the uploader from volume policy. \ No newline at end of file diff --git a/internal/volumehelper/volume_policy_helper.go b/internal/volumehelper/volume_policy_helper.go index 3259bdb43..7e23dd05f 100644 --- a/internal/volumehelper/volume_policy_helper.go +++ b/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 "" +} diff --git a/internal/volumehelper/volume_policy_helper_test.go b/internal/volumehelper/volume_policy_helper_test.go index 5e52ae73b..2c8a9151c 100644 --- a/internal/volumehelper/volume_policy_helper_test.go +++ b/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) + } + } + }) + } +} diff --git a/pkg/backup/actions/csi/pvc_action.go b/pkg/backup/actions/csi/pvc_action.go index 6998d13ce..b9debe031 100644 --- a/pkg/backup/actions/csi/pvc_action.go +++ b/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 { diff --git a/pkg/backup/actions/csi/pvc_action_test.go b/pkg/backup/actions/csi/pvc_action_test.go index e59591146..61141f2d5 100644 --- a/pkg/backup/actions/csi/pvc_action_test.go +++ b/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) diff --git a/pkg/util/volumehelper/volume_policy_helper.go b/pkg/util/volumehelper/volume_policy_helper.go index 6abdc73f8..a148b0432 100644 --- a/pkg/util/volumehelper/volume_policy_helper.go +++ b/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 }