BIAv2 async operations controller work

Signed-off-by: Scott Seago <sseago@redhat.com>

pkg/controller/async_backup_operations_controller.go

@@ -0,0 +1,483 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"bytes"
+	"context"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clocks "k8s.io/utils/clock"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
+	"github.com/vmware-tanzu/velero/pkg/persistence"
+	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
+	"github.com/vmware-tanzu/velero/pkg/util/encode"
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
+)
+
+const (
+	defaultAsyncBackupOperationsFrequency = 2 * time.Minute
+)
+
+type operationsForBackup struct {
+	operations         []*itemoperation.BackupOperation
+	changesSinceUpdate bool
+	errsSinceUpdate    []string
+}
+
+// FIXME: remove if handled by backup finalizer controller
+func (o *operationsForBackup) anyItemsToUpdate() bool {
+	for _, op := range o.operations {
+		if len(op.Spec.ItemsToUpdate) > 0 {
+			return true
+		}
+	}
+	return false
+}
+func (in *operationsForBackup) DeepCopy() *operationsForBackup {
+	if in == nil {
+		return nil
+	}
+	out := new(operationsForBackup)
+	in.DeepCopyInto(out)
+	return out
+}
+
+func (in *operationsForBackup) DeepCopyInto(out *operationsForBackup) {
+	*out = *in
+	if in.operations != nil {
+		in, out := &in.operations, &out.operations
+		*out = make([]*itemoperation.BackupOperation, len(*in))
+		for i := range *in {
+			if (*in)[i] != nil {
+				in, out := &(*in)[i], &(*out)[i]
+				*out = new(itemoperation.BackupOperation)
+				(*in).DeepCopyInto(*out)
+			}
+		}
+	}
+	if in.errsSinceUpdate != nil {
+		in, out := &in.errsSinceUpdate, &out.errsSinceUpdate
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+}
+
+func (o *operationsForBackup) uploadProgress(backupStore persistence.BackupStore, backupName string) error {
+	if len(o.operations) > 0 {
+		var backupItemOperations *bytes.Buffer
+		backupItemOperations, errs := encodeToJSONGzip(o.operations, "backup item operations list")
+		if errs != nil {
+			return errors.Wrap(errs[0], "error encoding item operations json")
+		}
+		err := backupStore.PutBackupItemOperations(backupName, backupItemOperations)
+		if err != nil {
+			return errors.Wrap(err, "error uploading item operations json")
+		}
+	}
+	o.changesSinceUpdate = false
+	o.errsSinceUpdate = nil
+	return nil
+}
+
+type BackupItemOperationsMap struct {
+	operations map[string]*operationsForBackup
+	opsLock    sync.Mutex
+}
+
+// If backup has changes not yet uploaded, upload them now
+func (m *BackupItemOperationsMap) UpdateForBackup(backupStore persistence.BackupStore, backupName string) error {
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+
+	operations, ok := m.operations[backupName]
+	// if operations for this backup aren't found, or if there are no changes
+	// or errors since last update, do nothing
+	if !ok || (!operations.changesSinceUpdate && len(operations.errsSinceUpdate) == 0) {
+		return nil
+	}
+	if err := operations.uploadProgress(backupStore, backupName); err != nil {
+		return err
+	}
+	return nil
+}
+
+type asyncBackupOperationsReconciler struct {
+	client.Client
+	logger            logrus.FieldLogger
+	clock             clocks.WithTickerAndDelayedExecution
+	frequency         time.Duration
+	itemOperationsMap *BackupItemOperationsMap
+	newPluginManager  func(logger logrus.FieldLogger) clientmgmt.Manager
+	backupStoreGetter persistence.ObjectBackupStoreGetter
+	metrics           *metrics.ServerMetrics
+}
+
+func NewAsyncBackupOperationsReconciler(
+	logger logrus.FieldLogger,
+	client client.Client,
+	frequency time.Duration,
+	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
+	backupStoreGetter persistence.ObjectBackupStoreGetter,
+	metrics *metrics.ServerMetrics,
+) (*asyncBackupOperationsReconciler, *BackupItemOperationsMap) {
+	abor := &asyncBackupOperationsReconciler{
+		Client:            client,
+		logger:            logger,
+		clock:             clocks.RealClock{},
+		frequency:         frequency,
+		itemOperationsMap: &BackupItemOperationsMap{operations: make(map[string]*operationsForBackup)},
+		newPluginManager:  newPluginManager,
+		backupStoreGetter: backupStoreGetter,
+		metrics:           metrics,
+	}
+	if abor.frequency <= 0 {
+		abor.frequency = defaultAsyncBackupOperationsFrequency
+	}
+	return abor, abor.itemOperationsMap
+}
+
+func (c *asyncBackupOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	s := kube.NewPeriodicalEnqueueSource(c.logger, mgr.GetClient(), &velerov1api.BackupList{}, c.frequency, kube.PeriodicalEnqueueSourceOption{})
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&velerov1api.Backup{}, builder.WithPredicates(kube.FalsePredicate{})).
+		Watches(s, nil).
+		Complete(c)
+}
+
+// +kubebuilder:rbac:groups=velero.io,resources=backups,verbs=get;list;watch;update
+// +kubebuilder:rbac:groups=velero.io,resources=backups/status,verbs=get
+// +kubebuilder:rbac:groups=velero.io,resources=backupstoragelocations,verbs=get
+func (c *asyncBackupOperationsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := c.logger.WithField("async backup operations for backup", req.String())
+	// FIXME: make this log.Debug
+	log.Info("asyncBackupOperationsReconciler getting backup")
+
+	original := &velerov1api.Backup{}
+	if err := c.Get(ctx, req.NamespacedName, original); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.WithError(err).Error("backup not found")
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{}, errors.Wrapf(err, "error getting backup %s", req.String())
+	}
+	backup := original.DeepCopy()
+	log.Debugf("backup: %s", backup.Name)
+
+	log = c.logger.WithFields(
+		logrus.Fields{
+			"backup": req.String(),
+		},
+	)
+
+	switch backup.Status.Phase {
+	case velerov1api.BackupPhaseWaitingForPluginOperations, velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed:
+		// only process backups waiting for plugin operations to complete
+	default:
+		log.Debug("Backup has no ongoing async plugin operations, skipping")
+		return ctrl.Result{}, nil
+	}
+
+	loc := &velerov1api.BackupStorageLocation{}
+	if err := c.Get(ctx, client.ObjectKey{
+		Namespace: req.Namespace,
+		Name:      backup.Spec.StorageLocation,
+	}, loc); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Warnf("Cannot check progress on async Backup operations because backup storage location %s does not exist; marking backup PartiallyFailed", backup.Spec.StorageLocation)
+			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+		} else {
+			log.Warnf("Cannot check progress on async Backup operations because backup storage location %s could not be retrieved: %s; marking backup PartiallyFailed", backup.Spec.StorageLocation, err.Error())
+			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+		}
+		err2 := c.updateBackupAndOperationsJSON(ctx, original, backup, nil, &operationsForBackup{errsSinceUpdate: []string{err.Error()}}, false, false)
+		if err2 != nil {
+			log.WithError(err2).Error("error updating Backup")
+		}
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup storage location")
+	}
+
+	if loc.Spec.AccessMode == velerov1api.BackupStorageLocationAccessModeReadOnly {
+		log.Infof("Cannot check progress on async Backup operations because backup storage location %s is currently in read-only mode; marking backup PartiallyFailed", loc.Name)
+		backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+
+		err := c.updateBackupAndOperationsJSON(ctx, original, backup, nil, &operationsForBackup{errsSinceUpdate: []string{"BSL is read-only"}}, false, false)
+		if err != nil {
+			log.WithError(err).Error("error updating Backup")
+		}
+		return ctrl.Result{}, nil
+	}
+
+	pluginManager := c.newPluginManager(c.logger)
+	defer pluginManager.CleanupClients()
+	backupStore, err := c.backupStoreGetter.Get(loc, pluginManager, c.logger)
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup store")
+	}
+
+	operations, err := c.getOperationsForBackup(backupStore, backup.Name)
+	if err != nil {
+		err2 := c.updateBackupAndOperationsJSON(ctx, original, backup, backupStore, &operationsForBackup{errsSinceUpdate: []string{err.Error()}}, false, false)
+		if err2 != nil {
+			return ctrl.Result{}, errors.Wrap(err2, "error updating Backup")
+		}
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup operations")
+	}
+	stillInProgress, changes, opsCompleted, opsFailed, errs := getBackupItemOperationProgress(backup, pluginManager, operations.operations)
+	// if len(errs)>0, need to update backup errors and error log
+	operations.errsSinceUpdate = append(operations.errsSinceUpdate, errs...)
+	backup.Status.Errors += len(operations.errsSinceUpdate)
+	asyncCompletionChanges := false
+	if backup.Status.AsyncBackupItemOperationsCompleted != opsCompleted || backup.Status.AsyncBackupItemOperationsFailed != opsFailed {
+		asyncCompletionChanges = true
+		backup.Status.AsyncBackupItemOperationsCompleted = opsCompleted
+		backup.Status.AsyncBackupItemOperationsFailed = opsFailed
+	}
+	if changes {
+		operations.changesSinceUpdate = true
+	}
+
+	// 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 FinalizingAfterPluginOperations", backup.Name)
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperations
+		} else {
+			log.Infof("Marking backup %s FinalizingAfterPluginOperationsPartiallyFailed", backup.Name)
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed
+		}
+	}
+	err = c.updateBackupAndOperationsJSON(ctx, original, backup, backupStore, operations, asyncCompletionChanges, changes)
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error updating Backup")
+	}
+	return ctrl.Result{}, nil
+}
+
+func (c *asyncBackupOperationsReconciler) updateBackupAndOperationsJSON(
+	ctx context.Context,
+	original, backup *velerov1api.Backup,
+	backupStore persistence.BackupStore,
+	operations *operationsForBackup,
+	changes bool,
+	asyncCompletionChanges bool) error {
+
+	backupScheduleName := backup.GetLabels()[velerov1api.ScheduleNameLabel]
+
+	if len(operations.errsSinceUpdate) > 0 {
+		c.metrics.RegisterBackupItemsErrorsGauge(backupScheduleName, backup.Status.Errors)
+		// FIXME: download/upload results once https://github.com/vmware-tanzu/velero/pull/5576 is merged
+	}
+	removeIfComplete := true
+	defer func() {
+		// remove local operations list if complete
+		c.itemOperationsMap.opsLock.Lock()
+		if removeIfComplete && (backup.Status.Phase == velerov1api.BackupPhaseCompleted ||
+			backup.Status.Phase == velerov1api.BackupPhasePartiallyFailed ||
+			backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperations ||
+			backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed) {
+
+			c.deleteOperationsForBackup(backup.Name)
+		} else if changes {
+			c.putOperationsForBackup(operations, backup.Name)
+		}
+		c.itemOperationsMap.opsLock.Unlock()
+	}()
+
+	// update backup and upload progress if errs or complete
+	if len(operations.errsSinceUpdate) > 0 ||
+		backup.Status.Phase == velerov1api.BackupPhaseCompleted ||
+		backup.Status.Phase == velerov1api.BackupPhasePartiallyFailed ||
+		backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperations ||
+		backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed {
+		// update file store
+		if backupStore != nil {
+			backupJSON := new(bytes.Buffer)
+			if err := encode.EncodeTo(backup, "json", backupJSON); err != nil {
+				removeIfComplete = false
+				return errors.Wrap(err, "error encoding backup json")
+			}
+			err := backupStore.PutBackupMetadata(backup.Name, backupJSON)
+			if err != nil {
+				removeIfComplete = false
+				return errors.Wrap(err, "error uploading backup json")
+			}
+			if err := operations.uploadProgress(backupStore, backup.Name); err != nil {
+				removeIfComplete = false
+				return err
+			}
+		}
+		// update backup
+		err := c.Client.Patch(ctx, backup, client.MergeFrom(original))
+		if err != nil {
+			removeIfComplete = false
+			return errors.Wrapf(err, "error updating Backup %s", backup.Name)
+		}
+	} else if asyncCompletionChanges {
+		// If backup is still incomplete and no new errors are found but there are some new operations
+		// completed, patch backup to reflect new completion numbers, but don't upload detailed json file
+		err := c.Client.Patch(ctx, backup, client.MergeFrom(original))
+		if err != nil {
+			return errors.Wrapf(err, "error updating Backup %s", backup.Name)
+		}
+	}
+	return nil
+}
+
+// returns a deep copy so we can minimize the time the map is locked
+func (c *asyncBackupOperationsReconciler) getOperationsForBackup(
+	backupStore persistence.BackupStore,
+	backupName string) (*operationsForBackup, error) {
+	var err error
+	// lock operations map
+	c.itemOperationsMap.opsLock.Lock()
+	defer c.itemOperationsMap.opsLock.Unlock()
+
+	operations, ok := c.itemOperationsMap.operations[backupName]
+	if !ok || len(operations.operations) == 0 {
+		operations = &operationsForBackup{}
+		operations.operations, err = backupStore.GetBackupItemOperations(backupName)
+		if err == nil {
+			c.itemOperationsMap.operations[backupName] = operations
+		}
+	}
+	return operations.DeepCopy(), err
+}
+
+func (c *asyncBackupOperationsReconciler) putOperationsForBackup(
+	operations *operationsForBackup,
+	backupName string) {
+	if operations != nil {
+		c.itemOperationsMap.operations[backupName] = operations
+	}
+}
+
+func (c *asyncBackupOperationsReconciler) deleteOperationsForBackup(backupName string) {
+	if _, ok := c.itemOperationsMap.operations[backupName]; ok {
+		delete(c.itemOperationsMap.operations, backupName)
+	}
+	return
+}
+
+func getBackupItemOperationProgress(
+	backup *velerov1api.Backup,
+	pluginManager clientmgmt.Manager,
+	operationsList []*itemoperation.BackupOperation) (bool, bool, int, int, []string) {
+	inProgressOperations := false
+	changes := false
+	var errs []string
+	var completedCount, failedCount int
+
+	for _, operation := range operationsList {
+		if operation.Status.Phase == itemoperation.OperationPhaseInProgress {
+			bia, err := pluginManager.GetBackupItemActionV2(operation.Spec.BackupItemAction)
+			if err != nil {
+				operation.Status.Phase = itemoperation.OperationPhaseFailed
+				operation.Status.Error = err.Error()
+				errs = append(errs, err.Error())
+				changes = true
+				failedCount++
+				continue
+			}
+			operationProgress, err := bia.Progress(operation.Spec.OperationID, backup)
+			if err != nil {
+				operation.Status.Phase = itemoperation.OperationPhaseFailed
+				operation.Status.Error = err.Error()
+				errs = append(errs, err.Error())
+				changes = true
+				failedCount++
+				continue
+			}
+			if operation.Status.NCompleted != operationProgress.NCompleted {
+				operation.Status.NCompleted = operationProgress.NCompleted
+				changes = true
+			}
+			if operation.Status.NTotal != operationProgress.NTotal {
+				operation.Status.NTotal = operationProgress.NTotal
+				changes = true
+			}
+			if operation.Status.OperationUnits != operationProgress.OperationUnits {
+				operation.Status.OperationUnits = operationProgress.OperationUnits
+				changes = true
+			}
+			if operation.Status.Description != operationProgress.Description {
+				operation.Status.Description = operationProgress.Description
+				changes = true
+			}
+			started := metav1.NewTime(operationProgress.Started)
+			if operation.Status.Started == nil || *(operation.Status.Started) != started {
+				operation.Status.Started = &started
+				changes = true
+			}
+			updated := metav1.NewTime(operationProgress.Updated)
+			if operation.Status.Updated == nil || *(operation.Status.Updated) != updated {
+				operation.Status.Updated = &updated
+				changes = true
+			}
+
+			if operationProgress.Completed {
+				if operationProgress.Err != "" {
+					operation.Status.Phase = itemoperation.OperationPhaseFailed
+					operation.Status.Error = operationProgress.Err
+					errs = append(errs, operationProgress.Err)
+					changes = true
+					failedCount++
+					continue
+				}
+				operation.Status.Phase = itemoperation.OperationPhaseCompleted
+				changes = true
+				completedCount++
+				continue
+			}
+			// cancel operation if past timeout period
+			if operation.Status.Created.Time.Add(backup.Spec.ItemOperationTimeout.Duration).Before(time.Now()) {
+				_ = bia.Cancel(operation.Spec.OperationID, backup)
+				operation.Status.Phase = itemoperation.OperationPhaseFailed
+				operation.Status.Error = "Asynchronous action timed out"
+				errs = append(errs, operation.Status.Error)
+				changes = true
+				failedCount++
+				continue
+			}
+			// if we reach this point, the operation is still running
+			inProgressOperations = true
+		} else if operation.Status.Phase == itemoperation.OperationPhaseCompleted {
+			completedCount++
+		} else if operation.Status.Phase == itemoperation.OperationPhaseFailed {
+			failedCount++
+		}
+	}
+	return inProgressOperations, changes, completedCount, failedCount, errs
+}

