Add fast fail support

Signed-off-by: Ming Qiu <[email protected]>

changelogs/unreleased/7558-qiuming-best

@@ -0,0 +1 @@
+Add fast fail support and fix snapshot leak

config/crd/v1/bases/velero.io_backups.yaml

@@ -481,6 +481,11 @@ spec:
                 description: UploaderConfig specifies the configuration for the uploader.
                 nullable: true
                 properties:
+                  failFast:
+                    description: FailFast fail the entire snapshot on source file
+                      or directory error
+                    nullable: true
+                    type: boolean
                   parallelFilesUpload:
                     description: ParallelFilesUpload is the number of files parallel
                       uploads to perform when using the uploader.

config/crd/v1/bases/velero.io_schedules.yaml

@@ -529,6 +529,11 @@ spec:
                       uploader.
                     nullable: true
                     properties:
+                      failFast:
+                        description: FailFast fail the entire snapshot on source file
+                          or directory error
+                        nullable: true
+                        type: boolean
                       parallelFilesUpload:
                         description: ParallelFilesUpload is the number of files parallel
                           uploads to perform when using the uploader.

pkg/apis/velero/v1/backup_types.go

@@ -187,6 +187,10 @@ type UploaderConfigForBackup struct {
 	// ParallelFilesUpload is the number of files parallel uploads to perform when using the uploader.
 	// +optional
 	ParallelFilesUpload int `json:"parallelFilesUpload,omitempty"`
+	// FailFast fail the entire snapshot on source file or directory error
+	// +optional
+	// +nullable
+	FailFast *bool `json:"failFast,omitempty"`
 }
 
 // BackupHooks contains custom behaviors that should be executed at different phases of the backup.

pkg/builder/backup_builder.go

@@ -129,6 +129,16 @@ func (b *BackupBuilder) FromSchedule(schedule *velerov1api.Schedule) *BackupBuil
 		b.ResourcePolicies(schedule.Spec.Template.ResourcePolicy.Name)
 	}
 
+	if schedule.Spec.Template.UploaderConfig != nil {
+		if schedule.Spec.Template.UploaderConfig.ParallelFilesUpload > 0 {
+			b.ParallelFilesUpload(schedule.Spec.Template.UploaderConfig.ParallelFilesUpload)
+		}
+
+		failFast := schedule.Spec.Template.UploaderConfig.FailFast
+		if failFast != nil && *failFast {
+			b.FailFast(*failFast)
+		}
+	}
 	return b
 }
 
@@ -309,6 +319,17 @@ func (b *BackupBuilder) ParallelFilesUpload(parallel int) *BackupBuilder {
 	return b
 }
 
+// FailFast sets the Backup's uploader FailFast flag
+func (b *BackupBuilder) FailFast(val bool) *BackupBuilder {
+	if b.object.Spec.UploaderConfig == nil {
+		b.object.Spec.UploaderConfig = &velerov1api.UploaderConfigForBackup{}
+	}
+
+	b.object.Spec.UploaderConfig.FailFast = &val
+
+	return b
+}
+
 // WithStatus sets the Backup's status.
 func (b *BackupBuilder) WithStatus(status velerov1api.BackupStatus) *BackupBuilder {
 	b.object.Status = status

pkg/cmd/cli/backup/create.go

@@ -107,14 +107,17 @@ type CreateOptions struct {
 	ResPoliciesConfigmap            string
 	client                          kbclient.WithWatch
 	ParallelFilesUpload             int
+	FailFast                        flag.OptionalBool
 }
 
 func NewCreateOptions() *CreateOptions {
+	failFast := false
 	return &CreateOptions{
 		IncludeNamespaces:       flag.NewStringArray("*"),
 		Labels:                  flag.NewMap(),
 		SnapshotVolumes:         flag.NewOptionalBool(nil),
 		IncludeClusterResources: flag.NewOptionalBool(nil),
+		FailFast:                flag.NewOptionalBool(&failFast),
 	}
 }
 
@@ -153,6 +156,9 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 	flags.StringVar(&o.ResPoliciesConfigmap, "resource-policies-configmap", "", "Reference to the resource policies configmap that backup using")
 	flags.StringVar(&o.DataMover, "data-mover", "", "Specify the data mover to be used by the backup. If the parameter is not set or set as 'velero', the built-in data mover will be used")
 	flags.IntVar(&o.ParallelFilesUpload, "parallel-files-upload", 0, "Number of files uploads simultaneously when running a backup. This is only applicable for the kopia uploader")
+
+	flags.VarPF(&o.FailFast, "fail-fast", "", "Specify whether the backup should fail fast if any of the resources fail to backup. If the parameter is not set, it is treated as setting to 'false'.")
+	f.NoOptDefVal = cmd.TRUE
 }
 
 // BindWait binds the wait flag separately so it is not called by other create
