Merge branch 'main' into resourcemodifier

Signed-off-by: Anshul Ahuja <anshul.ahu@gmail.com>

pkg/controller/backup_operations_controller.go

@@ -185,14 +185,15 @@ func (c *backupOperationsReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		operations.ChangesSinceUpdate = true
 	}
 
+	if len(operations.ErrsSinceUpdate) > 0 {
+		backup.Status.Phase = velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed
+	}
+
 	// if stillInProgress is false, backup moves to finalize phase and needs update
 	// if operations.ErrsSinceUpdate is not empty, then backup phase needs to change to
 	// BackupPhaseWaitingForPluginOperationsPartiallyFailed and needs update
 	// If the only changes are incremental progress, then no write is necessary, progress can remain in memory
 	if !stillInProgress {
-		if len(operations.ErrsSinceUpdate) > 0 {
-			backup.Status.Phase = velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed
-		}
 		if backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperations {
 			log.Infof("Marking backup %s Finalizing", backup.Name)
 			backup.Status.Phase = velerov1api.BackupPhaseFinalizing

pkg/controller/data_download_controller.go

@@ -44,6 +44,7 @@ import (
 	datamover "github.com/vmware-tanzu/velero/pkg/datamover"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
 	repository "github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
@@ -57,26 +58,30 @@ type DataDownloadReconciler struct {
 	logger            logrus.FieldLogger
 	credentialGetter  *credentials.CredentialGetter
 	fileSystem        filesystem.Interface
-	clock             clock.WithTickerAndDelayedExecution
+	Clock             clock.WithTickerAndDelayedExecution
 	restoreExposer    exposer.GenericRestoreExposer
 	nodeName          string
 	repositoryEnsurer *repository.Ensurer
 	dataPathMgr       *datapath.Manager
+	preparingTimeout  time.Duration
+	metrics           *metrics.ServerMetrics
 }
 
 func NewDataDownloadReconciler(client client.Client, kubeClient kubernetes.Interface,
-	repoEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter, nodeName string, logger logrus.FieldLogger) *DataDownloadReconciler {
+	repoEnsurer *repository.Ensurer, credentialGetter *credentials.CredentialGetter, nodeName string, preparingTimeout time.Duration, logger logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataDownloadReconciler {
 	return &DataDownloadReconciler{
 		client:            client,
 		kubeClient:        kubeClient,
 		logger:            logger.WithField("controller", "DataDownload"),
 		credentialGetter:  credentialGetter,
 		fileSystem:        filesystem.NewFileSystem(),
-		clock:             &clock.RealClock{},
+		Clock:             &clock.RealClock{},
 		nodeName:          nodeName,
 		repositoryEnsurer: repoEnsurer,
 		restoreExposer:    exposer.NewGenericRestoreExposer(kubeClient, logger),
 		dataPathMgr:       datapath.NewManager(1),
+		preparingTimeout:  preparingTimeout,
+		metrics:           metrics,
 	}
 }
 
@@ -132,9 +137,15 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 
 		log.Info("Data download is accepted")
 
+		if dd.Spec.Cancel {
+			log.Debugf("Data download is been canceled %s in Phase %s", dd.GetName(), dd.Status.Phase)
+			r.OnDataDownloadCancelled(ctx, dd.GetNamespace(), dd.GetName())
+			return ctrl.Result{}, nil
+		}
+
 		hostingPodLabels := map[string]string{velerov1api.DataDownloadLabel: dd.Name}
 
-		// ep.Expose() will trigger to create one pod whose volume is restored by a given volume snapshot,
+		// Expose() will trigger to create one pod whose volume is restored by a given volume snapshot,
 		// but the pod maybe is not in the same node of the current controller, so we need to return it here.
 		// And then only the controller who is in the same node could do the rest work.
 		err = r.restoreExposer.Expose(ctx, getDataDownloadOwnerObject(dd), dd.Spec.TargetVolume.PVC, dd.Spec.TargetVolume.Namespace, hostingPodLabels, dd.Spec.OperationTimeout.Duration)
@@ -143,9 +154,27 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		}
 		log.Info("Restore is exposed")
 
+		return ctrl.Result{}, nil
+	} else if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted {
+		if dd.Spec.Cancel {
+			log.Debugf("Data download is been canceled %s in Phase %s", dd.GetName(), dd.Status.Phase)
+			r.OnDataDownloadCancelled(ctx, dd.GetNamespace(), dd.GetName())
+		} else if dd.Status.StartTimestamp != nil {
+			if time.Since(dd.Status.StartTimestamp.Time) >= r.preparingTimeout {
+				r.onPrepareTimeout(ctx, dd)
+			}
+		}
+
 		return ctrl.Result{}, nil
 	} else if dd.Status.Phase == velerov2alpha1api.DataDownloadPhasePrepared {
 		log.Info("Data download is prepared")
+
+		if dd.Spec.Cancel {
+			log.Debugf("Data download is been canceled %s in Phase %s", dd.GetName(), dd.Status.Phase)
+			r.OnDataDownloadCancelled(ctx, dd.GetNamespace(), dd.GetName())
+			return ctrl.Result{}, nil
+		}
+
 		fsRestore := r.dataPathMgr.GetAsyncBR(dd.Name)
 
 		if fsRestore != nil {
@@ -184,7 +213,6 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		// Update status to InProgress
 		original := dd.DeepCopy()
 		dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseInProgress
-		dd.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
 		if err := r.client.Patch(ctx, dd, client.MergeFrom(original)); err != nil {
 			log.WithError(err).Error("Unable to update status to in progress")
 			return ctrl.Result{}, err
@@ -271,11 +299,12 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na
 
 	original := dd.DeepCopy()
 	dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseCompleted
-	dd.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 	if err := r.client.Patch(ctx, &dd, client.MergeFrom(original)); err != nil {
 		log.WithError(err).Error("error updating data download status")
 	} else {
 		log.Infof("Data download is marked as %s", dd.Status.Phase)
+		r.metrics.RegisterDataDownloadSuccess(r.nodeName)
 	}
 }
 
@@ -313,11 +342,13 @@ func (r *DataDownloadReconciler) OnDataDownloadCancelled(ctx context.Context, na
 		original := dd.DeepCopy()
 		dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseCanceled
 		if dd.Status.StartTimestamp.IsZero() {
-			dd.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
+			dd.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
 		}
-		dd.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+		dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 		if err := r.client.Patch(ctx, &dd, client.MergeFrom(original)); err != nil {
 			log.WithError(err).Error("error updating data download status")
+		} else {
+			r.metrics.RegisterDataDownloadCancel(r.nodeName)
 		}
 	}
 }
@@ -345,8 +376,15 @@ func (r *DataDownloadReconciler) OnDataDownloadProgress(ctx context.Context, nam
 // re-enqueue the previous related request once the related pod is in running status to keep going on the rest logic. and below logic will avoid handling the unwanted
 // pod status and also avoid block others CR handling
 func (r *DataDownloadReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	s := kube.NewPeriodicalEnqueueSource(r.logger, r.client, &velerov2alpha1api.DataDownloadList{}, preparingMonitorFrequency, kube.PeriodicalEnqueueSourceOption{})
+	gp := kube.NewGenericEventPredicate(func(object client.Object) bool {
+		dd := object.(*velerov2alpha1api.DataDownload)
+		return (dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted)
+	})
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&velerov2alpha1api.DataDownload{}).
+		Watches(s, nil, builder.WithPredicates(gp)).
 		Watches(&source.Kind{Type: &v1.Pod{}}, kube.EnqueueRequestsFromMapUpdateFunc(r.findSnapshotRestoreForPod),
 			builder.WithPredicates(predicate.Funcs{
 				UpdateFunc: func(ue event.UpdateEvent) bool {
@@ -382,15 +420,13 @@ func (r *DataDownloadReconciler) SetupWithManager(mgr ctrl.Manager) error {
 func (r *DataDownloadReconciler) findSnapshotRestoreForPod(podObj client.Object) []reconcile.Request {
 	pod := podObj.(*v1.Pod)
 
-	dd := &velerov2alpha1api.DataDownload{}
-	err := r.client.Get(context.Background(), types.NamespacedName{
-		Namespace: pod.Namespace,
-		Name:      pod.Labels[velerov1api.DataDownloadLabel],
-	}, dd)
-
+	dd, err := findDataDownloadByPod(r.client, *pod)
 	if err != nil {
 		r.logger.WithField("Restore pod", pod.Name).WithError(err).Error("unable to get DataDownload")
 		return []reconcile.Request{}
+	} else if dd == nil {
+		r.logger.WithField("Restore pod", pod.Name).Error("get empty DataDownload")
+		return []reconcile.Request{}
 	}
 
 	if dd.Status.Phase != velerov2alpha1api.DataDownloadPhaseAccepted {
@@ -400,9 +436,15 @@ func (r *DataDownloadReconciler) findSnapshotRestoreForPod(podObj client.Object)
 	requests := make([]reconcile.Request, 1)
 
 	r.logger.WithField("Restore pod", pod.Name).Infof("Preparing data download %s", dd.Name)
-	err = r.patchDataDownload(context.Background(), dd, r.prepareDataDownload)
-	if err != nil {
-		r.logger.WithField("Restore pod", pod.Name).WithError(err).Error("unable to patch data download")
+
+	// we don't expect anyone else update the CR during the Prepare process
+	updated, err := r.exclusiveUpdateDataDownload(context.Background(), dd, r.prepareDataDownload)
+	if err != nil || !updated {
+		r.logger.WithFields(logrus.Fields{
+			"Datadownload": dd.Name,
+			"Restore pod":  pod.Name,
+			"updated":      updated,
+		}).WithError(err).Warn("failed to patch datadownload, prepare will halt for this datadownload")
 		return []reconcile.Request{}
 	}
 
@@ -416,14 +458,28 @@ func (r *DataDownloadReconciler) findSnapshotRestoreForPod(podObj client.Object)
 	return requests
 }
 
-func (r *DataDownloadReconciler) patchDataDownload(ctx context.Context, req *velerov2alpha1api.DataDownload, mutate func(*velerov2alpha1api.DataDownload)) error {
-	original := req.DeepCopy()
-	mutate(req)
-	if err := r.client.Patch(ctx, req, client.MergeFrom(original)); err != nil {
-		return errors.Wrap(err, "error patching data download")
+func (r *DataDownloadReconciler) FindDataDownloads(ctx context.Context, cli client.Client, ns string) ([]*velerov2alpha1api.DataDownload, error) {
+	pods := &v1.PodList{}
+	var dataDownloads []*velerov2alpha1api.DataDownload
+	if err := cli.List(ctx, pods, &client.ListOptions{Namespace: ns}); err != nil {
+		r.logger.WithError(errors.WithStack(err)).Error("failed to list pods on current node")
+		return nil, errors.Wrapf(err, "failed to list pods on current node")
 	}
 
-	return nil
+	for _, pod := range pods.Items {
+		if pod.Spec.NodeName != r.nodeName {
+			r.logger.Debugf("Pod %s related data download will not handled by %s nodes", pod.GetName(), r.nodeName)
+			continue
+		}
+		dd, err := findDataDownloadByPod(cli, pod)
+		if err != nil {
+			r.logger.WithError(errors.WithStack(err)).Error("failed to get dataDownload by pod")
+			continue
+		} else if dd != nil {
+			dataDownloads = append(dataDownloads, dd)
+		}
+	}
+	return dataDownloads, nil
 }
 
 func (r *DataDownloadReconciler) prepareDataDownload(ssb *velerov2alpha1api.DataDownload) {
@@ -443,27 +499,76 @@ func (r *DataDownloadReconciler) updateStatusToFailed(ctx context.Context, dd *v
 	original := dd.DeepCopy()
 	dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseFailed
 	dd.Status.Message = errors.WithMessage(err, msg).Error()
-	dd.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 
 	if patchErr := r.client.Patch(ctx, dd, client.MergeFrom(original)); patchErr != nil {
 		log.WithError(patchErr).Error("error updating DataDownload status")
+	} else {
+		r.metrics.RegisterDataDownloadFailure(r.nodeName)
 	}
 
 	return err
 }
 
 func (r *DataDownloadReconciler) acceptDataDownload(ctx context.Context, dd *velerov2alpha1api.DataDownload) (bool, error) {
-	updated := dd.DeepCopy()
-	updated.Status.Phase = velerov2alpha1api.DataDownloadPhaseAccepted
+	r.logger.Infof("Accepting data download %s", dd.Name)
 
-	r.logger.Infof("Accepting snapshot restore %s", dd.Name)
 	// For all data download controller in each node-agent will try to update download CR, and only one controller will success,
 	// and the success one could handle later logic
+	succeeded, err := r.exclusiveUpdateDataDownload(ctx, dd, func(dd *velerov2alpha1api.DataDownload) {
+		dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseAccepted
+		dd.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
+	})
+
+	if err != nil {
+		return false, err
+	}
+
+	if succeeded {
+		r.logger.WithField("DataDownload", dd.Name).Infof("This datadownload has been accepted by %s", r.nodeName)
+		return true, nil
+	}
+
+	r.logger.WithField("DataDownload", dd.Name).Info("This datadownload has been accepted by others")
+	return false, nil
+}
+
+func (r *DataDownloadReconciler) onPrepareTimeout(ctx context.Context, dd *velerov2alpha1api.DataDownload) {
+	log := r.logger.WithField("DataDownload", dd.Name)
+
+	log.Info("Timeout happened for preparing datadownload")
+
+	succeeded, err := r.exclusiveUpdateDataDownload(ctx, dd, func(dd *velerov2alpha1api.DataDownload) {
+		dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseFailed
+		dd.Status.Message = "timeout on preparing data download"
+	})
+
+	if err != nil {
+		log.WithError(err).Warn("Failed to update datadownload")
+		return
+	}
+
+	if !succeeded {
+		log.Warn("Dataupload has been updated by others")
+		return
+	}
+
+	r.restoreExposer.CleanUp(ctx, getDataDownloadOwnerObject(dd))
+
+	log.Info("Dataupload has been cleaned up")
+
+	r.metrics.RegisterDataDownloadFailure(r.nodeName)
+}
+
+func (r *DataDownloadReconciler) exclusiveUpdateDataDownload(ctx context.Context, dd *velerov2alpha1api.DataDownload,
+	updateFunc func(*velerov2alpha1api.DataDownload)) (bool, error) {
+	updated := dd.DeepCopy()
+	updateFunc(updated)
+
 	err := r.client.Update(ctx, updated)
 	if err == nil {
 		return true, nil
 	} else if apierrors.IsConflict(err) {
-		r.logger.WithField("DataDownload", dd.Name).Error("This data download restore has been accepted by others")
 		return false, nil
 	} else {
 		return false, err
@@ -492,3 +597,20 @@ func getDataDownloadOwnerObject(dd *velerov2alpha1api.DataDownload) v1.ObjectRef
 		APIVersion: dd.APIVersion,
 	}
 }
+
+func findDataDownloadByPod(client client.Client, pod v1.Pod) (*velerov2alpha1api.DataDownload, error) {
+	if label, exist := pod.Labels[velerov1api.DataDownloadLabel]; exist {
+		dd := &velerov2alpha1api.DataDownload{}
+		err := client.Get(context.Background(), types.NamespacedName{
+			Namespace: pod.Namespace,
+			Name:      label,
+		}, dd)
+
+		if err != nil {
+			return nil, errors.Wrapf(err, "error to find DataDownload by pod %s/%s", pod.Namespace, pod.Name)
+		}
+		return dd, nil
+	}
+
+	return nil, nil
+}

pkg/controller/data_download_controller_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	clientgofake "k8s.io/client-go/kubernetes/fake"
@@ -45,6 +46,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	datapathmockes "github.com/vmware-tanzu/velero/pkg/datapath/mocks"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 
@@ -65,6 +67,29 @@ func dataDownloadBuilder() *builder.DataDownloadBuilder {
 }
 
 func initDataDownloadReconciler(objects []runtime.Object, needError ...bool) (*DataDownloadReconciler, error) {
+	var errs []error = make([]error, 4)
+	if len(needError) == 4 {
+		if needError[0] {
+			errs[0] = fmt.Errorf("Get error")
+		}
+
+		if needError[1] {
+			errs[1] = fmt.Errorf("Create error")
+		}
+
+		if needError[2] {
+			errs[2] = fmt.Errorf("Update error")
+		}
+
+		if needError[3] {
+			errs[3] = fmt.Errorf("Patch error")
+		}
+	}
+
+	return initDataDownloadReconcilerWithError(objects, errs...)
+}
+
+func initDataDownloadReconcilerWithError(objects []runtime.Object, needError ...error) (*DataDownloadReconciler, error) {
 	scheme := runtime.NewScheme()
 	err := velerov1api.AddToScheme(scheme)
 	if err != nil {
@@ -112,7 +137,7 @@ func initDataDownloadReconciler(objects []runtime.Object, needError ...bool) (*D
 	if err != nil {
 		return nil, err
 	}
-	return NewDataDownloadReconciler(fakeClient, fakeKubeClient, nil, &credentials.CredentialGetter{FromFile: credentialFileStore}, "test_node", velerotest.NewLogger()), nil
+	return NewDataDownloadReconciler(fakeClient, fakeKubeClient, nil, &credentials.CredentialGetter{FromFile: credentialFileStore}, "test_node", time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
 }
 
 func TestDataDownloadReconcile(t *testing.T) {
@@ -132,6 +157,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 		notMockCleanUp    bool
 		mockCancel        bool
 		mockClose         bool
+		expected          *velerov2alpha1api.DataDownload
 		expectedStatusMsg string
 		expectedResult    *ctrl.Result
 	}{
@@ -215,7 +241,7 @@ func TestDataDownloadReconcile(t *testing.T) {
 			dd:                builder.ForDataDownload("test-ns", dataDownloadName).Result(),
 			targetPVC:         builder.ForPersistentVolumeClaim("test-ns", "test-pvc").Result(),
 			needErrs:          []bool{true, false, false, false},
-			expectedStatusMsg: "Create error",
+			expectedStatusMsg: "Get error",
 		},
 		{
 			name:      "Unsupported dataDownload type",
@@ -246,6 +272,11 @@ func TestDataDownloadReconcile(t *testing.T) {
 			expectedStatusMsg: "Error to expose restore exposer",
 			isExposeErr:       true,
 		},
+		{
+			name:     "prepare timeout",
+			dd:       dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).StartTimestamp(&metav1.Time{Time: time.Now().Add(-time.Minute * 5)}).Result(),
+			expected: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseFailed).Result(),
+		},
 	}
 
 	for _, test := range tests {
@@ -345,6 +376,11 @@ func TestDataDownloadReconcile(t *testing.T) {
 				Namespace: test.dd.Namespace,
 			}, &dd)
 
+			if test.expected != nil {
+				require.NoError(t, err)
+				assert.Equal(t, dd.Status.Phase, test.expected.Status.Phase)
+			}
+
 			if test.isGetExposeErr {
 				assert.Contains(t, dd.Status.Message, test.expectedStatusMsg)
 			}
@@ -550,6 +586,14 @@ func TestFindDataDownloadForPod(t *testing.T) {
 				assert.Equal(t, du.Namespace, requests[0].Namespace)
 				assert.Equal(t, du.Name, requests[0].Name)
 			},
+		}, {
+			name: "no selected label found for pod",
+			du:   dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Result(),
+			pod:  builder.ForPod(velerov1api.DefaultNamespace, dataDownloadName).Result(),
+			checkFunc: func(du *velerov2alpha1api.DataDownload, requests []reconcile.Request) {
+				// Assert that the function returns a single request
+				assert.Empty(t, requests)
+			},
 		}, {
 			name: "no matched pod",
 			du:   dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Result(),
@@ -559,7 +603,7 @@ func TestFindDataDownloadForPod(t *testing.T) {
 			},
 		},
 		{
-			name: "dataDownload not accepte",
+			name: "dataDownload not accept",
 			du:   dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseInProgress).Result(),
 			pod:  builder.ForPod(velerov1api.DefaultNamespace, dataDownloadName).Labels(map[string]string{velerov1api.DataDownloadLabel: dataDownloadName}).Result(),
 			checkFunc: func(du *velerov2alpha1api.DataDownload, requests []reconcile.Request) {
@@ -580,3 +624,93 @@ func TestFindDataDownloadForPod(t *testing.T) {
 		}
 	}
 }
+
+func TestAcceptDataDownload(t *testing.T) {
+	tests := []struct {
+		name        string
+		dd          *velerov2alpha1api.DataDownload
+		needErrs    []error
+		succeeded   bool
+		expectedErr string
+	}{
+		{
+			name:        "update fail",
+			dd:          dataDownloadBuilder().Result(),
+			needErrs:    []error{nil, nil, fmt.Errorf("fake-update-error"), nil},
+			expectedErr: "fake-update-error",
+		},
+		{
+			name:     "accepted by others",
+			dd:       dataDownloadBuilder().Result(),
+			needErrs: []error{nil, nil, &fakeAPIStatus{metav1.StatusReasonConflict}, nil},
+		},
+		{
+			name:      "succeed",
+			dd:        dataDownloadBuilder().Result(),
+			needErrs:  []error{nil, nil, nil, nil},
+			succeeded: true,
+		},
+	}
+	for _, test := range tests {
+		ctx := context.Background()
+		r, err := initDataDownloadReconcilerWithError(nil, test.needErrs...)
+		require.NoError(t, err)
+
+		err = r.client.Create(ctx, test.dd)
+		require.NoError(t, err)
+
+		succeeded, err := r.acceptDataDownload(ctx, test.dd)
+		assert.Equal(t, test.succeeded, succeeded)
+		if test.expectedErr == "" {
+			assert.NoError(t, err)
+		} else {
+			assert.EqualError(t, err, test.expectedErr)
+		}
+	}
+}
+
+func TestOnDdPrepareTimeout(t *testing.T) {
+	tests := []struct {
+		name     string
+		dd       *velerov2alpha1api.DataDownload
+		needErrs []error
+		expected *velerov2alpha1api.DataDownload
+	}{
+		{
+			name:     "update fail",
+			dd:       dataDownloadBuilder().Result(),
+			needErrs: []error{nil, nil, fmt.Errorf("fake-update-error"), nil},
+			expected: dataDownloadBuilder().Result(),
+		},
+		{
+			name:     "update interrupted",
+			dd:       dataDownloadBuilder().Result(),
+			needErrs: []error{nil, nil, &fakeAPIStatus{metav1.StatusReasonConflict}, nil},
+			expected: dataDownloadBuilder().Result(),
+		},
+		{
+			name:     "succeed",
+			dd:       dataDownloadBuilder().Result(),
+			needErrs: []error{nil, nil, nil, nil},
+			expected: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseFailed).Result(),
+		},
+	}
+	for _, test := range tests {
+		ctx := context.Background()
+		r, err := initDataDownloadReconcilerWithError(nil, test.needErrs...)
+		require.NoError(t, err)
+
+		err = r.client.Create(ctx, test.dd)
+		require.NoError(t, err)
+
+		r.onPrepareTimeout(ctx, test.dd)
+
+		dd := velerov2alpha1api.DataDownload{}
+		_ = r.client.Get(ctx, kbclient.ObjectKey{
+			Name:      test.dd.Name,
+			Namespace: test.dd.Namespace,
+		}, &dd)
+
+		assert.Equal(t, test.expected.Status.Phase, dd.Status.Phase)
+	}
+}

pkg/controller/data_upload_controller.go

@@ -46,6 +46,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/datamover"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
@@ -55,29 +56,33 @@ import (
 const dataMoverType string = "velero"
 const dataUploadDownloadRequestor string = "snapshot-data-upload-download"
 
+const preparingMonitorFrequency time.Duration = time.Minute
+
 // DataUploadReconciler reconciles a DataUpload object
 type DataUploadReconciler struct {
 	client              client.Client
 	kubeClient          kubernetes.Interface
 	csiSnapshotClient   snapshotter.SnapshotV1Interface
 	repoEnsurer         *repository.Ensurer
-	clock               clocks.WithTickerAndDelayedExecution
+	Clock               clocks.WithTickerAndDelayedExecution
 	credentialGetter    *credentials.CredentialGetter
 	nodeName            string
 	fileSystem          filesystem.Interface
 	logger              logrus.FieldLogger
 	snapshotExposerList map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer
 	dataPathMgr         *datapath.Manager
+	preparingTimeout    time.Duration
+	metrics             *metrics.ServerMetrics
 }
 
 func NewDataUploadReconciler(client client.Client, kubeClient kubernetes.Interface,
 	csiSnapshotClient snapshotter.SnapshotV1Interface, repoEnsurer *repository.Ensurer, clock clocks.WithTickerAndDelayedExecution,
-	cred *credentials.CredentialGetter, nodeName string, fs filesystem.Interface, log logrus.FieldLogger) *DataUploadReconciler {
+	cred *credentials.CredentialGetter, nodeName string, fs filesystem.Interface, preparingTimeout time.Duration, log logrus.FieldLogger, metrics *metrics.ServerMetrics) *DataUploadReconciler {
 	return &DataUploadReconciler{
 		client:              client,
 		kubeClient:          kubeClient,
 		csiSnapshotClient:   csiSnapshotClient,
-		clock:               clock,
+		Clock:               clock,
 		credentialGetter:    cred,
 		nodeName:            nodeName,
 		fileSystem:          fs,
@@ -85,6 +90,8 @@ func NewDataUploadReconciler(client client.Client, kubeClient kubernetes.Interfa
 		repoEnsurer:         repoEnsurer,
 		snapshotExposerList: map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{velerov2alpha1api.SnapshotTypeCSI: exposer.NewCSISnapshotExposer(kubeClient, csiSnapshotClient, log)},
 		dataPathMgr:         datapath.NewManager(1),
+		preparingTimeout:    preparingTimeout,
+		metrics:             metrics,
 	}
 }
 
@@ -134,18 +141,39 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 
 		log.Info("Data upload is accepted")
 
+		if du.Spec.Cancel {
+			r.OnDataUploadCancelled(ctx, du.GetNamespace(), du.GetName())
+			return ctrl.Result{}, nil
+		}
+
 		exposeParam := r.setupExposeParam(&du)
 
 		if err := ep.Expose(ctx, getOwnerObject(&du), exposeParam); err != nil {
 			return r.errorOut(ctx, &du, err, "error to expose snapshot", log)
 		}
 		log.Info("Snapshot is exposed")
-		// ep.Expose() will trigger to create one pod whose volume is restored by a given volume snapshot,
+		// Expose() will trigger to create one pod whose volume is restored by a given volume snapshot,
 		// but the pod maybe is not in the same node of the current controller, so we need to return it here.
 		// And then only the controller who is in the same node could do the rest work.
+		return ctrl.Result{}, nil
+	} else if du.Status.Phase == velerov2alpha1api.DataUploadPhaseAccepted {
+		if du.Spec.Cancel {
+			r.OnDataUploadCancelled(ctx, du.GetNamespace(), du.GetName())
+		} else if du.Status.StartTimestamp != nil {
+			if time.Since(du.Status.StartTimestamp.Time) >= r.preparingTimeout {
+				r.onPrepareTimeout(ctx, &du)
+			}
+		}
+
 		return ctrl.Result{}, nil
 	} else if du.Status.Phase == velerov2alpha1api.DataUploadPhasePrepared {
 		log.Info("Data upload is prepared")
+
+		if du.Spec.Cancel {
+			r.OnDataUploadCancelled(ctx, du.GetNamespace(), du.GetName())
+			return ctrl.Result{}, nil
+		}
+
 		fsBackup := r.dataPathMgr.GetAsyncBR(du.Name)
 		if fsBackup != nil {
 			log.Info("Cancellable data path is already started")
@@ -183,7 +211,6 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		// Update status to InProgress
 		original := du.DeepCopy()
 		du.Status.Phase = velerov2alpha1api.DataUploadPhaseInProgress
-		du.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
 		if err := r.client.Patch(ctx, &du, client.MergeFrom(original)); err != nil {
 			return r.errorOut(ctx, &du, err, "error updating dataupload status", log)
 		}
@@ -277,15 +304,17 @@ func (r *DataUploadReconciler) OnDataUploadCompleted(ctx context.Context, namesp
 	du.Status.Path = result.Backup.Source.ByPath
 	du.Status.Phase = velerov2alpha1api.DataUploadPhaseCompleted
 	du.Status.SnapshotID = result.Backup.SnapshotID
-	du.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	du.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 	if result.Backup.EmptySnapshot {
 		du.Status.Message = "volume was empty so no data was upload"
 	}
 
 	if err := r.client.Patch(ctx, &du, client.MergeFrom(original)); err != nil {
 		log.WithError(err).Error("error updating DataUpload status")
+	} else {
+		log.Info("Data upload completed")
+		r.metrics.RegisterDataUploadSuccess(r.nodeName)
 	}
-	log.Info("Data upload completed")
 }
 
 func (r *DataUploadReconciler) OnDataUploadFailed(ctx context.Context, namespace string, duName string, err error) {
@@ -331,11 +360,13 @@ func (r *DataUploadReconciler) OnDataUploadCancelled(ctx context.Context, namesp
 		original := du.DeepCopy()
 		du.Status.Phase = velerov2alpha1api.DataUploadPhaseCanceled
 		if du.Status.StartTimestamp.IsZero() {
-			du.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
+			du.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
 		}
-		du.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+		du.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 		if err := r.client.Patch(ctx, &du, client.MergeFrom(original)); err != nil {
 			log.WithError(err).Error("error updating DataUpload status")
+		} else {
+			r.metrics.RegisterDataUploadCancel(r.nodeName)
 		}
 	}
 }
@@ -363,8 +394,15 @@ func (r *DataUploadReconciler) OnDataUploadProgress(ctx context.Context, namespa
 // re-enqueue the previous related request once the related pod is in running status to keep going on the rest logic. and below logic will avoid handling the unwanted
 // pod status and also avoid block others CR handling
 func (r *DataUploadReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	s := kube.NewPeriodicalEnqueueSource(r.logger, r.client, &velerov2alpha1api.DataUploadList{}, preparingMonitorFrequency, kube.PeriodicalEnqueueSourceOption{})
+	gp := kube.NewGenericEventPredicate(func(object client.Object) bool {
+		du := object.(*velerov2alpha1api.DataUpload)
+		return (du.Status.Phase == velerov2alpha1api.DataUploadPhaseAccepted)
+	})
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&velerov2alpha1api.DataUpload{}).
+		Watches(s, nil, builder.WithPredicates(gp)).
 		Watches(&source.Kind{Type: &corev1.Pod{}}, kube.EnqueueRequestsFromMapUpdateFunc(r.findDataUploadForPod),
 			builder.WithPredicates(predicate.Funcs{
 				UpdateFunc: func(ue event.UpdateEvent) bool {
@@ -399,16 +437,13 @@ func (r *DataUploadReconciler) SetupWithManager(mgr ctrl.Manager) error {
 
 func (r *DataUploadReconciler) findDataUploadForPod(podObj client.Object) []reconcile.Request {
 	pod := podObj.(*corev1.Pod)
-
-	du := &velerov2alpha1api.DataUpload{}
-	err := r.client.Get(context.Background(), types.NamespacedName{
-		Namespace: pod.Namespace,
-		Name:      pod.Labels[velerov1api.DataUploadLabel],
-	}, du)
-
+	du, err := findDataUploadByPod(r.client, *pod)
 	if err != nil {
 		r.logger.WithField("Backup pod", pod.Name).WithError(err).Error("unable to get dataupload")
 		return []reconcile.Request{}
+	} else if du == nil {
+		r.logger.WithField("Backup pod", pod.Name).Error("get empty DataUpload")
+		return []reconcile.Request{}
 	}
 
 	if du.Status.Phase != velerov2alpha1api.DataUploadPhaseAccepted {
@@ -416,8 +451,15 @@ func (r *DataUploadReconciler) findDataUploadForPod(podObj client.Object) []reco
 	}
 
 	r.logger.WithField("Backup pod", pod.Name).Infof("Preparing dataupload %s", du.Name)
-	if err := r.patchDataUpload(context.Background(), du, r.prepareDataUpload); err != nil {
-		r.logger.WithField("Backup pod", pod.Name).WithError(err).Error("failed to patch dataupload")
+
+	// we don't expect anyone else update the CR during the Prepare process
+	updated, err := r.exclusiveUpdateDataUpload(context.Background(), du, r.prepareDataUpload)
+	if err != nil || !updated {
+		r.logger.WithFields(logrus.Fields{
+			"Dataupload": du.Name,
+			"Backup pod": pod.Name,
+			"updated":    updated,
+		}).WithError(err).Warn("failed to patch dataupload, prepare will halt for this dataupload")
 		return []reconcile.Request{}
 	}
 
@@ -430,14 +472,28 @@ func (r *DataUploadReconciler) findDataUploadForPod(podObj client.Object) []reco
 	return []reconcile.Request{requests}
 }
 
-func (r *DataUploadReconciler) patchDataUpload(ctx context.Context, req *velerov2alpha1api.DataUpload, mutate func(*velerov2alpha1api.DataUpload)) error {
-	original := req.DeepCopy()
-	mutate(req)
-	if err := r.client.Patch(ctx, req, client.MergeFrom(original)); err != nil {
-		return errors.Wrap(err, "error patching DataUpload")
+func (r *DataUploadReconciler) FindDataUploads(ctx context.Context, cli client.Client, ns string) ([]velerov2alpha1api.DataUpload, error) {
+	pods := &corev1.PodList{}
+	var dataUploads []velerov2alpha1api.DataUpload
+	if err := cli.List(ctx, pods, &client.ListOptions{Namespace: ns}); err != nil {
+		r.logger.WithError(errors.WithStack(err)).Error("failed to list pods on current node")
+		return nil, errors.Wrapf(err, "failed to list pods on current node")
 	}
 
-	return nil
+	for _, pod := range pods.Items {
+		if pod.Spec.NodeName != r.nodeName {
+			r.logger.Debugf("Pod %s related data upload will not handled by %s nodes", pod.GetName(), r.nodeName)
+			continue
+		}
+		du, err := findDataUploadByPod(cli, pod)
+		if err != nil {
+			r.logger.WithError(errors.WithStack(err)).Error("failed to get dataUpload by pod")
+			continue
+		} else if du != nil {
+			dataUploads = append(dataUploads, *du)
+		}
+	}
+	return dataUploads, nil
 }
 
 func (r *DataUploadReconciler) prepareDataUpload(du *velerov2alpha1api.DataUpload) {
@@ -463,31 +519,89 @@ func (r *DataUploadReconciler) updateStatusToFailed(ctx context.Context, du *vel
 	du.Status.Phase = velerov2alpha1api.DataUploadPhaseFailed
 	du.Status.Message = errors.WithMessage(err, msg).Error()
 	if du.Status.StartTimestamp.IsZero() {
-		du.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
+		du.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
 	}
 
-	du.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	du.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
 	if patchErr := r.client.Patch(ctx, du, client.MergeFrom(original)); patchErr != nil {
 		log.WithError(patchErr).Error("error updating DataUpload status")
+	} else {
+		r.metrics.RegisterDataUploadFailure(r.nodeName)
 	}
 
 	return err
 }
 
 func (r *DataUploadReconciler) acceptDataUpload(ctx context.Context, du *velerov2alpha1api.DataUpload) (bool, error) {
-	updated := du.DeepCopy()
-	updated.Status.Phase = velerov2alpha1api.DataUploadPhaseAccepted
-
-	r.logger.Infof("Accepting snapshot backup %s", du.Name)
+	r.logger.Infof("Accepting data upload %s", du.Name)
 
 	// For all data upload controller in each node-agent will try to update dataupload CR, and only one controller will success,
 	// and the success one could handle later logic
+	succeeded, err := r.exclusiveUpdateDataUpload(ctx, du, func(du *velerov2alpha1api.DataUpload) {
+		du.Status.Phase = velerov2alpha1api.DataUploadPhaseAccepted
+		du.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
+	})
+
+	if err != nil {
+		return false, err
+	}
+
+	if succeeded {
+		r.logger.WithField("Dataupload", du.Name).Infof("This datauplod has been accepted by %s", r.nodeName)
+		return true, nil
+	}
+
+	r.logger.WithField("Dataupload", du.Name).Info("This datauplod has been accepted by others")
+	return false, nil
+}
+
+func (r *DataUploadReconciler) onPrepareTimeout(ctx context.Context, du *velerov2alpha1api.DataUpload) {
+	log := r.logger.WithField("Dataupload", du.Name)
+
+	log.Info("Timeout happened for preparing dataupload")
+
+	succeeded, err := r.exclusiveUpdateDataUpload(ctx, du, func(du *velerov2alpha1api.DataUpload) {
+		du.Status.Phase = velerov2alpha1api.DataUploadPhaseFailed
+		du.Status.Message = "timeout on preparing data upload"
+	})
+
+	if err != nil {
+		log.WithError(err).Warn("Failed to update dataupload")
+		return
+	}
+
+	if !succeeded {
+		log.Warn("Dataupload has been updated by others")
+		return
+	}
+
+	ep, ok := r.snapshotExposerList[du.Spec.SnapshotType]
+	if !ok {
+		log.WithError(fmt.Errorf("%v type of snapshot exposer is not exist", du.Spec.SnapshotType)).
+			Warn("Failed to clean up resources on canceled")
+	} else {
+		var volumeSnapshotName string
+		if du.Spec.SnapshotType == velerov2alpha1api.SnapshotTypeCSI { // Other exposer should have another condition
+			volumeSnapshotName = du.Spec.CSISnapshot.VolumeSnapshot
+		}
+
+		ep.CleanUp(ctx, getOwnerObject(du), volumeSnapshotName, du.Spec.SourceNamespace)
+
+		log.Info("Dataupload has been cleaned up")
+	}
+
+	r.metrics.RegisterDataUploadFailure(r.nodeName)
+}
+
+func (r *DataUploadReconciler) exclusiveUpdateDataUpload(ctx context.Context, du *velerov2alpha1api.DataUpload,
+	updateFunc func(*velerov2alpha1api.DataUpload)) (bool, error) {
+	updated := du.DeepCopy()
+	updateFunc(updated)
+
 	err := r.client.Update(ctx, updated)
 	if err == nil {
-		r.logger.WithField("Dataupload", du.Name).Infof("This datauplod backup has been accepted by %s", r.nodeName)
 		return true, nil
 	} else if apierrors.IsConflict(err) {
-		r.logger.WithField("Dataupload", du.Name).Info("This datauplod backup has been accepted by others")
 		return false, nil
 	} else {
 		return false, err
@@ -536,3 +650,19 @@ func getOwnerObject(du *velerov2alpha1api.DataUpload) corev1.ObjectReference {
 		APIVersion: du.APIVersion,
 	}
 }
+
+func findDataUploadByPod(client client.Client, pod corev1.Pod) (*velerov2alpha1api.DataUpload, error) {
+	if label, exist := pod.Labels[velerov1api.DataUploadLabel]; exist {
+		du := &velerov2alpha1api.DataUpload{}
+		err := client.Get(context.Background(), types.NamespacedName{
+			Namespace: pod.Namespace,
+			Name:      label,
+		}, du)
+
+		if err != nil {
+			return nil, errors.Wrapf(err, "error to find DataUpload by pod %s/%s", pod.Namespace, pod.Name)
+		}
+		return du, nil
+	}
+	return nil, nil
+}

pkg/controller/data_upload_controller_test.go

@@ -46,6 +46,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/datapath"
 	"github.com/vmware-tanzu/velero/pkg/exposer"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/repository"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/uploader"
@@ -58,45 +59,68 @@ const fakeSnapshotType velerov2alpha1api.SnapshotType = "fake-snapshot"
 
 type FakeClient struct {
 	kbclient.Client
-	getError    bool
-	createError bool
-	updateError bool
-	patchError  bool
+	getError    error
+	createError error
+	updateError error
+	patchError  error
 }
 
 func (c *FakeClient) Get(ctx context.Context, key kbclient.ObjectKey, obj kbclient.Object) error {
-	if c.getError {
-		return fmt.Errorf("Create error")
+	if c.getError != nil {
+		return c.getError
 	}
 
 	return c.Client.Get(ctx, key, obj)
 }
 
 func (c *FakeClient) Create(ctx context.Context, obj kbclient.Object, opts ...kbclient.CreateOption) error {
-	if c.createError {
-		return fmt.Errorf("Create error")
+	if c.createError != nil {
+		return c.createError
 	}
 
 	return c.Client.Create(ctx, obj, opts...)
 }
 
 func (c *FakeClient) Update(ctx context.Context, obj kbclient.Object, opts ...kbclient.UpdateOption) error {
-	if c.updateError {
-		return fmt.Errorf("Update error")
+	if c.updateError != nil {
+		return c.updateError
 	}
 
 	return c.Client.Update(ctx, obj, opts...)
 }
 
 func (c *FakeClient) Patch(ctx context.Context, obj kbclient.Object, patch kbclient.Patch, opts ...kbclient.PatchOption) error {
-	if c.patchError {
-		return fmt.Errorf("Patch error")
+	if c.patchError != nil {
+		return c.patchError
 	}
 
 	return c.Client.Patch(ctx, obj, patch, opts...)
 }
 
 func initDataUploaderReconciler(needError ...bool) (*DataUploadReconciler, error) {
+	var errs []error = make([]error, 4)
+	if len(needError) == 4 {
+		if needError[0] {
+			errs[0] = fmt.Errorf("Get error")
+		}
+
+		if needError[1] {
+			errs[1] = fmt.Errorf("Create error")
+		}
+
+		if needError[2] {
+			errs[2] = fmt.Errorf("Update error")
+		}
+
+		if needError[3] {
+			errs[3] = fmt.Errorf("Patch error")
+		}
+	}
+
+	return initDataUploaderReconcilerWithError(errs...)
+}
+
+func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconciler, error) {
 	vscName := "fake-vsc"
 	vsObject := &snapshotv1api.VolumeSnapshot{
 		ObjectMeta: metav1.ObjectMeta{
@@ -170,7 +194,7 @@ func initDataUploaderReconciler(needError ...bool) (*DataUploadReconciler, error
 		return nil, err
 	}
 	return NewDataUploadReconciler(fakeClient, fakeKubeClient, fakeSnapshotClient.SnapshotV1(), nil,
-		testclocks.NewFakeClock(now), &credentials.CredentialGetter{FromFile: credentialFileStore}, "test_node", fakeFS, velerotest.NewLogger()), nil
+		testclocks.NewFakeClock(now), &credentials.CredentialGetter{FromFile: credentialFileStore}, "test_node", fakeFS, time.Minute*5, velerotest.NewLogger(), metrics.NewServerMetrics()), nil
 }
 
 func dataUploadBuilder() *builder.DataUploadBuilder {
@@ -277,7 +301,7 @@ func TestReconcile(t *testing.T) {
 			expectedProcessed: false,
 			expected:          nil,
 			expectedRequeue:   ctrl.Result{},
-			expectedErrMsg:    "getting DataUpload: Create error",
+			expectedErrMsg:    "getting DataUpload: Get error",
 			needErrs:          []bool{true, false, false, false},
 		}, {
 			name:              "Unsupported data mover type",
@@ -334,11 +358,16 @@ func TestReconcile(t *testing.T) {
 			name:              "runCancelableDataUpload is concurrent limited",
 			dataMgr:           datapath.NewManager(0),
 			pod:               builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).Result(),
-			du:                dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhasePrepared).SnapshotType(fakeSnapshotType).Cancel(true).Result(),
+			du:                dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhasePrepared).SnapshotType(fakeSnapshotType).Result(),
 			expectedProcessed: false,
 			expected:          dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhasePrepared).Result(),
 			expectedRequeue:   ctrl.Result{Requeue: true, RequeueAfter: time.Minute},
 		},
+		{
+			name:     "prepare timeout",
+			du:       dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).SnapshotType(fakeSnapshotType).StartTimestamp(&metav1.Time{Time: time.Now().Add(-time.Minute * 5)}).Result(),
+			expected: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
+		},
 	}
 
 	for _, test := range tests {
@@ -369,7 +398,7 @@ func TestReconcile(t *testing.T) {
 			}
 
 			if test.du.Spec.SnapshotType == fakeSnapshotType {
-				r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{fakeSnapshotType: &fakeSnapshotExposer{r.client, r.clock}}
+				r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{fakeSnapshotType: &fakeSnapshotExposer{r.client, r.Clock}}
 			} else if test.du.Spec.SnapshotType == velerov2alpha1api.SnapshotTypeCSI {
 				r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{velerov2alpha1api.SnapshotTypeCSI: exposer.NewCSISnapshotExposer(r.kubeClient, r.csiSnapshotClient, velerotest.NewLogger())}
 			}
@@ -378,7 +407,7 @@ func TestReconcile(t *testing.T) {
 				return &fakeDataUploadFSBR{
 					du:         test.du,
 					kubeClient: r.client,
-					clock:      r.clock,
+					clock:      r.Clock,
 				}
 			}
 
@@ -569,6 +598,14 @@ func TestFindDataUploadForPod(t *testing.T) {
 				assert.Equal(t, du.Namespace, requests[0].Namespace)
 				assert.Equal(t, du.Name, requests[0].Name)
 			},
+		}, {
+			name: "no selected label found for pod",
+			du:   dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).Result(),
+			pod:  builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Result(),
+			checkFunc: func(du *velerov2alpha1api.DataUpload, requests []reconcile.Request) {
+				// Assert that the function returns a single request
+				assert.Empty(t, requests)
+			},
 		}, {
 			name: "no matched pod",
 			du:   dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).Result(),
@@ -599,3 +636,107 @@ func TestFindDataUploadForPod(t *testing.T) {
 		}
 	}
 }
+
+type fakeAPIStatus struct {
+	reason metav1.StatusReason
+}
+
+func (f *fakeAPIStatus) Status() metav1.Status {
+	return metav1.Status{
+		Reason: f.reason,
+	}
+}
+
+func (f *fakeAPIStatus) Error() string {
+	return string(f.reason)
+}
+
+func TestAcceptDataUpload(t *testing.T) {
+	tests := []struct {
+		name        string
+		du          *velerov2alpha1api.DataUpload
+		needErrs    []error
+		succeeded   bool
+		expectedErr string
+	}{
+		{
+			name:        "update fail",
+			du:          dataUploadBuilder().Result(),
+			needErrs:    []error{nil, nil, fmt.Errorf("fake-update-error"), nil},
+			expectedErr: "fake-update-error",
+		},
+		{
+			name:     "accepted by others",
+			du:       dataUploadBuilder().Result(),
+			needErrs: []error{nil, nil, &fakeAPIStatus{metav1.StatusReasonConflict}, nil},
+		},
+		{
+			name:      "succeed",
+			du:        dataUploadBuilder().Result(),
+			needErrs:  []error{nil, nil, nil, nil},
+			succeeded: true,
+		},
+	}
+	for _, test := range tests {
+		ctx := context.Background()
+		r, err := initDataUploaderReconcilerWithError(test.needErrs...)
+		require.NoError(t, err)
+
+		err = r.client.Create(ctx, test.du)
+		require.NoError(t, err)
+
+		succeeded, err := r.acceptDataUpload(ctx, test.du)
+		assert.Equal(t, test.succeeded, succeeded)
+		if test.expectedErr == "" {
+			assert.NoError(t, err)
+		} else {
+			assert.EqualError(t, err, test.expectedErr)
+		}
+	}
+}
+
+func TestOnDuPrepareTimeout(t *testing.T) {
+	tests := []struct {
+		name     string
+		du       *velerov2alpha1api.DataUpload
+		needErrs []error
+		expected *velerov2alpha1api.DataUpload
+	}{
+		{
+			name:     "update fail",
+			du:       dataUploadBuilder().Result(),
+			needErrs: []error{nil, nil, fmt.Errorf("fake-update-error"), nil},
+			expected: dataUploadBuilder().Result(),
+		},
+		{
+			name:     "update interrupted",
+			du:       dataUploadBuilder().Result(),
+			needErrs: []error{nil, nil, &fakeAPIStatus{metav1.StatusReasonConflict}, nil},
+			expected: dataUploadBuilder().Result(),
+		},
+		{
+			name:     "succeed",
+			du:       dataUploadBuilder().Result(),
+			needErrs: []error{nil, nil, nil, nil},
+			expected: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
+		},
+	}
+	for _, test := range tests {
+		ctx := context.Background()
+		r, err := initDataUploaderReconcilerWithError(test.needErrs...)
+		require.NoError(t, err)
+
+		err = r.client.Create(ctx, test.du)
+		require.NoError(t, err)
+
+		r.onPrepareTimeout(ctx, test.du)
+
+		du := velerov2alpha1api.DataUpload{}
+		_ = r.client.Get(ctx, kbclient.ObjectKey{
+			Name:      test.du.Name,
+			Namespace: test.du.Namespace,
+		}, &du)
+
+		assert.Equal(t, test.expected.Status.Phase, du.Status.Phase)
+	}
+}

pkg/controller/pod_volume_backup_controller_test.go

@@ -192,7 +192,7 @@ var _ = Describe("PodVolumeBackup Reconciler", func() {
 			r := PodVolumeBackupReconciler{
 				Client:           fakeClient,
 				clock:            testclocks.NewFakeClock(now),
-				metrics:          metrics.NewPodVolumeMetrics(),
+				metrics:          metrics.NewNodeMetrics(),
 				credentialGetter: &credentials.CredentialGetter{FromFile: credentialFileStore},
 				nodeName:         "test_node",
 				fileSystem:       fakeFS,

pkg/controller/restore_operations_controller.go

@@ -177,14 +177,15 @@ func (r *restoreOperationsReconciler) Reconcile(ctx context.Context, req ctrl.Re
 		operations.ChangesSinceUpdate = true
 	}
 
+	if len(operations.ErrsSinceUpdate) > 0 {
+		restore.Status.Phase = velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed
+	}
+
 	// if stillInProgress is false, restore moves to terminal phase and needs update
 	// if operations.ErrsSinceUpdate is not empty, then restore phase needs to change to
 	// RestorePhaseWaitingForPluginOperationsPartiallyFailed and needs update
 	// If the only changes are incremental progress, then no write is necessary, progress can remain in memory
 	if !stillInProgress {
-		if len(operations.ErrsSinceUpdate) > 0 {
-			restore.Status.Phase = velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed
-		}
 		if restore.Status.Phase == velerov1api.RestorePhaseWaitingForPluginOperations {
 			log.Infof("Marking restore %s completed", restore.Name)
 			restore.Status.Phase = velerov1api.RestorePhaseCompleted