pkg/controller/async_backup_operations_controller_test.go

@@ -0,0 +1,308 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	testclocks "k8s.io/utils/clock/testing"
+	ctrl "sigs.k8s.io/controller-runtime"
+	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
+	persistencemocks "github.com/vmware-tanzu/velero/pkg/persistence/mocks"
+	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
+	pluginmocks "github.com/vmware-tanzu/velero/pkg/plugin/mocks"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	biav2mocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks/backupitemaction/v2"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+var (
+	pluginManager = &pluginmocks.Manager{}
+	backupStore   = &persistencemocks.BackupStore{}
+	bia           = &biav2mocks.BackupItemAction{}
+)
+
+func mockAsyncBackupOperationsReconciler(fakeClient kbclient.Client, fakeClock *testclocks.FakeClock, freq time.Duration) (*asyncBackupOperationsReconciler, *BackupItemOperationsMap) {
+	abor, biaMap := NewAsyncBackupOperationsReconciler(
+		logrus.StandardLogger(),
+		fakeClient,
+		freq,
+		func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
+		NewFakeSingleObjectBackupStoreGetter(backupStore),
+		metrics.NewServerMetrics(),
+	)
+	abor.clock = fakeClock
+	return abor, biaMap
+}
+
+func TestAsyncBackupOperationsReconcile(t *testing.T) {
+	fakeClock := testclocks.NewFakeClock(time.Now())
+	metav1Now := metav1.NewTime(fakeClock.Now())
+
+	defaultBackupLocation := builder.ForBackupStorageLocation(velerov1api.DefaultNamespace, "default").Result()
+
+	tests := []struct {
+		name              string
+		backup            *velerov1api.Backup
+		backupOperations  []*itemoperation.BackupOperation
+		backupLocation    *velerov1api.BackupStorageLocation
+		operationComplete bool
+		operationErr      string
+		expectError       bool
+		expectPhase       velerov1api.BackupPhase
+	}{
+		{
+			name: "WaitingForPluginOperations backup with completed operations is FinalizingAfterPluginOperations",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
+				StorageLocation("default").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo")).
+				Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-1",
+						BackupUID:        "foo",
+						BackupItemAction: "foo",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-1",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperations backup with incomplete operations is still incomplete",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-2").
+				StorageLocation("default").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo")).
+				Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: false,
+			expectPhase:       velerov1api.BackupPhaseWaitingForPluginOperations,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-2",
+						BackupUID:        "foo-2",
+						BackupItemAction: "foo-2",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-2",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperations backup with completed failed operations is FinalizingAfterPluginOperationsPartiallyFailed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-3").
+				StorageLocation("default").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo")).
+				Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			operationErr:      "failed",
+			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-3",
+						BackupUID:        "foo-3",
+						BackupItemAction: "foo-3",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-3",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperationsPartiallyFailed backup with completed operations is FinalizingAfterPluginOperationsPartiallyFailed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
+				StorageLocation("default").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo")).
+				Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-4",
+						BackupUID:        "foo-4",
+						BackupItemAction: "foo-4",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-4",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperationsPartiallyFailed backup with incomplete operations is still incomplete",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-2").
+				StorageLocation("default").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo")).
+				Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: false,
+			expectPhase:       velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-5",
+						BackupUID:        "foo-5",
+						BackupItemAction: "foo-5",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-5",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperationsPartiallyFailed backup with completed failed operations is FinalizingAfterPluginOperationsPartiallyFailed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-3").
+				StorageLocation("default").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo")).
+				Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			operationErr:      "failed",
+			expectPhase:       velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-6",
+						BackupUID:        "foo-6",
+						BackupItemAction: "foo-6",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-6",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if test.backup == nil {
+				return
+			}
+
+			initObjs := []runtime.Object{}
+			initObjs = append(initObjs, test.backup)
+
+			if test.backupLocation != nil {
+				initObjs = append(initObjs, test.backupLocation)
+			}
+
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, initObjs...)
+			reconciler, _ := mockAsyncBackupOperationsReconciler(fakeClient, fakeClock, defaultAsyncBackupOperationsFrequency)
+			pluginManager.On("CleanupClients").Return(nil)
+			backupStore.On("GetBackupItemOperations", test.backup.Name).Return(test.backupOperations, nil)
+			backupStore.On("PutBackupItemOperations", mock.Anything, mock.Anything).Return(nil)
+			backupStore.On("PutBackupMetadata", mock.Anything, mock.Anything).Return(nil)
+			for _, operation := range test.backupOperations {
+				bia.On("Progress", operation.Spec.OperationID, mock.Anything).
+					Return(velero.OperationProgress{
+						Completed: test.operationComplete,
+						Err:       test.operationErr,
+					}, nil)
+				pluginManager.On("GetBackupItemActionV2", operation.Spec.BackupItemAction).Return(bia, nil)
+			}
+			_, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: test.backup.Namespace, Name: test.backup.Name}})
+			gotErr := err != nil
+			assert.Equal(t, test.expectError, gotErr)
+
+			backupAfter := velerov1api.Backup{}
+			err = fakeClient.Get(context.TODO(), types.NamespacedName{
+				Namespace: test.backup.Namespace,
+				Name:      test.backup.Name,
+			}, &backupAfter)
+
+			require.NoError(t, err)
+			assert.Equal(t, test.expectPhase, backupAfter.Status.Phase)
+		})
+	}
+}