@@ -401,6 +407,9 @@ func (o *CreateOptions) BuildBackup(namespace string) (*velerov1api.Backup, erro
 		if o.ParallelFilesUpload > 0 {
 			backupBuilder.ParallelFilesUpload(o.ParallelFilesUpload)
 		}
+		if o.FailFast.Value != nil && *o.FailFast.Value {
+			backupBuilder.FailFast(*o.FailFast.Value)
+		}
 	}
 
 	backup := backupBuilder.ObjectMeta(builder.WithLabelsMap(o.Labels.Data())).Result()

pkg/cmd/cli/backup/create_test.go

@@ -186,6 +186,7 @@ func TestCreateCommand(t *testing.T) {
 		resPoliciesConfigmap := "cm-name-2"
 		dataMover := "velero"
 		parallelFilesUpload := 10
+		failFast := "true"
 		flags := new(flag.FlagSet)
 		o := NewCreateOptions()
 		o.BindFlags(flags)
@@ -214,6 +215,7 @@ func TestCreateCommand(t *testing.T) {
 		flags.Parse([]string{"--resource-policies-configmap", resPoliciesConfigmap})
 		flags.Parse([]string{"--data-mover", dataMover})
 		flags.Parse([]string{"--parallel-files-upload", fmt.Sprintf("%d", parallelFilesUpload)})
+		flags.Parse([]string{"--fail-fast", failFast})
 		//flags.Parse([]string{"--wait"})
 
 		client := velerotest.NewFakeControllerRuntimeClient(t).(kbclient.WithWatch)
@@ -263,6 +265,7 @@ func TestCreateCommand(t *testing.T) {
 		require.Equal(t, resPoliciesConfigmap, o.ResPoliciesConfigmap)
 		require.Equal(t, dataMover, o.DataMover)
 		require.Equal(t, parallelFilesUpload, o.ParallelFilesUpload)
+		require.Equal(t, failFast, o.FailFast.String())
 		//assert.Equal(t, true, o.Wait)
 
 		// verify oldAndNewFilterParametersUsedTogether

pkg/cmd/cli/nodeagent/server.go

@@ -406,7 +406,7 @@ func (s *nodeAgentServer) markInProgressPVBsFailed(client ctrlclient.Client) {
 			continue
 		}
 
-		if err := controller.UpdatePVBStatusToFailed(s.ctx, client, &pvbs.Items[i],
+		if err := controller.UpdatePVBStatusToFailed(s.ctx, client, &pvbs.Items[i], "",
 			fmt.Sprintf("get a podvolumebackup with status %q during the server starting, mark it as %q", velerov1api.PodVolumeBackupPhaseInProgress, velerov1api.PodVolumeBackupPhaseFailed),
 			time.Now(), s.logger); err != nil {
 			s.logger.WithError(errors.WithStack(err)).Errorf("failed to patch podvolumebackup %q", pvb.GetName())

pkg/cmd/cli/schedule/create.go

@@ -172,9 +172,18 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 	}
 
 	if o.BackupOptions.ParallelFilesUpload > 0 {
-		schedule.Spec.Template.UploaderConfig = &api.UploaderConfigForBackup{
-			ParallelFilesUpload: o.BackupOptions.ParallelFilesUpload,
+		if schedule.Spec.Template.UploaderConfig == nil {
+			schedule.Spec.Template.UploaderConfig = &api.UploaderConfigForBackup{}
 		}
+		schedule.Spec.Template.UploaderConfig.ParallelFilesUpload = o.BackupOptions.ParallelFilesUpload
+	}
+
+	fastFail := o.BackupOptions.FailFast.Value
+	if fastFail != nil && *fastFail {
+		if schedule.Spec.Template.UploaderConfig == nil {
+			schedule.Spec.Template.UploaderConfig = &api.UploaderConfigForBackup{}
+		}
+		schedule.Spec.Template.UploaderConfig.FailFast = fastFail
 	}
 
 	if printed, err := output.PrintWithFormat(c, schedule); printed || err != nil {

pkg/cmd/util/output/backup_describer.go

@@ -89,7 +89,7 @@ func DescribeBackup(
 			DescribeResourcePolicies(d, backup.Spec.ResourcePolicy)
 		}
 
-		if backup.Spec.UploaderConfig != nil && backup.Spec.UploaderConfig.ParallelFilesUpload > 0 {
+		if backup.Spec.UploaderConfig != nil {
 			d.Println()
 			DescribeUploaderConfigForBackup(d, backup.Spec)
 		}
@@ -129,7 +129,12 @@ func DescribeResourcePolicies(d *Describer, resPolicies *v1.TypedLocalObjectRefe
 // DescribeUploaderConfigForBackup describes uploader config in human-readable format
 func DescribeUploaderConfigForBackup(d *Describer, spec velerov1api.BackupSpec) {
 	d.Printf("Uploader config:\n")
-	d.Printf("\tParallel files upload:\t%d\n", spec.UploaderConfig.ParallelFilesUpload)
+	if spec.UploaderConfig.ParallelFilesUpload > 0 {
+		d.Printf("\tParallel files upload:\t%d\n", spec.UploaderConfig.ParallelFilesUpload)
+	}
+	if spec.UploaderConfig.FailFast != nil {
+		d.Printf("\tFail fast:\t%s\n", strconv.FormatBool(*spec.UploaderConfig.FailFast))
+	}
 }
 
 // DescribeBackupSpec describes a backup spec in human-readable format.

pkg/cmd/util/output/schedule_describer.go

@@ -48,7 +48,7 @@ func DescribeSchedule(schedule *v1.Schedule) string {
 			DescribeResourcePolicies(d, schedule.Spec.Template.ResourcePolicy)
 		}
 
-		if schedule.Spec.Template.UploaderConfig != nil && schedule.Spec.Template.UploaderConfig.ParallelFilesUpload > 0 {
+		if schedule.Spec.Template.UploaderConfig != nil {
 			d.Println()
 			DescribeUploaderConfigForBackup(d, schedule.Spec.Template)
 		}

pkg/controller/data_download_controller.go

@@ -373,7 +373,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na
 	}
 }
 
-func (r *DataDownloadReconciler) OnDataDownloadFailed(ctx context.Context, namespace string, ddName string, err error) {
+func (r *DataDownloadReconciler) OnDataDownloadFailed(ctx context.Context, namespace, ddName, snapshotID string, err error) {
 	defer r.closeDataPath(ctx, ddName)
 
 	log := r.logger.WithField("datadownload", ddName)

pkg/controller/data_download_controller_test.go

@@ -461,7 +461,7 @@ func TestOnDataDownloadFailed(t *testing.T) {
 		ddName := dd.Name
 		// Add the DataDownload object to the fake client
 		assert.NoError(t, r.client.Create(ctx, dd))
-		r.OnDataDownloadFailed(ctx, namespace, ddName, fmt.Errorf("Failed to handle %v", ddName))
+		r.OnDataDownloadFailed(ctx, namespace, ddName, "", fmt.Errorf("Failed to handle %v", ddName))
 		updatedDD := &velerov2alpha1api.DataDownload{}
 		if getErr {
 			assert.Error(t, r.client.Get(ctx, types.NamespacedName{Name: ddName, Namespace: namespace}, updatedDD))

pkg/controller/data_upload_controller.go

@@ -395,7 +395,7 @@ func (r *DataUploadReconciler) OnDataUploadCompleted(ctx context.Context, namesp
 	}
 }
 
-func (r *DataUploadReconciler) OnDataUploadFailed(ctx context.Context, namespace string, duName string, err error) {
+func (r *DataUploadReconciler) OnDataUploadFailed(ctx context.Context, namespace, duName, snapshotID string, err error) {
 	defer r.closeDataPath(ctx, duName)
 
 	log := r.logger.WithField("dataupload", duName)
@@ -406,7 +406,7 @@ func (r *DataUploadReconciler) OnDataUploadFailed(ctx context.Context, namespace
 	if getErr := r.client.Get(ctx, types.NamespacedName{Name: duName, Namespace: namespace}, &du); getErr != nil {
 		log.WithError(getErr).Warn("Failed to get dataupload on failure")
 	} else {
-		if _, errOut := r.errorOut(ctx, &du, err, "data path backup failed", log); err != nil {
+		if errOut := r.errorOutWithSnapshotID(ctx, &du, snapshotID, "data path backup failed", err, log); errOut != nil {
 			log.WithError(err).Warnf("Failed to patch dataupload with err %v", errOut)
 		}
 	}
@@ -669,28 +669,37 @@ func (r *DataUploadReconciler) prepareDataUpload(du *velerov2alpha1api.DataUploa
 	du.Status.Node = r.nodeName
 }
 
-func (r *DataUploadReconciler) errorOut(ctx context.Context, du *velerov2alpha1api.DataUpload, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
+func (r *DataUploadReconciler) errorOutWithSnapshotID(ctx context.Context, du *velerov2alpha1api.DataUpload, snapshotID, msg string, err error, log logrus.FieldLogger) error {
+	err = r.cleanUpExposedSnapshot(ctx, du, err)
+	return r.updateStatusToFailed(ctx, du, err, snapshotID, msg, log)
+}
+
+func (r *DataUploadReconciler) cleanUpExposedSnapshot(ctx context.Context, du *velerov2alpha1api.DataUpload, err error) error {
 	if se, ok := r.snapshotExposerList[du.Spec.SnapshotType]; ok {
 		var volumeSnapshotName string
 		if du.Spec.SnapshotType == velerov2alpha1api.SnapshotTypeCSI { // Other exposer should have another condition
 			volumeSnapshotName = du.Spec.CSISnapshot.VolumeSnapshot
 		}
 		se.CleanUp(ctx, getOwnerObject(du), volumeSnapshotName, du.Spec.SourceNamespace)
 	} else {
-		err = errors.Wrapf(err, "failed to clean up exposed snapshot with could not find %s snapshot exposer", du.Spec.SnapshotType)
+		return errors.Wrapf(err, "failed to clean up exposed snapshot with could not find %s snapshot exposer", du.Spec.SnapshotType)
 	}
+	return err
+}
 
-	return ctrl.Result{}, r.updateStatusToFailed(ctx, du, err, msg, log)
+func (r *DataUploadReconciler) errorOut(ctx context.Context, du *velerov2alpha1api.DataUpload, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
+	err = r.cleanUpExposedSnapshot(ctx, du, err)
+	return ctrl.Result{}, r.updateStatusToFailed(ctx, du, err, "", msg, log)
 }
 
-func (r *DataUploadReconciler) updateStatusToFailed(ctx context.Context, du *velerov2alpha1api.DataUpload, err error, msg string, log logrus.FieldLogger) error {
+func (r *DataUploadReconciler) updateStatusToFailed(ctx context.Context, du *velerov2alpha1api.DataUpload, err error, snapshtoID, msg string, log logrus.FieldLogger) error {
 	original := du.DeepCopy()
 	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.SnapshotID = snapshtoID
 	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")

pkg/controller/data_upload_controller_test.go

@@ -652,7 +652,7 @@ func TestOnDataUploadFailed(t *testing.T) {
 	// Add the DataUpload object to the fake client
 	assert.NoError(t, r.client.Create(ctx, du))
 	r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{velerov2alpha1api.SnapshotTypeCSI: exposer.NewCSISnapshotExposer(r.kubeClient, r.csiSnapshotClient, velerotest.NewLogger())}
-	r.OnDataUploadFailed(ctx, namespace, duName, fmt.Errorf("Failed to handle %v", duName))
+	r.OnDataUploadFailed(ctx, namespace, duName, "", fmt.Errorf("Failed to handle %v", duName))
 	updatedDu := &velerov2alpha1api.DataUpload{}
 	assert.NoError(t, r.client.Get(ctx, types.NamespacedName{Name: duName, Namespace: namespace}, updatedDu))
 	assert.Equal(t, velerov2alpha1api.DataUploadPhaseFailed, updatedDu.Status.Phase)

pkg/controller/pod_volume_backup_controller.go

@@ -124,6 +124,7 @@ func (r *PodVolumeBackupReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	}
 
 	fsBackup, err := r.dataPathMgr.CreateFileSystemBR(pvb.Name, pVBRRequestor, ctx, r.Client, pvb.Namespace, callbacks, log)
+
 	if err != nil {
 		if err == datapath.ConcurrentLimitExceed {
 			return ctrl.Result{Requeue: true, RequeueAfter: time.Second * 5}, nil
@@ -225,7 +226,7 @@ func (r *PodVolumeBackupReconciler) OnDataPathCompleted(ctx context.Context, nam
 	log.Info("PodVolumeBackup completed")
 }
 
-func (r *PodVolumeBackupReconciler) OnDataPathFailed(ctx context.Context, namespace string, pvbName string, err error) {
+func (r *PodVolumeBackupReconciler) OnDataPathFailed(ctx context.Context, namespace, pvbName, snapshotID string, err error) {
 	defer r.closeDataPath(ctx, pvbName)
 
 	log := r.logger.WithField("pvb", pvbName)
@@ -236,7 +237,9 @@ func (r *PodVolumeBackupReconciler) OnDataPathFailed(ctx context.Context, namesp
 	if getErr := r.Client.Get(ctx, types.NamespacedName{Name: pvbName, Namespace: namespace}, &pvb); getErr != nil {
 		log.WithError(getErr).Warn("Failed to get PVB on failure")
 	} else {
-		_, _ = r.errorOut(ctx, &pvb, err, "data path backup failed", log)
+		if getErr := r.errorOutWithSnapshotID(ctx, &pvb, snapshotID, "data path backup failed", err, log); getErr != nil {
+			log.WithError(getErr).Warn("Failed to update PVB status")
+		}
 	}
 }
 
@@ -346,19 +349,26 @@ func (r *PodVolumeBackupReconciler) closeDataPath(ctx context.Context, pvbName s
 	r.dataPathMgr.RemoveAsyncBR(pvbName)
 }
 
+func (r *PodVolumeBackupReconciler) errorOutWithSnapshotID(ctx context.Context, pvb *velerov1api.PodVolumeBackup, snapshotID, msg string, err error, log logrus.FieldLogger) error {
+	r.closeDataPath(ctx, pvb.Name)
+	err = errors.WithMessage(err, msg)
+	return UpdatePVBStatusToFailed(ctx, r.Client, pvb, snapshotID, err.Error(), r.clock.Now(), log)
+}
+
 func (r *PodVolumeBackupReconciler) errorOut(ctx context.Context, pvb *velerov1api.PodVolumeBackup, err error, msg string, log logrus.FieldLogger) (ctrl.Result, error) {
 	r.closeDataPath(ctx, pvb.Name)
-	_ = UpdatePVBStatusToFailed(ctx, r.Client, pvb, errors.WithMessage(err, msg).Error(), r.clock.Now(), log)
+	err = errors.WithMessage(err, msg)
+	_ = UpdatePVBStatusToFailed(ctx, r.Client, pvb, "", err.Error(), r.clock.Now(), log)
 
 	return ctrl.Result{}, err
 }
 
-func UpdatePVBStatusToFailed(ctx context.Context, c client.Client, pvb *velerov1api.PodVolumeBackup, errString string, time time.Time, log logrus.FieldLogger) error {
+func UpdatePVBStatusToFailed(ctx context.Context, c client.Client, pvb *velerov1api.PodVolumeBackup, snapshotID, errString string, time time.Time, log logrus.FieldLogger) error {
 	original := pvb.DeepCopy()
 	pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseFailed
 	pvb.Status.Message = errString
 	pvb.Status.CompletionTimestamp = &metav1.Time{Time: time}
-
+	pvb.Status.SnapshotID = snapshotID
 	err := c.Patch(ctx, pvb, client.MergeFrom(original))
 	if err != nil {
 		log.WithError(err).Error("error updating PodVolumeBackup status")

pkg/controller/pod_volume_restore_controller.go

@@ -317,7 +317,7 @@ func (c *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, na
 	log.Info("Restore completed")
 }
 
-func (c *PodVolumeRestoreReconciler) OnDataPathFailed(ctx context.Context, namespace string, pvrName string, err error) {
+func (c *PodVolumeRestoreReconciler) OnDataPathFailed(ctx context.Context, namespace, pvrName, snapshotID string, err error) {
 	defer c.closeDataPath(ctx, pvrName)
 
 	log := c.logger.WithField("pvr", pvrName)

pkg/datapath/file_system.go

@@ -141,7 +141,7 @@ func (fs *fileSystemBR) StartBackup(source AccessPoint, realSource string, paren
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)
 		} else if err != nil {
-			fs.callbacks.OnFailed(context.Background(), fs.namespace, fs.jobName, err)
+			fs.callbacks.OnFailed(context.Background(), fs.namespace, fs.jobName, snapshotID, err)
 		} else {
 			fs.callbacks.OnCompleted(context.Background(), fs.namespace, fs.jobName, Result{Backup: BackupResult{snapshotID, emptySnapshot, source}})
 		}
@@ -161,7 +161,7 @@ func (fs *fileSystemBR) StartRestore(snapshotID string, target AccessPoint, uplo
 		if err == provider.ErrorCanceled {
 			fs.callbacks.OnCancelled(context.Background(), fs.namespace, fs.jobName)
 		} else if err != nil {
-			fs.callbacks.OnFailed(context.Background(), fs.namespace, fs.jobName, err)
+			fs.callbacks.OnFailed(context.Background(), fs.namespace, fs.jobName, snapshotID, err)
 		} else {
 			fs.callbacks.OnCompleted(context.Background(), fs.namespace, fs.jobName, Result{Restore: RestoreResult{Target: target}})
 		}