pkg/controller/backup_controller.go

@@ -74,27 +74,28 @@ import (
 
 type backupController struct {
 	*genericController
-	discoveryHelper           discovery.Helper
-	backupper                 pkgbackup.Backupper
-	lister                    velerov1listers.BackupLister
-	client                    velerov1client.BackupsGetter
-	kbClient                  kbclient.Client
-	clock                     clocks.WithTickerAndDelayedExecution
-	backupLogLevel            logrus.Level
-	newPluginManager          func(logrus.FieldLogger) clientmgmt.Manager
-	backupTracker             BackupTracker
-	defaultBackupLocation     string
-	defaultVolumesToFsBackup  bool
-	defaultBackupTTL          time.Duration
-	defaultCSISnapshotTimeout time.Duration
-	snapshotLocationLister    velerov1listers.VolumeSnapshotLocationLister
-	defaultSnapshotLocations  map[string]string
-	metrics                   *metrics.ServerMetrics
-	backupStoreGetter         persistence.ObjectBackupStoreGetter
-	formatFlag                logging.Format
-	volumeSnapshotLister      snapshotv1listers.VolumeSnapshotLister
-	volumeSnapshotClient      snapshotterClientSet.Interface
-	credentialFileStore       credentials.FileStore
+	discoveryHelper             discovery.Helper
+	backupper                   pkgbackup.Backupper
+	lister                      velerov1listers.BackupLister
+	client                      velerov1client.BackupsGetter
+	kbClient                    kbclient.Client
+	clock                       clocks.WithTickerAndDelayedExecution
+	backupLogLevel              logrus.Level
+	newPluginManager            func(logrus.FieldLogger) clientmgmt.Manager
+	backupTracker               BackupTracker
+	defaultBackupLocation       string
+	defaultVolumesToFsBackup    bool
+	defaultBackupTTL            time.Duration
+	defaultCSISnapshotTimeout   time.Duration
+	defaultItemOperationTimeout time.Duration
+	snapshotLocationLister      velerov1listers.VolumeSnapshotLocationLister
+	defaultSnapshotLocations    map[string]string
+	metrics                     *metrics.ServerMetrics
+	backupStoreGetter           persistence.ObjectBackupStoreGetter
+	formatFlag                  logging.Format
+	volumeSnapshotLister        snapshotv1listers.VolumeSnapshotLister
+	volumeSnapshotClient        snapshotterClientSet.Interface
+	credentialFileStore         credentials.FileStore
 }
 
 func NewBackupController(
@@ -111,6 +112,7 @@ func NewBackupController(
 	defaultVolumesToFsBackup bool,
 	defaultBackupTTL time.Duration,
 	defaultCSISnapshotTimeout time.Duration,
+	defaultItemOperationTimeout time.Duration,
 	volumeSnapshotLocationLister velerov1listers.VolumeSnapshotLocationLister,
 	defaultSnapshotLocations map[string]string,
 	metrics *metrics.ServerMetrics,
@@ -121,28 +123,29 @@ func NewBackupController(
 	credentialStore credentials.FileStore,
 ) Interface {
 	c := &backupController{
-		genericController:         newGenericController(Backup, logger),
-		discoveryHelper:           discoveryHelper,
-		backupper:                 backupper,
-		lister:                    backupInformer.Lister(),
-		client:                    client,
-		clock:                     &clocks.RealClock{},
-		backupLogLevel:            backupLogLevel,
-		newPluginManager:          newPluginManager,
-		backupTracker:             backupTracker,
-		kbClient:                  kbClient,
-		defaultBackupLocation:     defaultBackupLocation,
-		defaultVolumesToFsBackup:  defaultVolumesToFsBackup,
-		defaultBackupTTL:          defaultBackupTTL,
-		defaultCSISnapshotTimeout: defaultCSISnapshotTimeout,
-		snapshotLocationLister:    volumeSnapshotLocationLister,
-		defaultSnapshotLocations:  defaultSnapshotLocations,
-		metrics:                   metrics,
-		backupStoreGetter:         backupStoreGetter,
-		formatFlag:                formatFlag,
-		volumeSnapshotLister:      volumeSnapshotLister,
-		volumeSnapshotClient:      volumeSnapshotClient,
-		credentialFileStore:       credentialStore,
+		genericController:           newGenericController(Backup, logger),
+		discoveryHelper:             discoveryHelper,
+		backupper:                   backupper,
+		lister:                      backupInformer.Lister(),
+		client:                      client,
+		clock:                       &clocks.RealClock{},
+		backupLogLevel:              backupLogLevel,
+		newPluginManager:            newPluginManager,
+		backupTracker:               backupTracker,
+		kbClient:                    kbClient,
+		defaultBackupLocation:       defaultBackupLocation,
+		defaultVolumesToFsBackup:    defaultVolumesToFsBackup,
+		defaultBackupTTL:            defaultBackupTTL,
+		defaultCSISnapshotTimeout:   defaultCSISnapshotTimeout,
+		defaultItemOperationTimeout: defaultItemOperationTimeout,
+		snapshotLocationLister:      volumeSnapshotLocationLister,
+		defaultSnapshotLocations:    defaultSnapshotLocations,
+		metrics:                     metrics,
+		backupStoreGetter:           backupStoreGetter,
+		formatFlag:                  formatFlag,
+		volumeSnapshotLister:        volumeSnapshotLister,
+		volumeSnapshotClient:        volumeSnapshotClient,
+		credentialFileStore:         credentialStore,
 	}
 
 	c.syncHandler = c.processBackup
@@ -366,6 +369,11 @@ func (c *backupController) prepareBackupRequest(backup *velerov1api.Backup, logg
 		request.Spec.CSISnapshotTimeout.Duration = c.defaultCSISnapshotTimeout
 	}
 
+	if request.Spec.ItemOperationTimeout.Duration == 0 {
+		// set default item operation timeout
+		request.Spec.ItemOperationTimeout.Duration = c.defaultItemOperationTimeout
+	}
+
 	// calculate expiration
 	request.Status.Expiration = &metav1.Time{Time: c.clock.Now().Add(request.Spec.TTL.Duration)}
 
@@ -705,10 +713,6 @@ func (c *backupController) runBackup(backup *pkgbackup.Request) error {
 		}
 	}
 
-	// Mark completion timestamp before serializing and uploading.
-	// Otherwise, the JSON file in object storage has a CompletionTimestamp of 'null'.
-	backup.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
-
 	backup.Status.VolumeSnapshotsAttempted = len(backup.VolumeSnapshots)
 	for _, snap := range backup.VolumeSnapshots {
 		if snap.Status.Phase == volume.SnapshotPhaseCompleted {
@@ -723,11 +727,24 @@ func (c *backupController) runBackup(backup *pkgbackup.Request) error {
 		}
 	}
 
+	// Iterate over backup item operations and update progress.
+	// Any errors on operations at this point should be added to backup errors.
+	// If any operations are still not complete, then back will not be set to
+	// Completed yet.
+	inProgressOperations, _, opsCompleted, opsFailed, errs := getBackupItemOperationProgress(backup.Backup, pluginManager, *backup.GetItemOperationsList())
+	if len(errs) > 0 {
+		for err := range errs {
+			backupLog.Error(err)
+		}
+	}
+
+	backup.Status.AsyncBackupItemOperationsAttempted = len(*backup.GetItemOperationsList())
+	backup.Status.AsyncBackupItemOperationsCompleted = opsCompleted
+	backup.Status.AsyncBackupItemOperationsFailed = opsFailed
+
 	backup.Status.Warnings = logCounter.GetCount(logrus.WarnLevel)
 	backup.Status.Errors = logCounter.GetCount(logrus.ErrorLevel)
 
-	recordBackupMetrics(backupLog, backup.Backup, backupFile, c.metrics)
-
 	backupWarnings := logCounter.GetEntries(logrus.WarnLevel)
 	backupErrors := logCounter.GetEntries(logrus.ErrorLevel)
 	results := map[string]results.Result{
@@ -747,10 +764,26 @@ func (c *backupController) runBackup(backup *pkgbackup.Request) error {
 	case len(fatalErrs) > 0:
 		backup.Status.Phase = velerov1api.BackupPhaseFailed
 	case logCounter.GetCount(logrus.ErrorLevel) > 0:
-		backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+		if inProgressOperations {
+			backup.Status.Phase = velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed
+		} else {
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed
+		}
 	default:
-		backup.Status.Phase = velerov1api.BackupPhaseCompleted
+		if inProgressOperations {
+			backup.Status.Phase = velerov1api.BackupPhaseWaitingForPluginOperations
+		} else {
+			backup.Status.Phase = velerov1api.BackupPhaseFinalizingAfterPluginOperations
+		}
 	}
+	// Mark completion timestamp before serializing and uploading.
+	// Otherwise, the JSON file in object storage has a CompletionTimestamp of 'null'.
+	if backup.Status.Phase == velerov1api.BackupPhaseFailed ||
+		backup.Status.Phase == velerov1api.BackupPhasePartiallyFailed ||
+		backup.Status.Phase == velerov1api.BackupPhaseCompleted {
+		backup.Status.CompletionTimestamp = &metav1.Time{Time: c.clock.Now()}
+	}
+	recordBackupMetrics(backupLog, backup.Backup, backupFile, c.metrics, false)
 
 	// re-instantiate the backup store because credentials could have changed since the original
 	// instantiation, if this was a long-running backup
@@ -771,37 +804,43 @@ func (c *backupController) runBackup(backup *pkgbackup.Request) error {
 	return kerrors.NewAggregate(fatalErrs)
 }
 
-func recordBackupMetrics(log logrus.FieldLogger, backup *velerov1api.Backup, backupFile *os.File, serverMetrics *metrics.ServerMetrics) {
+func recordBackupMetrics(log logrus.FieldLogger, backup *velerov1api.Backup, backupFile *os.File, serverMetrics *metrics.ServerMetrics, finalize bool) {
 	backupScheduleName := backup.GetLabels()[velerov1api.ScheduleNameLabel]
 
-	var backupSizeBytes int64
-	if backupFileStat, err := backupFile.Stat(); err != nil {
-		log.WithError(errors.WithStack(err)).Error("Error getting backup file info")
-	} else {
-		backupSizeBytes = backupFileStat.Size()
-	}
-	serverMetrics.SetBackupTarballSizeBytesGauge(backupScheduleName, backupSizeBytes)
-
-	backupDuration := backup.Status.CompletionTimestamp.Time.Sub(backup.Status.StartTimestamp.Time)
-	backupDurationSeconds := float64(backupDuration / time.Second)
-	serverMetrics.RegisterBackupDuration(backupScheduleName, backupDurationSeconds)
-	serverMetrics.RegisterVolumeSnapshotAttempts(backupScheduleName, backup.Status.VolumeSnapshotsAttempted)
-	serverMetrics.RegisterVolumeSnapshotSuccesses(backupScheduleName, backup.Status.VolumeSnapshotsCompleted)
-	serverMetrics.RegisterVolumeSnapshotFailures(backupScheduleName, backup.Status.VolumeSnapshotsAttempted-backup.Status.VolumeSnapshotsCompleted)
-
-	if features.IsEnabled(velerov1api.CSIFeatureFlag) {
-		serverMetrics.RegisterCSISnapshotAttempts(backupScheduleName, backup.Name, backup.Status.CSIVolumeSnapshotsAttempted)
-		serverMetrics.RegisterCSISnapshotSuccesses(backupScheduleName, backup.Name, backup.Status.CSIVolumeSnapshotsCompleted)
-		serverMetrics.RegisterCSISnapshotFailures(backupScheduleName, backup.Name, backup.Status.CSIVolumeSnapshotsAttempted-backup.Status.CSIVolumeSnapshotsCompleted)
+	if backupFile != nil {
+		var backupSizeBytes int64
+		if backupFileStat, err := backupFile.Stat(); err != nil {
+			log.WithError(errors.WithStack(err)).Error("Error getting backup file info")
+		} else {
+			backupSizeBytes = backupFileStat.Size()
+		}
+		serverMetrics.SetBackupTarballSizeBytesGauge(backupScheduleName, backupSizeBytes)
 	}
 
-	if backup.Status.Progress != nil {
-		serverMetrics.RegisterBackupItemsTotalGauge(backupScheduleName, backup.Status.Progress.TotalItems)
+	if backup.Status.CompletionTimestamp != nil {
+		backupDuration := backup.Status.CompletionTimestamp.Time.Sub(backup.Status.StartTimestamp.Time)
+		backupDurationSeconds := float64(backupDuration / time.Second)
+		serverMetrics.RegisterBackupDuration(backupScheduleName, backupDurationSeconds)
 	}
-	serverMetrics.RegisterBackupItemsErrorsGauge(backupScheduleName, backup.Status.Errors)
+	if !finalize {
+		serverMetrics.RegisterVolumeSnapshotAttempts(backupScheduleName, backup.Status.VolumeSnapshotsAttempted)
+		serverMetrics.RegisterVolumeSnapshotSuccesses(backupScheduleName, backup.Status.VolumeSnapshotsCompleted)
+		serverMetrics.RegisterVolumeSnapshotFailures(backupScheduleName, backup.Status.VolumeSnapshotsAttempted-backup.Status.VolumeSnapshotsCompleted)
 
-	if backup.Status.Warnings > 0 {
-		serverMetrics.RegisterBackupWarning(backupScheduleName)
+		if features.IsEnabled(velerov1api.CSIFeatureFlag) {
+			serverMetrics.RegisterCSISnapshotAttempts(backupScheduleName, backup.Name, backup.Status.CSIVolumeSnapshotsAttempted)
+			serverMetrics.RegisterCSISnapshotSuccesses(backupScheduleName, backup.Name, backup.Status.CSIVolumeSnapshotsCompleted)
+			serverMetrics.RegisterCSISnapshotFailures(backupScheduleName, backup.Name, backup.Status.CSIVolumeSnapshotsAttempted-backup.Status.CSIVolumeSnapshotsCompleted)
+		}
+
+		if backup.Status.Progress != nil {
+			serverMetrics.RegisterBackupItemsTotalGauge(backupScheduleName, backup.Status.Progress.TotalItems)
+		}
+		serverMetrics.RegisterBackupItemsErrorsGauge(backupScheduleName, backup.Status.Errors)
+
+		if backup.Status.Warnings > 0 {
+			serverMetrics.RegisterBackupWarning(backupScheduleName)
+		}
 	}
 }
 
@@ -826,6 +865,12 @@ func persistBackup(backup *pkgbackup.Request,
 		persistErrs = append(persistErrs, errs...)
 	}
 
+	var backupItemOperations *bytes.Buffer
+	backupItemOperations, errs = encodeToJSONGzip(backup.GetItemOperationsList(), "backup item operations list")
+	if errs != nil {
+		persistErrs = append(persistErrs, errs...)
+	}
+
 	podVolumeBackups, errs := encodeToJSONGzip(backup.PodVolumeBackups, "pod volume backups list")
 	if errs != nil {
 		persistErrs = append(persistErrs, errs...)
@@ -860,6 +905,7 @@ func persistBackup(backup *pkgbackup.Request,
 		backupJSON = nil
 		backupContents = nil
 		nativeVolumeSnapshots = nil
+		backupItemOperations = nil
 		backupResourceList = nil
 		csiSnapshotJSON = nil
 		csiSnapshotContentsJSON = nil
@@ -875,6 +921,7 @@ func persistBackup(backup *pkgbackup.Request,
 		BackupResults:             backupResult,
 		PodVolumeBackups:          podVolumeBackups,
 		VolumeSnapshots:           nativeVolumeSnapshots,
+		BackupItemOperations:      backupItemOperations,
 		BackupResourceList:        backupResourceList,
 		CSIVolumeSnapshots:        csiSnapshotJSON,
 		CSIVolumeSnapshotContents: csiSnapshotContentsJSON,

pkg/controller/backup_controller_test.go

@@ -47,6 +47,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/discovery"
 	"github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/fake"
 	informers "github.com/vmware-tanzu/velero/pkg/generated/informers/externalversions"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/persistence"
 	persistencemocks "github.com/vmware-tanzu/velero/pkg/persistence/mocks"
@@ -75,6 +76,13 @@ func (b *fakeBackupper) BackupWithResolvers(logger logrus.FieldLogger, backup *p
 	return args.Error(0)
 }
 
+func (b *fakeBackupper) FinalizeBackup(logger logrus.FieldLogger, backup *pkgbackup.Request, inBackupFile io.Reader, outBackupFile io.Writer,
+	backupItemActionResolver framework.BackupItemActionResolverV2,
+	asyncBIAOperations []*itemoperation.BackupOperation) error {
+	args := b.Called(logger, backup, inBackupFile, outBackupFile, backupItemActionResolver, asyncBIAOperations)
+	return args.Error(0)
+}
+
 func defaultBackup() *builder.BackupBuilder {
 	return builder.ForBackup(velerov1api.DefaultNamespace, "backup-1")
 }
@@ -597,7 +605,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 		backupExists             bool
 		existenceCheckError      error
 	}{
-		// Completed
+		// FinalizingAfterPluginOperations
 		{
 			name:                     "backup with no backup location gets the default",
 			backup:                   defaultBackup().Result(),
@@ -625,12 +633,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -661,12 +668,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -700,12 +706,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -737,12 +742,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					Expiration:          &metav1.Time{now.Add(10 * time.Minute)},
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					Expiration:     &metav1.Time{now.Add(10 * time.Minute)},
+					StartTimestamp: &timestamp,
 				},
 			},
 		},
@@ -774,12 +778,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -812,12 +815,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -850,12 +852,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -888,12 +889,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.True(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -926,12 +926,11 @@ func TestProcessBackupCompletions(t *testing.T) {
 					DefaultVolumesToFsBackup: boolptr.False(),
 				},
 				Status: velerov1api.BackupStatus{
-					Phase:               velerov1api.BackupPhaseCompleted,
-					Version:             1,
-					FormatVersion:       "1.1.0",
-					StartTimestamp:      &timestamp,
-					CompletionTimestamp: &timestamp,
-					Expiration:          &timestamp,
+					Phase:          velerov1api.BackupPhaseFinalizingAfterPluginOperations,
+					Version:        1,
+					FormatVersion:  "1.1.0",
+					StartTimestamp: &timestamp,
+					Expiration:     &timestamp,
 				},
 			},
 		},
@@ -1079,13 +1078,16 @@ func TestProcessBackupCompletions(t *testing.T) {
 
 			// Ensure we have a CompletionTimestamp when uploading and that the backup name matches the backup in the object store.
 			// Failures will display the bytes in buf.
-			hasNameAndCompletionTimestamp := func(info persistence.BackupInfo) bool {
+			hasNameAndCompletionTimestampIfCompleted := func(info persistence.BackupInfo) bool {
 				buf := new(bytes.Buffer)
 				buf.ReadFrom(info.Metadata)
 				return info.Name == test.backup.Name &&
-					strings.Contains(buf.String(), `"completionTimestamp": "2006-01-02T22:04:05Z"`)
+					(!(strings.Contains(buf.String(), `"phase": "Completed"`) ||
+						strings.Contains(buf.String(), `"phase": "Failed"`) ||
+						strings.Contains(buf.String(), `"phase": "PartiallyFailed"`)) ||
+						strings.Contains(buf.String(), `"completionTimestamp": "2006-01-02T22:04:05Z"`))
 			}
-			backupStore.On("PutBackup", mock.MatchedBy(hasNameAndCompletionTimestamp)).Return(nil)
+			backupStore.On("PutBackup", mock.MatchedBy(hasNameAndCompletionTimestampIfCompleted)).Return(nil)
 
 			// add the test's backup to the informer/lister store
 			require.NotNil(t, test.backup)

pkg/controller/backup_finalizer_controller.go

@@ -0,0 +1,204 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"bytes"
+	"context"
+	"io/ioutil"
+	"os"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	clocks "k8s.io/utils/clock"
+	ctrl "sigs.k8s.io/controller-runtime"
+	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	pkgbackup "github.com/vmware-tanzu/velero/pkg/backup"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
+	"github.com/vmware-tanzu/velero/pkg/persistence"
+	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
+	"github.com/vmware-tanzu/velero/pkg/util/encode"
+)
+
+// backupFinalizerReconciler reconciles a Backup object
+type backupFinalizerReconciler struct {
+	client            kbclient.Client
+	clock             clocks.WithTickerAndDelayedExecution
+	backupper         pkgbackup.Backupper
+	newPluginManager  func(logrus.FieldLogger) clientmgmt.Manager
+	metrics           *metrics.ServerMetrics
+	backupStoreGetter persistence.ObjectBackupStoreGetter
+	log               logrus.FieldLogger
+}
+
+// NewBackupFinalizerReconciler initializes and returns backupFinalizerReconciler struct.
+func NewBackupFinalizerReconciler(
+	client kbclient.Client,
+	clock clocks.WithTickerAndDelayedExecution,
+	backupper pkgbackup.Backupper,
+	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
+	backupStoreGetter persistence.ObjectBackupStoreGetter,
+	log logrus.FieldLogger,
+	metrics *metrics.ServerMetrics,
+) *backupFinalizerReconciler {
+	return &backupFinalizerReconciler{
+		client:            client,
+		clock:             clock,
+		backupper:         backupper,
+		newPluginManager:  newPluginManager,
+		backupStoreGetter: backupStoreGetter,
+		log:               log,
+		metrics:           metrics,
+	}
+}
+
+// +kubebuilder:rbac:groups=velero.io,resources=backups,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=velero.io,resources=backups/status,verbs=get;update;patch
+func (r *backupFinalizerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := r.log.WithFields(logrus.Fields{
+		"controller": "backup-finalizer",
+		"backup":     req.NamespacedName,
+	})
+
+	// Fetch the Backup instance.
+	log.Debug("Getting Backup")
+	backup := &velerov1api.Backup{}
+	if err := r.client.Get(ctx, req.NamespacedName, backup); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Debug("Unable to find Backup")
+			return ctrl.Result{}, nil
+		}
+
+		log.WithError(err).Error("Error getting Backup")
+		return ctrl.Result{}, errors.WithStack(err)
+	}
+
+	switch backup.Status.Phase {
+	case velerov1api.BackupPhaseFinalizingAfterPluginOperations, velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed:
+		// only process backups finalizing after  plugin operations are complete
+	default:
+		log.Debug("Backup is not awaiting finalizing, skipping")
+		return ctrl.Result{}, nil
+	}
+
+	original := backup.DeepCopy()
+	defer func() {
+		// Always attempt to Patch the backup object and status after each reconciliation.
+		if err := r.client.Patch(ctx, backup, kbclient.MergeFrom(original)); err != nil {
+			log.WithError(err).Error("Error updating backup")
+			return
+		}
+	}()
+
+	location := &velerov1api.BackupStorageLocation{}
+	if err := r.client.Get(ctx, kbclient.ObjectKey{
+		Namespace: backup.Namespace,
+		Name:      backup.Spec.StorageLocation,
+	}, location); err != nil {
+		return ctrl.Result{}, errors.WithStack(err)
+	}
+	pluginManager := r.newPluginManager(log)
+	defer pluginManager.CleanupClients()
+
+	backupStore, err := r.backupStoreGetter.Get(location, pluginManager, log)
+	if err != nil {
+		log.WithError(err).Error("Error getting a backup store")
+		return ctrl.Result{}, errors.WithStack(err)
+	}
+
+	// Download item operations list and backup contents
+	operations, err := backupStore.GetBackupItemOperations(backup.Name)
+	if err != nil {
+		log.WithError(err).Error("Error getting backup item operations")
+		return ctrl.Result{}, errors.WithStack(err)
+	}
+
+	backupRequest := &pkgbackup.Request{
+		Backup:          backup,
+		StorageLocation: location,
+	}
+	var outBackupFile *os.File
+	if len(operations) > 0 {
+		// Call itemBackupper.BackupItem for the list of items updated by async operations
+		log.Info("Setting up finalized backup temp file")
+		inBackupFile, err := downloadToTempFile(backup.Name, backupStore, log)
+		if err != nil {
+			return ctrl.Result{}, errors.Wrap(err, "error downloading backup")
+		}
+		defer closeAndRemoveFile(inBackupFile, log)
+		outBackupFile, err = ioutil.TempFile("", "")
+		if err != nil {
+			log.WithError(err).Error("error creating temp file for backup")
+			return ctrl.Result{}, errors.WithStack(err)
+		}
+		defer closeAndRemoveFile(outBackupFile, log)
+
+		log.Info("Getting backup item actions")
+		actions, err := pluginManager.GetBackupItemActionsV2()
+		if err != nil {
+			log.WithError(err).Error("error getting Backup Item Actions")
+			return ctrl.Result{}, errors.WithStack(err)
+		}
+		backupItemActionsResolver := framework.NewBackupItemActionResolverV2(actions)
+		err = r.backupper.FinalizeBackup(log, backupRequest, inBackupFile, outBackupFile, backupItemActionsResolver, operations)
+		if err != nil {
+			log.WithError(err).Error("error finalizing Backup")
+			return ctrl.Result{}, errors.WithStack(err)
+		}
+	}
+	backupScheduleName := backupRequest.GetLabels()[velerov1api.ScheduleNameLabel]
+	switch backup.Status.Phase {
+	case velerov1api.BackupPhaseFinalizingAfterPluginOperations:
+		backup.Status.Phase = velerov1api.BackupPhaseCompleted
+		r.metrics.RegisterBackupSuccess(backupScheduleName)
+		r.metrics.RegisterBackupLastStatus(backupScheduleName, metrics.BackupLastStatusSucc)
+	case velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed:
+		backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+		r.metrics.RegisterBackupPartialFailure(backupScheduleName)
+		r.metrics.RegisterBackupLastStatus(backupScheduleName, metrics.BackupLastStatusFailure)
+	}
+	backup.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	recordBackupMetrics(log, backup, outBackupFile, r.metrics, true)
+
+	// update backup metadata in object store
+	backupJSON := new(bytes.Buffer)
+	if err := encode.EncodeTo(backup, "json", backupJSON); err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error encoding backup json")
+	}
+	err = backupStore.PutBackupMetadata(backup.Name, backupJSON)
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error uploading backup json")
+	}
+	if len(operations) > 0 {
+		err = backupStore.PutBackupContents(backup.Name, outBackupFile)
+		if err != nil {
+			return ctrl.Result{}, errors.Wrap(err, "error uploading backup final contents")
+		}
+	}
+	return ctrl.Result{}, nil
+}
+
+func (r *backupFinalizerReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&velerov1api.Backup{}).
+		Complete(r)
+}

pkg/controller/backup_finalizer_controller_test.go

@@ -0,0 +1,184 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"bytes"
+	"context"
+	"io/ioutil"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	testclocks "k8s.io/utils/clock/testing"
+	ctrl "sigs.k8s.io/controller-runtime"
+	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/kuberesource"
+	"github.com/vmware-tanzu/velero/pkg/metrics"
+	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func mockBackupFinalizerReconciler(fakeClient kbclient.Client, fakeClock *testclocks.FakeClock) (*backupFinalizerReconciler, *fakeBackupper) {
+	backupper := new(fakeBackupper)
+	return NewBackupFinalizerReconciler(
+		fakeClient,
+		fakeClock,
+		backupper,
+		func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
+		NewFakeSingleObjectBackupStoreGetter(backupStore),
+		logrus.StandardLogger(),
+		metrics.NewServerMetrics(),
+	), backupper
+}
+func TestBackupFinalizerReconcile(t *testing.T) {
+	fakeClock := testclocks.NewFakeClock(time.Now())
+	metav1Now := metav1.NewTime(fakeClock.Now())
+
+	defaultBackupLocation := builder.ForBackupStorageLocation(velerov1api.DefaultNamespace, "default").Result()
+
+	tests := []struct {
+		name             string
+		backup           *velerov1api.Backup
+		backupOperations []*itemoperation.BackupOperation
+		backupLocation   *velerov1api.BackupStorageLocation
+		expectError      bool
+		expectPhase      velerov1api.BackupPhase
+	}{
+		{
+			name: "FinalizingAfterPluginOperations backup is completed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
+				StorageLocation("default").
+				ObjectMeta(builder.WithUID("foo")).
+				StartTimestamp(fakeClock.Now()).
+				Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).Result(),
+			backupLocation: defaultBackupLocation,
+			expectPhase:    velerov1api.BackupPhaseCompleted,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-1",
+						BackupUID:        "foo",
+						BackupItemAction: "foo",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						ItemsToUpdate: []velero.ResourceIdentifier{
+							{
+								GroupResource: kuberesource.Secrets,
+								Namespace:     "ns-1",
+								Name:          "secret-1",
+							},
+						},
+						OperationID: "operation-1",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseCompleted,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "FinalizingAfterPluginOperationsPartiallyFailed backup is partially failed",
+			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-2").
+				StorageLocation("default").
+				ObjectMeta(builder.WithUID("foo")).
+				StartTimestamp(fakeClock.Now()).
+				Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed).Result(),
+			backupLocation: defaultBackupLocation,
+			expectPhase:    velerov1api.BackupPhasePartiallyFailed,
+			backupOperations: []*itemoperation.BackupOperation{
+				{
+					Spec: itemoperation.BackupOperationSpec{
+						BackupName:       "backup-2",
+						BackupUID:        "foo",
+						BackupItemAction: "foo",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-2",
+							Name:          "pod-2",
+						},
+						ItemsToUpdate: []velero.ResourceIdentifier{
+							{
+								GroupResource: kuberesource.Secrets,
+								Namespace:     "ns-2",
+								Name:          "secret-2",
+							},
+						},
+						OperationID: "operation-2",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseCompleted,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if test.backup == nil {
+				return
+			}
+
+			initObjs := []runtime.Object{}
+			initObjs = append(initObjs, test.backup)
+
+			if test.backupLocation != nil {
+				initObjs = append(initObjs, test.backupLocation)
+			}
+
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, initObjs...)
+			reconciler, backupper := mockBackupFinalizerReconciler(fakeClient, fakeClock)
+			pluginManager.On("CleanupClients").Return(nil)
+			backupStore.On("GetBackupItemOperations", test.backup.Name).Return(test.backupOperations, nil)
+			backupStore.On("GetBackupContents", mock.Anything).Return(ioutil.NopCloser(bytes.NewReader([]byte("hello world"))), nil)
+			backupStore.On("PutBackupContents", mock.Anything, mock.Anything).Return(nil)
+			backupStore.On("PutBackupMetadata", mock.Anything, mock.Anything).Return(nil)
+			pluginManager.On("GetBackupItemActionsV2").Return(nil, nil)
+			backupper.On("FinalizeBackup", mock.Anything, mock.Anything, mock.Anything, mock.Anything, framework.BackupItemActionResolverV2{}, mock.Anything).Return(nil)
+			_, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: test.backup.Namespace, Name: test.backup.Name}})
+			gotErr := err != nil
+			assert.Equal(t, test.expectError, gotErr)
+
+			backupAfter := velerov1api.Backup{}
+			err = fakeClient.Get(context.TODO(), types.NamespacedName{
+				Namespace: test.backup.Namespace,
+				Name:      test.backup.Name,
+			}, &backupAfter)
+
+			require.NoError(t, err)
+			assert.Equal(t, test.expectPhase, backupAfter.Status.Phase)
+		})
+	}
+}

pkg/controller/backup_sync_controller.go

@@ -148,6 +148,26 @@ func (b *backupSyncReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			continue
 		}
 
+		if backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperations ||
+			backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed {
+
+			if backup.Status.Expiration == nil || backup.Status.Expiration.After(time.Now()) {
+				log.Debugf("Skipping non-expired WaitingForPluginOperations backup %v", backup.Name)
+				continue
+			}
+			log.Debug("WaitingForPluginOperations Backup is past expiration, syncing for garbage collection")
+			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+		}
+		if backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperations ||
+			backup.Status.Phase == velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed {
+
+			if backup.Status.Expiration == nil || backup.Status.Expiration.After(time.Now()) {
+				log.Debugf("Skipping non-expired FinalizingAfterPluginOperations backup %v", backup.Name)
+				continue
+			}
+			log.Debug("FinalizingAfterPluginOperations Backup is past expiration, syncing for garbage collection")
+			backup.Status.Phase = velerov1api.BackupPhasePartiallyFailed
+		}
 		backup.Namespace = b.namespace
 		backup.ResourceVersion = ""
 

pkg/controller/backup_sync_controller_test.go

@@ -25,6 +25,7 @@ import (
 	. "github.com/onsi/gomega"
 
 	"github.com/sirupsen/logrus"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -32,6 +33,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
 	core "k8s.io/client-go/testing"
+	testclocks "k8s.io/utils/clock/testing"
 
 	ctrl "sigs.k8s.io/controller-runtime"
 	ctrlClient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -155,9 +157,11 @@ func numBackups(c ctrlClient.WithWatch, ns string) (int, error) {
 
 var _ = Describe("Backup Sync Reconciler", func() {
 	It("Test Backup Sync Reconciler basic function", func() {
+		fakeClock := testclocks.NewFakeClock(time.Now())
 		type cloudBackupData struct {
-			backup           *velerov1api.Backup
-			podVolumeBackups []*velerov1api.PodVolumeBackup
+			backup               *velerov1api.Backup
+			podVolumeBackups     []*velerov1api.PodVolumeBackup
+			backupShouldSkipSync bool // backups waiting for plugin operations should not sync
 		}
 
 		tests := []struct {
@@ -187,6 +191,98 @@ var _ = Describe("Backup Sync Reconciler", func() {
 					},
 				},
 			},
+			{
+				name:      "backups waiting for plugin operations aren't synced",
+				namespace: "ns-1",
+				location:  defaultLocation("ns-1"),
+				cloudBackups: []*cloudBackupData{
+					{
+						backup: builder.ForBackup("ns-1", "backup-1").
+							Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-2").
+							Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-3").
+							Phase(velerov1api.BackupPhaseWaitingForPluginOperations).Result(),
+						podVolumeBackups: []*velerov1api.PodVolumeBackup{
+							builder.ForPodVolumeBackup("ns-1", "pvb-1").Result(),
+						},
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-4").
+							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-5").
+							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-6").
+							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).Result(),
+						podVolumeBackups: []*velerov1api.PodVolumeBackup{
+							builder.ForPodVolumeBackup("ns-1", "pvb-2").Result(),
+						},
+						backupShouldSkipSync: true,
+					},
+				},
+			},
+			{
+				name:      "expired backups waiting for plugin operations are synced",
+				namespace: "ns-1",
+				location:  defaultLocation("ns-1"),
+				cloudBackups: []*cloudBackupData{
+					{
+						backup: builder.ForBackup("ns-1", "backup-1").
+							Phase(velerov1api.BackupPhaseWaitingForPluginOperations).
+							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-2").
+							Phase(velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed).
+							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-3").
+							Phase(velerov1api.BackupPhaseWaitingForPluginOperations).
+							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+						podVolumeBackups: []*velerov1api.PodVolumeBackup{
+							builder.ForPodVolumeBackup("ns-1", "pvb-1").Result(),
+						},
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-4").
+							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).
+							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-5").
+							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperationsPartiallyFailed).
+							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+						backupShouldSkipSync: true,
+					},
+					{
+						backup: builder.ForBackup("ns-1", "backup-6").
+							Phase(velerov1api.BackupPhaseFinalizingAfterPluginOperations).
+							Expiration(fakeClock.Now().Add(-time.Hour)).Result(),
+						podVolumeBackups: []*velerov1api.PodVolumeBackup{
+							builder.ForPodVolumeBackup("ns-1", "pvb-2").Result(),
+						},
+						backupShouldSkipSync: true,
+					},
+				},
+			},
 			{
 				name:      "all synced backups get created in Velero server's namespace",
 				namespace: "velero",
@@ -364,36 +460,42 @@ var _ = Describe("Backup Sync Reconciler", func() {
 						Namespace: cloudBackupData.backup.Namespace,
 						Name:      cloudBackupData.backup.Name},
 					obj)
-				Expect(err).To(BeNil())
-
-				// did this cloud backup already exist in the cluster?
-				var existing *velerov1api.Backup
-				for _, obj := range test.existingBackups {
-					if obj.Name == cloudBackupData.backup.Name {
-						existing = obj
-						break
-					}
-				}
-
-				if existing != nil {
-					// if this cloud backup already exists in the cluster, make sure that what we get from the
-					// client is the existing backup, not the cloud one.
-
-					// verify that the in-cluster backup has its storage location populated, if it's not already.
-					expected := existing.DeepCopy()
-					expected.Spec.StorageLocation = test.location.Name
-
-					Expect(expected).To(BeEquivalentTo(obj))
+				if cloudBackupData.backupShouldSkipSync &&
+					(cloudBackupData.backup.Status.Expiration == nil ||
+						cloudBackupData.backup.Status.Expiration.After(fakeClock.Now())) {
+					Expect(apierrors.IsNotFound(err)).To(BeTrue())
 				} else {
-					// verify that the storage location field and label are set properly
-					Expect(test.location.Name).To(BeEquivalentTo(obj.Spec.StorageLocation))
+					Expect(err).To(BeNil())
 
-					locationName := test.location.Name
-					if test.longLocationNameEnabled {
-						locationName = label.GetValidName(locationName)
+					// did this cloud backup already exist in the cluster?
+					var existing *velerov1api.Backup
+					for _, obj := range test.existingBackups {
+						if obj.Name == cloudBackupData.backup.Name {
+							existing = obj
+							break
+						}
+					}
+
+					if existing != nil {
+						// if this cloud backup already exists in the cluster, make sure that what we get from the
+						// client is the existing backup, not the cloud one.
+
+						// verify that the in-cluster backup has its storage location populated, if it's not already.
+						expected := existing.DeepCopy()
+						expected.Spec.StorageLocation = test.location.Name
+
+						Expect(expected).To(BeEquivalentTo(obj))
+					} else {
+						// verify that the storage location field and label are set properly
+						Expect(test.location.Name).To(BeEquivalentTo(obj.Spec.StorageLocation))
+
+						locationName := test.location.Name
+						if test.longLocationNameEnabled {
+							locationName = label.GetValidName(locationName)
+						}
+						Expect(locationName).To(BeEquivalentTo(obj.Labels[velerov1api.StorageLocationLabel]))
+						Expect(len(obj.Labels[velerov1api.StorageLocationLabel]) <= validation.DNS1035LabelMaxLength).To(BeTrue())
 					}
-					Expect(locationName).To(BeEquivalentTo(obj.Labels[velerov1api.StorageLocationLabel]))
-					Expect(len(obj.Labels[velerov1api.StorageLocationLabel]) <= validation.DNS1035LabelMaxLength).To(BeTrue())
 				}
 
 				// process the cloud pod volume backups for this backup, if any
@@ -406,22 +508,28 @@ var _ = Describe("Backup Sync Reconciler", func() {
 							Name:      podVolumeBackup.Name,
 						},
 						objPodVolumeBackup)
-					Expect(err).ShouldNot(HaveOccurred())
+					if cloudBackupData.backupShouldSkipSync &&
+						(cloudBackupData.backup.Status.Expiration == nil ||
+							cloudBackupData.backup.Status.Expiration.After(fakeClock.Now())) {
+						Expect(apierrors.IsNotFound(err)).To(BeTrue())
+					} else {
+						Expect(err).ShouldNot(HaveOccurred())
 
-					// did this cloud pod volume backup already exist in the cluster?
-					var existingPodVolumeBackup *velerov1api.PodVolumeBackup
-					for _, objPodVolumeBackup := range test.existingPodVolumeBackups {
-						if objPodVolumeBackup.Name == podVolumeBackup.Name {
-							existingPodVolumeBackup = objPodVolumeBackup
-							break
+						// did this cloud pod volume backup already exist in the cluster?
+						var existingPodVolumeBackup *velerov1api.PodVolumeBackup
+						for _, objPodVolumeBackup := range test.existingPodVolumeBackups {
+							if objPodVolumeBackup.Name == podVolumeBackup.Name {
+								existingPodVolumeBackup = objPodVolumeBackup
+								break
+							}
 						}
-					}
 
-					if existingPodVolumeBackup != nil {
-						// if this cloud pod volume backup already exists in the cluster, make sure that what we get from the
-						// client is the existing backup, not the cloud one.
-						expected := existingPodVolumeBackup.DeepCopy()
-						Expect(expected).To(BeEquivalentTo(objPodVolumeBackup))
+						if existingPodVolumeBackup != nil {
+							// if this cloud pod volume backup already exists in the cluster, make sure that what we get from the
+							// client is the existing backup, not the cloud one.
+							expected := existingPodVolumeBackup.DeepCopy()
+							Expect(expected).To(BeEquivalentTo(objPodVolumeBackup))
+						}
 					}
 				}
 			}

pkg/controller/constants.go

@@ -17,8 +17,10 @@ limitations under the License.
 package controller
 
 const (
+	AsyncBackupOperations = "async-backup-operations"
 	Backup                = "backup"
 	BackupDeletion        = "backup-deletion"
+	BackupFinalizer       = "backup-finalizer"
 	BackupStorageLocation = "backup-storage-location"
 	BackupSync            = "backup-sync"
 	DownloadRequest       = "download-request"
@@ -33,8 +35,10 @@ const (
 
 // DisableableControllers is a list of controllers that can be disabled
 var DisableableControllers = []string{
+	AsyncBackupOperations,
 	Backup,
 	BackupDeletion,
+	BackupFinalizer,
 	BackupSync,
 	DownloadRequest,
 	GarbageCollection,

pkg/controller/download_request_controller.go

@@ -41,6 +41,9 @@ type downloadRequestReconciler struct {
 	newPluginManager  func(logrus.FieldLogger) clientmgmt.Manager
 	backupStoreGetter persistence.ObjectBackupStoreGetter
 
+	// used to force update of async backup item operations before processing download request
+	backupItemOperationsMap *BackupItemOperationsMap
+
 	log logrus.FieldLogger
 }
 
@@ -51,13 +54,15 @@ func NewDownloadRequestReconciler(
 	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	log logrus.FieldLogger,
+	backupItemOperationsMap *BackupItemOperationsMap,
 ) *downloadRequestReconciler {
 	return &downloadRequestReconciler{
-		client:            client,
-		clock:             clock,
-		newPluginManager:  newPluginManager,
-		backupStoreGetter: backupStoreGetter,
-		log:               log,
+		client:                  client,
+		clock:                   clock,
+		newPluginManager:        newPluginManager,
+		backupStoreGetter:       backupStoreGetter,
+		backupItemOperationsMap: backupItemOperationsMap,
+		log:                     log,
 	}
 }
 
@@ -158,6 +163,13 @@ func (r *downloadRequestReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			return ctrl.Result{}, errors.WithStack(err)
 		}
 
+		// If this is a request for backup item operations, force update of in-memory operations that
+		// are not yet uploaded
+		if downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindBackupItemOperations &&
+			r.backupItemOperationsMap != nil {
+			// ignore errors here. If we can't upload anything here, process the download as usual
+			_ = r.backupItemOperationsMap.UpdateForBackup(backupStore, backupName)
+		}
 		if downloadRequest.Status.DownloadURL, err = backupStore.GetDownloadURL(downloadRequest.Spec.Target); err != nil {
 			return ctrl.Result{Requeue: true}, errors.WithStack(err)
 		}

pkg/controller/download_request_controller_test.go

@@ -112,6 +112,7 @@ var _ = Describe("Download Request Reconciler", func() {
 				func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
 				NewFakeObjectBackupStoreGetter(backupStores),
 				velerotest.NewLogger(),
+				nil,
 			)
 
 			if test.backupLocation != nil && test.expectGetsURL {