RIAv2 async operations controller work

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

changelogs/unreleased/5993-sseago

@@ -0,0 +1 @@
+RIAv2 async operations controller work

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

@@ -238,6 +238,10 @@ spec:
                   type: string
                 nullable: true
                 type: array
+              itemOperationTimeout:
+                description: ItemOperationTimeout specifies the time used to wait
+                  for RestoreItemAction operations The default value is 1 hour.
+                type: string
               labelSelector:
                 description: LabelSelector is a metav1.LabelSelector to filter with
                   when restoring individual objects from the backup. If empty or nil,
@@ -434,6 +438,20 @@ spec:
                       due to plugins that return additional related items to restore
                     type: integer
                 type: object
+              restoreItemOperationsAttempted:
+                description: RestoreItemOperationsAttempted is the total number of
+                  attempted async RestoreItemAction operations for this restore.
+                type: integer
+              restoreItemOperationsCompleted:
+                description: RestoreItemOperationsCompleted is the total number of
+                  successfully completed async RestoreItemAction operations for this
+                  restore.
+                type: integer
+              restoreItemOperationsFailed:
+                description: RestoreItemOperationsFailed is the total number of async
+                  RestoreItemAction operations for this restore which ended with an
+                  error.
+                type: integer
               startTimestamp:
                 description: StartTimestamp records the time the restore operation
                   was started. The server's time is used for StartTimestamps

config/rbac/role.yaml

@@ -184,6 +184,12 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - velero.io
+  resources:
+  - restorestoragelocations
+  verbs:
+  - get
 - apiGroups:
   - velero.io
   resources:

design/riav2-design.md

@@ -98,6 +98,27 @@ message OperationProgress {
 }
 ```
 
+In addition to the three new rpc methods added to the RestoreItemAction interface, there is also a new `Name()` method. This one is only actually used internally by Velero to get the name that the plugin was registered with, but it still must be defined in a plugin which implements RestoreItemActionV2 in order to implement the interface. It doesn't really matter what it returns, though, as this particular method is not delegated to the plugin via RPC calls. The new (and modified) interface methods for `RestoreItemAction` are as follows:
+```
+type BackupItemAction interface {
+...
+	Name() string
+...
+	Progress(operationID string, restore *api.Restore) (velero.OperationProgress, error)
+	Cancel(operationID string, backup *api.Restore) error
+	AreAdditionalItemsReady(AdditionalItems []velero.ResourceIdentifier, restore *api.Restore) (bool, error)
+...
+}
+type RestoreItemActionExecuteOutput struct {
+	UpdatedItem            runtime.Unstructured
+	AdditionalItems        []ResourceIdentifier
+	SkipRestore            bool
+	OperationID            string
+	WaitForAdditionalItems bool
+}
+
+```
+
 A new PluginKind, `RestoreItemActionV2`, will be created, and the restore process will be modified to use this plugin kind.
 See [Plugin Versioning](plugin-versioning.md) for more details on implementation plans, including v1 adapters, etc.
 

pkg/apis/velero/v1/restore_types.go

@@ -112,6 +112,11 @@ type RestoreSpec struct {
 	// +optional
 	// +nullable
 	ExistingResourcePolicy PolicyType `json:"existingResourcePolicy,omitempty"`
+
+	// ItemOperationTimeout specifies the time used to wait for RestoreItemAction operations
+	// The default value is 1 hour.
+	// +optional
+	ItemOperationTimeout metav1.Duration `json:"itemOperationTimeout,omitempty"`
 }
 
 // RestoreHooks contains custom behaviors that should be executed during or post restore.
@@ -314,6 +319,21 @@ type RestoreStatus struct {
 	// +optional
 	// +nullable
 	Progress *RestoreProgress `json:"progress,omitempty"`
+
+	// RestoreItemOperationsAttempted is the total number of attempted
+	// async RestoreItemAction operations for this restore.
+	// +optional
+	RestoreItemOperationsAttempted int `json:"restoreItemOperationsAttempted,omitempty"`
+
+	// RestoreItemOperationsCompleted is the total number of successfully completed
+	// async RestoreItemAction operations for this restore.
+	// +optional
+	RestoreItemOperationsCompleted int `json:"restoreItemOperationsCompleted,omitempty"`
+
+	// RestoreItemOperationsFailed is the total number of async
+	// RestoreItemAction operations for this restore which ended with an error.
+	// +optional
+	RestoreItemOperationsFailed int `json:"restoreItemOperationsFailed,omitempty"`
 }
 
 // RestoreProgress stores information about the restore's execution progress

pkg/apis/velero/v1/zz_generated.deepcopy.go

@@ -1321,6 +1321,7 @@ func (in *RestoreSpec) DeepCopyInto(out *RestoreSpec) {
 		**out = **in
 	}
 	in.Hooks.DeepCopyInto(&out.Hooks)
+	out.ItemOperationTimeout = in.ItemOperationTimeout
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RestoreSpec.

pkg/builder/restore_builder.go

@@ -165,3 +165,9 @@ func (b *RestoreBuilder) CompletionTimestamp(val time.Time) *RestoreBuilder {
 	b.object.Status.CompletionTimestamp = &metav1.Time{Time: val}
 	return b
 }
+
+// ItemOperationTimeout sets the Restore's ItemOperationTimeout
+func (b *RestoreBuilder) ItemOperationTimeout(timeout time.Duration) *RestoreBuilder {
+	b.object.Spec.ItemOperationTimeout.Duration = timeout
+	return b
+}

pkg/cmd/cli/restore/create.go

@@ -92,6 +92,7 @@ type CreateOptions struct {
 	IncludeClusterResources flag.OptionalBool
 	Wait                    bool
 	AllowPartiallyFailed    flag.OptionalBool
+	ItemOperationTimeout    time.Duration
 
 	client veleroclient.Interface
 }
@@ -120,6 +121,7 @@ func (o *CreateOptions) BindFlags(flags *pflag.FlagSet) {
 	flags.Var(&o.StatusIncludeResources, "status-include-resources", "Resources to include in the restore status, formatted as resource.group, such as storageclasses.storage.k8s.io.")
 	flags.Var(&o.StatusExcludeResources, "status-exclude-resources", "Resources to exclude from the restore status, formatted as resource.group, such as storageclasses.storage.k8s.io.")
 	flags.VarP(&o.Selector, "selector", "l", "Only restore resources matching this label selector.")
+	flags.DurationVar(&o.ItemOperationTimeout, "item-operation-timeout", o.ItemOperationTimeout, "How long to wait for async plugin operations before timeout.")
 	f := flags.VarPF(&o.RestoreVolumes, "restore-volumes", "", "Whether to restore volumes from snapshots.")
 	// this allows the user to just specify "--restore-volumes" as shorthand for "--restore-volumes=true"
 	// like a normal bool flag
@@ -280,6 +282,9 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
 			RestorePVs:              o.RestoreVolumes.Value,
 			PreserveNodePorts:       o.PreserveNodePorts.Value,
 			IncludeClusterResources: o.IncludeClusterResources.Value,
+			ItemOperationTimeout: metav1.Duration{
+				Duration: o.ItemOperationTimeout,
+			},
 		},
 	}
 

pkg/cmd/cli/restore/logs.go

@@ -63,8 +63,8 @@ func NewLogsCommand(f client.Factory) *cobra.Command {
 			}
 
 			switch restore.Status.Phase {
-			case velerov1api.RestorePhaseCompleted, velerov1api.RestorePhaseFailed, velerov1api.RestorePhasePartiallyFailed:
-				// terminal phases, don't exit.
+			case velerov1api.RestorePhaseCompleted, velerov1api.RestorePhaseFailed, velerov1api.RestorePhasePartiallyFailed, velerov1api.RestorePhaseWaitingForPluginOperations, velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed:
+				// terminal and waiting for plugin operations phases, don't exit.
 			default:
 				cmd.Exit("Logs for restore %q are not available until it's finished processing. Please wait "+
 					"until the restore has a phase of Completed or Failed and try again.", restoreName)

pkg/cmd/server/server.go

@@ -652,6 +652,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		controller.DownloadRequest:     {},
 		controller.GarbageCollection:   {},
 		controller.Restore:             {},
+		controller.RestoreOperations:   {},
 		controller.Schedule:            {},
 		controller.ServerStatusRequest: {},
 	}
@@ -831,6 +832,23 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		}
 	}
 
+	restoreOpsMap := itemoperationmap.NewRestoreItemOperationsMap()
+	if _, ok := enabledRuntimeControllers[controller.RestoreOperations]; ok {
+		r := controller.NewRestoreOperationsReconciler(
+			s.logger,
+			s.namespace,
+			s.mgr.GetClient(),
+			s.config.itemOperationSyncFrequency,
+			newPluginManager,
+			backupStoreGetter,
+			s.metrics,
+			restoreOpsMap,
+		)
+		if err := r.SetupWithManager(s.mgr); err != nil {
+			s.logger.Fatal(err, "unable to create controller", "controller", controller.BackupOperations)
+		}
+	}
+
 	if _, ok := enabledRuntimeControllers[controller.DownloadRequest]; ok {
 		r := controller.NewDownloadRequestReconciler(
 			s.mgr.GetClient(),
@@ -839,6 +857,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			backupStoreGetter,
 			s.logger,
 			backupOpsMap,
+			restoreOpsMap,
 		)
 		if err := r.SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", controller.DownloadRequest)
@@ -890,6 +909,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			backupStoreGetter,
 			s.metrics,
 			s.config.formatFlag.Parse(),
+			s.config.defaultItemOperationTimeout,
 		)
 
 		if err = r.SetupWithManager(s.mgr); err != nil {

pkg/cmd/util/output/restore_describer.go

@@ -32,6 +32,7 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/downloadrequest"
 	clientset "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/util/results"
 )
 
@@ -154,11 +155,14 @@ func DescribeRestore(ctx context.Context, kbClient kbclient.Client, restore *vel
 			s = string(restore.Spec.ExistingResourcePolicy)
 		}
 		d.Printf("Existing Resource Policy: \t%s\n", s)
+		d.Printf("ItemOperationTimeout:\t%s\n", restore.Spec.ItemOperationTimeout.Duration)
 
 		d.Println()
 		d.Printf("Preserve Service NodePorts:\t%s\n", BoolPointerString(restore.Spec.PreserveNodePorts, "false", "true", "auto"))
 
 		d.Println()
+		describeRestoreItemOperations(ctx, kbClient, d, restore, details, insecureSkipTLSVerify, caCertFile)
+
 		if details {
 			describeRestoreResourceList(ctx, kbClient, d, restore, insecureSkipTLSVerify, caCertFile)
 			d.Println()
@@ -166,6 +170,33 @@ func DescribeRestore(ctx context.Context, kbClient kbclient.Client, restore *vel
 	})
 }
 
+func describeRestoreItemOperations(ctx context.Context, kbClient kbclient.Client, d *Describer, restore *velerov1api.Restore, details bool, insecureSkipTLSVerify bool, caCertPath string) {
+	status := restore.Status
+	if status.RestoreItemOperationsAttempted > 0 {
+		if !details {
+			d.Printf("Restore Item Operations:\t%d of %d completed successfully, %d failed (specify --details for more information)\n", status.RestoreItemOperationsCompleted, status.RestoreItemOperationsAttempted, status.RestoreItemOperationsFailed)
+			return
+		}
+
+		buf := new(bytes.Buffer)
+		if err := downloadrequest.Stream(ctx, kbClient, restore.Namespace, restore.Name, velerov1api.DownloadTargetKindRestoreItemOperations, buf, downloadRequestTimeout, insecureSkipTLSVerify, caCertPath); err != nil {
+			d.Printf("Restore Item Operations:\t<error getting operation info: %v>\n", err)
+			return
+		}
+
+		var operations []*itemoperation.RestoreOperation
+		if err := json.NewDecoder(buf).Decode(&operations); err != nil {
+			d.Printf("Restore Item Operations:\t<error reading operation info: %v>\n", err)
+			return
+		}
+
+		d.Printf("Restore Item Operations:\n")
+		for _, operation := range operations {
+			describeRestoreItemOperation(d, operation)
+		}
+	}
+}
+
 func describeRestoreResults(ctx context.Context, kbClient kbclient.Client, d *Describer, restore *velerov1api.Restore, insecureSkipTLSVerify bool, caCertPath string) {
 	if restore.Status.Warnings == 0 && restore.Status.Errors == 0 {
 		return
@@ -208,6 +239,34 @@ func describeResult(d *Describer, name string, result results.Result) {
 	}
 }
 
+func describeRestoreItemOperation(d *Describer, operation *itemoperation.RestoreOperation) {
+	d.Printf("\tOperation for %s %s/%s:\n", operation.Spec.ResourceIdentifier, operation.Spec.ResourceIdentifier.Namespace, operation.Spec.ResourceIdentifier.Name)
+	d.Printf("\t\tRestore Item Action Plugin:\t%s\n", operation.Spec.RestoreItemAction)
+	d.Printf("\t\tOperation ID:\t%s\n", operation.Spec.OperationID)
+	d.Printf("\t\tPhase:\t%s\n", operation.Status.Phase)
+	if operation.Status.Error != "" {
+		d.Printf("\t\tOperation Error:\t%s\n", operation.Status.Error)
+	}
+	if operation.Status.NTotal > 0 || operation.Status.NCompleted > 0 {
+		d.Printf("\t\tProgress:\t%v of %v complete (%s)\n",
+			operation.Status.NCompleted,
+			operation.Status.NTotal,
+			operation.Status.OperationUnits)
+	}
+	if operation.Status.Description != "" {
+		d.Printf("\t\tProgress description:\t%s\n", operation.Status.Description)
+	}
+	if operation.Status.Created != nil {
+		d.Printf("\t\tCreated:\t%s\n", operation.Status.Created.String())
+	}
+	if operation.Status.Started != nil {
+		d.Printf("\t\tStarted:\t%s\n", operation.Status.Started.String())
+	}
+	if operation.Status.Updated != nil {
+		d.Printf("\t\tUpdated:\t%s\n", operation.Status.Updated.String())
+	}
+}
+
 // describePodVolumeRestores describes pod volume restores in human-readable format.
 func describePodVolumeRestores(d *Describer, restores []velerov1api.PodVolumeRestore, details bool) {
 	// Get the type of pod volume uploader. Since the uploader only comes from a single source, we can

pkg/controller/constants.go

@@ -17,8 +17,8 @@ limitations under the License.
 package controller
 
 const (
-	BackupOperations      = "backup-operations"
 	Backup                = "backup"
+	BackupOperations      = "backup-operations"
 	BackupDeletion        = "backup-deletion"
 	BackupFinalizer       = "backup-finalizer"
 	BackupRepo            = "backup-repo"
@@ -29,14 +29,15 @@ const (
 	PodVolumeBackup       = "pod-volume-backup"
 	PodVolumeRestore      = "pod-volume-restore"
 	Restore               = "restore"
+	RestoreOperations     = "restore-operations"
 	Schedule              = "schedule"
 	ServerStatusRequest   = "server-status-request"
 )
 
 // DisableableControllers is a list of controllers that can be disabled
 var DisableableControllers = []string{
-	BackupOperations,
 	Backup,
+	BackupOperations,
 	BackupDeletion,
 	BackupFinalizer,
 	BackupSync,
@@ -44,6 +45,7 @@ var DisableableControllers = []string{
 	GarbageCollection,
 	BackupRepo,
 	Restore,
+	RestoreOperations,
 	Schedule,
 	ServerStatusRequest,
 }

pkg/controller/download_request_controller.go

@@ -44,6 +44,8 @@ type downloadRequestReconciler struct {
 
 	// used to force update of async backup item operations before processing download request
 	backupItemOperationsMap *itemoperationmap.BackupItemOperationsMap
+	// used to force update of async restore item operations before processing download request
+	restoreItemOperationsMap *itemoperationmap.RestoreItemOperationsMap
 
 	log logrus.FieldLogger
 }
@@ -56,14 +58,16 @@ func NewDownloadRequestReconciler(
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	log logrus.FieldLogger,
 	backupItemOperationsMap *itemoperationmap.BackupItemOperationsMap,
+	restoreItemOperationsMap *itemoperationmap.RestoreItemOperationsMap,
 ) *downloadRequestReconciler {
 	return &downloadRequestReconciler{
-		client:                  client,
-		clock:                   clock,
-		newPluginManager:        newPluginManager,
-		backupStoreGetter:       backupStoreGetter,
-		backupItemOperationsMap: backupItemOperationsMap,
-		log:                     log,
+		client:                   client,
+		clock:                    clock,
+		newPluginManager:         newPluginManager,
+		backupStoreGetter:        backupStoreGetter,
+		backupItemOperationsMap:  backupItemOperationsMap,
+		restoreItemOperationsMap: restoreItemOperationsMap,
+		log:                      log,
 	}
 }
 
@@ -129,7 +133,8 @@ func (r *downloadRequestReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 
 		if downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindRestoreLog ||
 			downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindRestoreResults ||
-			downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindRestoreResourceList {
+			downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindRestoreResourceList ||
+			downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindRestoreItemOperations {
 			restore := &velerov1api.Restore{}
 			if err := r.client.Get(ctx, kbclient.ObjectKey{
 				Namespace: downloadRequest.Namespace,
@@ -172,6 +177,13 @@ func (r *downloadRequestReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 			// ignore errors here. If we can't upload anything here, process the download as usual
 			_ = r.backupItemOperationsMap.UpdateForBackup(backupStore, backupName)
 		}
+		// If this is a request for restore item operations, force upload of in-memory operations that
+		// are not yet uploaded (if there are any)
+		if downloadRequest.Spec.Target.Kind == velerov1api.DownloadTargetKindRestoreItemOperations &&
+			r.restoreItemOperationsMap != nil {
+			// ignore errors here. If we can't upload anything here, process the download as usual
+			_ = r.restoreItemOperationsMap.UpdateForRestore(backupStore, downloadRequest.Spec.Target.Name)
+		}
 		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

@@ -113,6 +113,7 @@ var _ = Describe("Download Request Reconciler", func() {
 				NewFakeObjectBackupStoreGetter(backupStores),
 				velerotest.NewLogger(),
 				nil,
+				nil,
 			)
 
 			if test.backupLocation != nil && test.expectGetsURL {

pkg/controller/restore_controller.go

@@ -40,6 +40,7 @@ import (
 
 	"github.com/vmware-tanzu/velero/internal/hook"
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/metrics"
 	"github.com/vmware-tanzu/velero/pkg/persistence"
@@ -84,15 +85,16 @@ var nonRestorableResources = []string{
 }
 
 type restoreReconciler struct {
-	ctx             context.Context
-	namespace       string
-	restorer        pkgrestore.Restorer
-	kbClient        client.Client
-	restoreLogLevel logrus.Level
-	logger          logrus.FieldLogger
-	metrics         *metrics.ServerMetrics
-	logFormat       logging.Format
-	clock           clock.WithTickerAndDelayedExecution
+	ctx                         context.Context
+	namespace                   string
+	restorer                    pkgrestore.Restorer
+	kbClient                    client.Client
+	restoreLogLevel             logrus.Level
+	logger                      logrus.FieldLogger
+	metrics                     *metrics.ServerMetrics
+	logFormat                   logging.Format
+	clock                       clock.WithTickerAndDelayedExecution
+	defaultItemOperationTimeout time.Duration
 
 	newPluginManager  func(logger logrus.FieldLogger) clientmgmt.Manager
 	backupStoreGetter persistence.ObjectBackupStoreGetter
@@ -114,17 +116,19 @@ func NewRestoreReconciler(
 	backupStoreGetter persistence.ObjectBackupStoreGetter,
 	metrics *metrics.ServerMetrics,
 	logFormat logging.Format,
+	defaultItemOperationTimeout time.Duration,
 ) *restoreReconciler {
 	r := &restoreReconciler{
-		ctx:             ctx,
-		namespace:       namespace,
-		restorer:        restorer,
-		kbClient:        kbClient,
-		logger:          logger,
-		restoreLogLevel: restoreLogLevel,
-		metrics:         metrics,
-		logFormat:       logFormat,
-		clock:           &clock.RealClock{},
+		ctx:                         ctx,
+		namespace:                   namespace,
+		restorer:                    restorer,
+		kbClient:                    kbClient,
+		logger:                      logger,
+		restoreLogLevel:             restoreLogLevel,
+		metrics:                     metrics,
+		logFormat:                   logFormat,
+		clock:                       &clock.RealClock{},
+		defaultItemOperationTimeout: defaultItemOperationTimeout,
 
 		// use variables to refer to these functions so they can be
 		// replaced with fakes for testing.
@@ -172,6 +176,10 @@ func (r *restoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		restore.Status.StartTimestamp = &metav1.Time{Time: r.clock.Now()}
 		restore.Status.Phase = api.RestorePhaseInProgress
 	}
+	if restore.Spec.ItemOperationTimeout.Duration == 0 {
+		// set default item operation timeout
+		restore.Spec.ItemOperationTimeout.Duration = r.defaultItemOperationTimeout
+	}
 
 	// patch to update status and persist to API
 	err = kubeutil.PatchResource(original, restore, r.kbClient)
@@ -194,17 +202,14 @@ func (r *restoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		restore.Status.Phase = api.RestorePhaseFailed
 		restore.Status.FailureReason = err.Error()
 		r.metrics.RegisterRestoreFailed(backupScheduleName)
-	} else if restore.Status.Errors > 0 {
-		log.Debug("Restore partially failed")
-		restore.Status.Phase = api.RestorePhasePartiallyFailed
-		r.metrics.RegisterRestorePartialFailure(backupScheduleName)
-	} else {
-		log.Debug("Restore completed")
-		restore.Status.Phase = api.RestorePhaseCompleted
-		r.metrics.RegisterRestoreSuccess(backupScheduleName)
 	}
 
-	restore.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	// mark completion if in terminal phase
+	if restore.Status.Phase == api.RestorePhaseFailed ||
+		restore.Status.Phase == api.RestorePhasePartiallyFailed ||
+		restore.Status.Phase == api.RestorePhaseCompleted {
+		restore.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
+	}
 	log.Debug("Updating restore's final status")
 	if err = kubeutil.PatchResource(original, restore, r.kbClient); err != nil {
 		log.WithError(errors.WithStack(err)).Info("Error updating restore's final status")
@@ -375,15 +380,19 @@ func mostRecentCompletedBackup(backups []api.Backup) api.Backup {
 // fetchBackupInfo checks the backup lister for a backup that matches the given name. If it doesn't
 // find it, it returns an error.
 func (r *restoreReconciler) fetchBackupInfo(backupName string) (backupInfo, error) {
+	return fetchBackupInfoInternal(r.kbClient, r.namespace, backupName)
+}
+
+func fetchBackupInfoInternal(kbClient client.Client, namespace, backupName string) (backupInfo, error) {
 	backup := &api.Backup{}
-	err := r.kbClient.Get(context.Background(), types.NamespacedName{Namespace: r.namespace, Name: backupName}, backup)
+	err := kbClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: backupName}, backup)
 	if err != nil {
-		return backupInfo{}, err
+		return backupInfo{}, errors.Wrap(err, fmt.Sprintf("can't find backup %s/%s", namespace, backupName))
 	}
 
 	location := &api.BackupStorageLocation{}
-	if err := r.kbClient.Get(context.Background(), client.ObjectKey{
-		Namespace: r.namespace,
+	if err := kbClient.Get(context.Background(), client.ObjectKey{
+		Namespace: namespace,
 		Name:      backup.Spec.StorageLocation,
 	}, location); err != nil {
 		return backupInfo{}, errors.WithStack(err)
@@ -469,6 +478,21 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 	restoreWarnings, restoreErrors := r.restorer.RestoreWithResolvers(restoreReq, actionsResolver, snapshotItemResolver,
 		pluginManager)
 
+	// Iterate over restore item operations and update progress.
+	// Any errors on operations at this point should be added to restore errors.
+	// If any operations are still not complete, then restore will not be set to
+	// Completed yet.
+	inProgressOperations, _, opsCompleted, opsFailed, errs := getRestoreItemOperationProgress(restoreReq.Restore, pluginManager, *restoreReq.GetItemOperationsList())
+	if len(errs) > 0 {
+		for err := range errs {
+			restoreLog.Error(err)
+		}
+	}
+
+	restore.Status.RestoreItemOperationsAttempted = len(*restoreReq.GetItemOperationsList())
+	restore.Status.RestoreItemOperationsCompleted = opsCompleted
+	restore.Status.RestoreItemOperationsFailed = opsFailed
+
 	// log errors and warnings to the restore log
 	for _, msg := range restoreErrors.Velero {
 		restoreLog.Errorf("Velero restore error: %v", msg)
@@ -537,6 +561,29 @@ func (r *restoreReconciler) runValidatedRestore(restore *api.Restore, info backu
 		r.logger.WithError(err).Error("Error uploading restored resource list to backup storage")
 	}
 
+	if err := putOperationsForRestore(restore, *restoreReq.GetItemOperationsList(), backupStore); err != nil {
+		r.logger.WithError(err).Error("Error uploading restore item action operation resource list to backup storage")
+	}
+
+	if restore.Status.Errors > 0 {
+		if inProgressOperations {
+			r.logger.Debug("Restore WaitingForPluginOperationsPartiallyFailed")
+			restore.Status.Phase = api.RestorePhaseWaitingForPluginOperationsPartiallyFailed
+		} else {
+			r.logger.Debug("Restore partially failed")
+			restore.Status.Phase = api.RestorePhasePartiallyFailed
+			r.metrics.RegisterRestorePartialFailure(restore.Spec.ScheduleName)
+		}
+	} else {
+		if inProgressOperations {
+			r.logger.Debug("Restore WaitingForPluginOperations")
+			restore.Status.Phase = api.RestorePhaseWaitingForPluginOperations
+		} else {
+			r.logger.Debug("Restore completed")
+			restore.Status.Phase = api.RestorePhaseCompleted
+			r.metrics.RegisterRestoreSuccess(restore.Spec.ScheduleName)
+		}
+	}
 	return nil
 }
 
@@ -603,6 +650,26 @@ func putRestoredResourceList(restore *api.Restore, list map[string][]string, bac
 	return nil
 }
 
+func putOperationsForRestore(restore *api.Restore, operations []*itemoperation.RestoreOperation, backupStore persistence.BackupStore) error {
+	buf := new(bytes.Buffer)
+	gzw := gzip.NewWriter(buf)
+	defer gzw.Close()
+
+	if err := json.NewEncoder(gzw).Encode(operations); err != nil {
+		return errors.Wrap(err, "error encoding restore item operations list to JSON")
+	}
+
+	if err := gzw.Close(); err != nil {
+		return errors.Wrap(err, "error closing gzip writer")
+	}
+
+	if err := backupStore.PutRestoreItemOperations(restore.Name, buf); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func downloadToTempFile(backupName string, backupStore persistence.BackupStore, logger logrus.FieldLogger) (*os.File, error) {
 	readCloser, err := backupStore.GetBackupContents(backupName)
 	if err != nil {

pkg/controller/restore_controller_test.go

@@ -111,6 +111,7 @@ func TestFetchBackupInfo(t *testing.T) {
 				NewFakeSingleObjectBackupStoreGetter(backupStore),
 				metrics.NewServerMetrics(),
 				formatFlag,
+				60*time.Minute,
 			)
 
 			if test.backupStoreError == nil {
@@ -193,6 +194,7 @@ func TestProcessQueueItemSkips(t *testing.T) {
 				nil, // backupStoreGetter
 				metrics.NewServerMetrics(),
 				formatFlag,
+				60*time.Minute,
 			)
 
 			_, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: types.NamespacedName{
@@ -422,6 +424,7 @@ func TestRestoreReconcile(t *testing.T) {
 				NewFakeSingleObjectBackupStoreGetter(backupStore),
 				metrics.NewServerMetrics(),
 				formatFlag,
+				60*time.Minute,
 			)
 
 			r.clock = clocktesting.NewFakeClock(now)
@@ -453,6 +456,7 @@ func TestRestoreReconcile(t *testing.T) {
 
 				backupStore.On("PutRestoreResults", test.backup.Name, test.restore.Name, mock.Anything).Return(nil)
 				backupStore.On("PutRestoredResourceList", test.restore.Name, mock.Anything).Return(nil)
+				backupStore.On("PutRestoreItemOperations", mock.Anything, mock.Anything).Return(nil)
 
 				volumeSnapshots := []*volume.Snapshot{
 					{
@@ -586,6 +590,7 @@ func TestValidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 		NewFakeSingleObjectBackupStoreGetter(backupStore),
 		metrics.NewServerMetrics(),
 		formatFlag,
+		60*time.Minute,
 	)
 
 	restore := &velerov1api.Restore{
@@ -746,7 +751,7 @@ func TestMostRecentCompletedBackup(t *testing.T) {
 }
 
 func NewRestore(ns, name, backup, includeNS, includeResource string, phase velerov1api.RestorePhase) *builder.RestoreBuilder {
-	restore := builder.ForRestore(ns, name).Phase(phase).Backup(backup)
+	restore := builder.ForRestore(ns, name).Phase(phase).Backup(backup).ItemOperationTimeout(60 * time.Minute)
 
 	if includeNS != "" {
 		restore = restore.IncludedNamespaces(includeNS)

pkg/controller/restore_operations_controller.go

@@ -0,0 +1,352 @@
+/*
+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"
+	"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/itemoperationmap"
+	"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/kube"
+)
+
+const (
+	defaultRestoreOperationsFrequency = 10 * time.Second
+)
+
+type restoreOperationsReconciler struct {
+	client.Client
+	namespace         string
+	logger            logrus.FieldLogger
+	clock             clocks.WithTickerAndDelayedExecution
+	frequency         time.Duration
+	itemOperationsMap *itemoperationmap.RestoreItemOperationsMap
+	newPluginManager  func(logger logrus.FieldLogger) clientmgmt.Manager
+	backupStoreGetter persistence.ObjectBackupStoreGetter
+	metrics           *metrics.ServerMetrics
+}
+
+func NewRestoreOperationsReconciler(
+	logger logrus.FieldLogger,
+	namespace string,
+	client client.Client,
+	frequency time.Duration,
+	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
+	backupStoreGetter persistence.ObjectBackupStoreGetter,
+	metrics *metrics.ServerMetrics,
+	itemOperationsMap *itemoperationmap.RestoreItemOperationsMap,
+) *restoreOperationsReconciler {
+	abor := &restoreOperationsReconciler{
+		Client:            client,
+		logger:            logger,
+		namespace:         namespace,
+		clock:             clocks.RealClock{},
+		frequency:         frequency,
+		itemOperationsMap: itemOperationsMap,
+		newPluginManager:  newPluginManager,
+		backupStoreGetter: backupStoreGetter,
+		metrics:           metrics,
+	}
+	if abor.frequency <= 0 {
+		abor.frequency = defaultRestoreOperationsFrequency
+	}
+	return abor
+}
+
+func (r *restoreOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	s := kube.NewPeriodicalEnqueueSource(r.logger, mgr.GetClient(), &velerov1api.RestoreList{}, r.frequency, kube.PeriodicalEnqueueSourceOption{})
+	gp := kube.NewGenericEventPredicate(func(object client.Object) bool {
+		restore := object.(*velerov1api.Restore)
+		return (restore.Status.Phase == velerov1api.RestorePhaseWaitingForPluginOperations ||
+			restore.Status.Phase == velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed)
+	})
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&velerov1api.Restore{}, builder.WithPredicates(kube.FalsePredicate{})).
+		Watches(s, nil, builder.WithPredicates(gp)).
+		Complete(r)
+}
+
+// +kubebuilder:rbac:groups=velero.io,resources=restores,verbs=get;list;watch;update
+// +kubebuilder:rbac:groups=velero.io,resources=restores/status,verbs=get
+// +kubebuilder:rbac:groups=velero.io,resources=restorestoragelocations,verbs=get
+
+func (r *restoreOperationsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := r.logger.WithField("restore operations for restore", req.String())
+	log.Debug("restoreOperationsReconciler getting restore")
+
+	original := &velerov1api.Restore{}
+	if err := r.Get(ctx, req.NamespacedName, original); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.WithError(err).Error("restore not found")
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{}, errors.Wrapf(err, "error getting restore %s", req.String())
+	}
+	restore := original.DeepCopy()
+	log.Debugf("restore: %s", restore.Name)
+
+	log = r.logger.WithFields(
+		logrus.Fields{
+			"restore": req.String(),
+		},
+	)
+
+	switch restore.Status.Phase {
+	case velerov1api.RestorePhaseWaitingForPluginOperations, velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed:
+		// only process restores waiting for plugin operations to complete
+	default:
+		log.Debug("Restore has no ongoing plugin operations, skipping")
+		return ctrl.Result{}, nil
+	}
+
+	info, err := r.fetchBackupInfo(restore.Spec.BackupName)
+	if err != nil {
+		log.Warnf("Cannot check progress on Restore operations because backup info is unavailable %s; marking restore PartiallyFailed", err.Error())
+		restore.Status.Phase = velerov1api.RestorePhasePartiallyFailed
+		err2 := r.updateRestoreAndOperationsJSON(ctx, original, restore, nil, &itemoperationmap.OperationsForRestore{ErrsSinceUpdate: []string{err.Error()}}, false, false)
+		if err2 != nil {
+			log.WithError(err2).Error("error updating Restore")
+		}
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup info")
+	}
+
+	if info.location.Spec.AccessMode == velerov1api.BackupStorageLocationAccessModeReadOnly {
+		log.Infof("Cannot check progress on Restore operations because backup storage location %s is currently in read-only mode; marking restore PartiallyFailed", info.location.Name)
+		restore.Status.Phase = velerov1api.RestorePhasePartiallyFailed
+
+		err := r.updateRestoreAndOperationsJSON(ctx, original, restore, nil, &itemoperationmap.OperationsForRestore{ErrsSinceUpdate: []string{"BSL is read-only"}}, false, false)
+		if err != nil {
+			log.WithError(err).Error("error updating Restore")
+		}
+		return ctrl.Result{}, nil
+	}
+
+	pluginManager := r.newPluginManager(r.logger)
+	defer pluginManager.CleanupClients()
+	backupStore, err := r.backupStoreGetter.Get(info.location, pluginManager, r.logger)
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error getting backup store")
+	}
+
+	operations, err := r.itemOperationsMap.GetOperationsForRestore(backupStore, restore.Name)
+	if err != nil {
+		err2 := r.updateRestoreAndOperationsJSON(ctx, original, restore, backupStore, &itemoperationmap.OperationsForRestore{ErrsSinceUpdate: []string{err.Error()}}, false, false)
+		if err2 != nil {
+			return ctrl.Result{}, errors.Wrap(err2, "error updating Restore")
+		}
+		return ctrl.Result{}, errors.Wrap(err, "error getting restore operations")
+	}
+	stillInProgress, changes, opsCompleted, opsFailed, errs := getRestoreItemOperationProgress(restore, pluginManager, operations.Operations)
+	// if len(errs)>0, need to update restore errors and error log
+	operations.ErrsSinceUpdate = append(operations.ErrsSinceUpdate, errs...)
+	restore.Status.Errors += len(operations.ErrsSinceUpdate)
+	completionChanges := false
+	if restore.Status.RestoreItemOperationsCompleted != opsCompleted || restore.Status.RestoreItemOperationsFailed != opsFailed {
+		completionChanges = true
+		restore.Status.RestoreItemOperationsCompleted = opsCompleted
+		restore.Status.RestoreItemOperationsFailed = opsFailed
+	}
+	if changes {
+		operations.ChangesSinceUpdate = true
+	}
+
+	// if stillInProgress is false, restore moves to terminal phase and needs update
+	// if operations.ErrsSinceUpdate is not empty, then restore phase needs to change to
+	// RestorePhaseWaitingForPluginOperationsPartiallyFailed and needs update
+	// If the only changes are incremental progress, then no write is necessary, progress can remain in memory
+	if !stillInProgress {
+		if len(operations.ErrsSinceUpdate) > 0 {
+			restore.Status.Phase = velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed
+		}
+		if restore.Status.Phase == velerov1api.RestorePhaseWaitingForPluginOperations {
+			log.Infof("Marking restore %s completed", restore.Name)
+			restore.Status.Phase = velerov1api.RestorePhaseCompleted
+			r.metrics.RegisterRestoreSuccess(restore.Spec.ScheduleName)
+		} else {
+			log.Infof("Marking restore %s FinalizingPartiallyFailed", restore.Name)
+			restore.Status.Phase = velerov1api.RestorePhasePartiallyFailed
+			r.metrics.RegisterRestorePartialFailure(restore.Spec.ScheduleName)
+		}
+	}
+	err = r.updateRestoreAndOperationsJSON(ctx, original, restore, backupStore, operations, changes, completionChanges)
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "error updating Restore")
+	}
+	return ctrl.Result{}, nil
+}
+
+// fetchBackupInfo checks the backup lister for a backup that matches the given name. If it doesn't
+// find it, it returns an error.
+func (r *restoreOperationsReconciler) fetchBackupInfo(backupName string) (backupInfo, error) {
+	return fetchBackupInfoInternal(r.Client, r.namespace, backupName)
+}
+
+func (r *restoreOperationsReconciler) updateRestoreAndOperationsJSON(
+	ctx context.Context,
+	original, restore *velerov1api.Restore,
+	backupStore persistence.BackupStore,
+	operations *itemoperationmap.OperationsForRestore,
+	changes bool,
+	completionChanges bool) error {
+
+	if len(operations.ErrsSinceUpdate) > 0 {
+		// FIXME: download/upload results
+	}
+	removeIfComplete := true
+	defer func() {
+		// remove local operations list if complete
+		if removeIfComplete && (restore.Status.Phase == velerov1api.RestorePhaseCompleted ||
+			restore.Status.Phase == velerov1api.RestorePhasePartiallyFailed) {
+
+			r.itemOperationsMap.DeleteOperationsForRestore(restore.Name)
+		} else if changes {
+			r.itemOperationsMap.PutOperationsForRestore(operations, restore.Name)
+		}
+	}()
+
+	// update restore and upload progress if errs or complete
+	if len(operations.ErrsSinceUpdate) > 0 ||
+		restore.Status.Phase == velerov1api.RestorePhaseCompleted ||
+		restore.Status.Phase == velerov1api.RestorePhasePartiallyFailed {
+		// update file store
+		if backupStore != nil {
+			if err := r.itemOperationsMap.UploadProgressAndPutOperationsForRestore(backupStore, operations, restore.Name); err != nil {
+				removeIfComplete = false
+				return err
+			}
+		}
+		// update restore
+		err := r.Client.Patch(ctx, restore, client.MergeFrom(original))
+		if err != nil {
+			removeIfComplete = false
+			return errors.Wrapf(err, "error updating Restore %s", restore.Name)
+		}
+	} else if completionChanges {
+		// If restore is still incomplete and no new errors are found but there are some new operations
+		// completed, patch restore to reflect new completion numbers, but don't upload detailed json file
+		err := r.Client.Patch(ctx, restore, client.MergeFrom(original))
+		if err != nil {
+			return errors.Wrapf(err, "error updating Restore %s", restore.Name)
+		}
+	}
+	return nil
+}
+
+func getRestoreItemOperationProgress(
+	restore *velerov1api.Restore,
+	pluginManager clientmgmt.Manager,
+	operationsList []*itemoperation.RestoreOperation) (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 {
+			ria, err := pluginManager.GetRestoreItemActionV2(operation.Spec.RestoreItemAction)
+			if err != nil {
+				operation.Status.Phase = itemoperation.OperationPhaseFailed
+				operation.Status.Error = err.Error()
+				errs = append(errs, err.Error())
+				changes = true
+				failedCount++
+				continue
+			}
+			operationProgress, err := ria.Progress(operation.Spec.OperationID, restore)
+			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(restore.Spec.ItemOperationTimeout.Duration).Before(time.Now()) {
+				_ = ria.Cancel(operation.Spec.OperationID, restore)
+				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/restore_operations_controller_test.go

@@ -0,0 +1,320 @@
+/*
+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/itemoperationmap"
+	"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"
+	riav2mocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks/restoreitemaction/v2"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+var (
+	restorePluginManager = &pluginmocks.Manager{}
+	restoreBackupStore   = &persistencemocks.BackupStore{}
+	ria                  = &riav2mocks.RestoreItemAction{}
+)
+
+func mockRestoreOperationsReconciler(fakeClient kbclient.Client, fakeClock *testclocks.FakeClock, freq time.Duration) *restoreOperationsReconciler {
+	abor := NewRestoreOperationsReconciler(
+		logrus.StandardLogger(),
+		velerov1api.DefaultNamespace,
+		fakeClient,
+		freq,
+		func(logrus.FieldLogger) clientmgmt.Manager { return restorePluginManager },
+		NewFakeSingleObjectBackupStoreGetter(restoreBackupStore),
+		metrics.NewServerMetrics(),
+		itemoperationmap.NewRestoreItemOperationsMap(),
+	)
+	abor.clock = fakeClock
+	return abor
+}
+
+func TestRestoreOperationsReconcile(t *testing.T) {
+	fakeClock := testclocks.NewFakeClock(time.Now())
+	metav1Now := metav1.NewTime(fakeClock.Now())
+
+	defaultBackupLocation := builder.ForBackupStorageLocation(velerov1api.DefaultNamespace, "default").Result()
+
+	tests := []struct {
+		name              string
+		restore           *velerov1api.Restore
+		restoreOperations []*itemoperation.RestoreOperation
+		backup            *velerov1api.Backup
+		backupLocation    *velerov1api.BackupStorageLocation
+		operationComplete bool
+		operationErr      string
+		expectError       bool
+		expectPhase       velerov1api.RestorePhase
+	}{
+		{
+			name: "WaitingForPluginOperations restore with completed operations is Completed",
+			restore: builder.ForRestore(velerov1api.DefaultNamespace, "restore-11").
+				Backup("backup-1").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo-11")).
+				Phase(velerov1api.RestorePhaseWaitingForPluginOperations).Result(),
+			backup:            defaultBackup().StorageLocation("default").Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			expectPhase:       velerov1api.RestorePhaseCompleted,
+			restoreOperations: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-11",
+						RestoreUID:        "foo-11",
+						RestoreItemAction: "foo-11",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-11",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperations restore with incomplete operations is still incomplete",
+			restore: builder.ForRestore(velerov1api.DefaultNamespace, "restore-12").
+				Backup("backup-1").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo-12")).
+				Phase(velerov1api.RestorePhaseWaitingForPluginOperations).Result(),
+			backup:            defaultBackup().StorageLocation("default").Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: false,
+			expectPhase:       velerov1api.RestorePhaseWaitingForPluginOperations,
+			restoreOperations: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-12",
+						RestoreUID:        "foo-12",
+						RestoreItemAction: "foo-12",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-12",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperations restore with completed failed operations is PartiallyFailed",
+			restore: builder.ForRestore(velerov1api.DefaultNamespace, "restore-13").
+				Backup("backup-1").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo-13")).
+				Phase(velerov1api.RestorePhaseWaitingForPluginOperations).Result(),
+			backup:            defaultBackup().StorageLocation("default").Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			operationErr:      "failed",
+			expectPhase:       velerov1api.RestorePhasePartiallyFailed,
+			restoreOperations: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-13",
+						RestoreUID:        "foo-13",
+						RestoreItemAction: "foo-13",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-13",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperationsPartiallyFailed restore with completed operations is PartiallyFailed",
+			restore: builder.ForRestore(velerov1api.DefaultNamespace, "restore-14").
+				Backup("backup-1").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo-14")).
+				Phase(velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed).Result(),
+			backup:            defaultBackup().StorageLocation("default").Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			expectPhase:       velerov1api.RestorePhasePartiallyFailed,
+			restoreOperations: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-14",
+						RestoreUID:        "foo-14",
+						RestoreItemAction: "foo-14",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-14",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperationsPartiallyFailed restore with incomplete operations is still incomplete",
+			restore: builder.ForRestore(velerov1api.DefaultNamespace, "restore-15").
+				Backup("backup-1").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo-15")).
+				Phase(velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed).Result(),
+			backup:            defaultBackup().StorageLocation("default").Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: false,
+			expectPhase:       velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed,
+			restoreOperations: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-15",
+						RestoreUID:        "foo-15",
+						RestoreItemAction: "foo-15",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-15",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+		{
+			name: "WaitingForPluginOperationsPartiallyFailed restore with completed failed operations is PartiallyFailed",
+			restore: builder.ForRestore(velerov1api.DefaultNamespace, "restore-16").
+				Backup("backup-1").
+				ItemOperationTimeout(60 * time.Minute).
+				ObjectMeta(builder.WithUID("foo-16")).
+				Phase(velerov1api.RestorePhaseWaitingForPluginOperationsPartiallyFailed).Result(),
+			backup:            defaultBackup().StorageLocation("default").Result(),
+			backupLocation:    defaultBackupLocation,
+			operationComplete: true,
+			operationErr:      "failed",
+			expectPhase:       velerov1api.RestorePhasePartiallyFailed,
+			restoreOperations: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName:       "restore-16",
+						RestoreUID:        "foo-16",
+						RestoreItemAction: "foo-16",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1",
+						},
+						OperationID: "operation-16",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase:   itemoperation.OperationPhaseInProgress,
+						Created: &metav1Now,
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if test.restore == nil {
+				return
+			}
+
+			initObjs := []runtime.Object{}
+			initObjs = append(initObjs, test.restore)
+			initObjs = append(initObjs, test.backup)
+
+			if test.backupLocation != nil {
+				initObjs = append(initObjs, test.backupLocation)
+			}
+
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, initObjs...)
+			reconciler := mockRestoreOperationsReconciler(fakeClient, fakeClock, defaultRestoreOperationsFrequency)
+			restorePluginManager.On("CleanupClients").Return(nil)
+			restoreBackupStore.On("GetRestoreItemOperations", test.restore.Name).Return(test.restoreOperations, nil)
+			restoreBackupStore.On("PutRestoreItemOperations", mock.Anything, mock.Anything).Return(nil)
+			restoreBackupStore.On("PutRestoreMetadata", mock.Anything, mock.Anything).Return(nil)
+			for _, operation := range test.restoreOperations {
+				ria.On("Progress", operation.Spec.OperationID, mock.Anything).
+					Return(velero.OperationProgress{
+						Completed: test.operationComplete,
+						Err:       test.operationErr,
+					}, nil)
+				restorePluginManager.On("GetRestoreItemActionV2", operation.Spec.RestoreItemAction).Return(ria, nil)
+			}
+
+			_, err := reconciler.Reconcile(context.TODO(), ctrl.Request{NamespacedName: types.NamespacedName{Namespace: test.restore.Namespace, Name: test.restore.Name}})
+			gotErr := err != nil
+			assert.Equal(t, test.expectError, gotErr)
+
+			restoreAfter := velerov1api.Restore{}
+			err = fakeClient.Get(context.TODO(), types.NamespacedName{
+				Namespace: test.restore.Namespace,
+				Name:      test.restore.Name,
+			}, &restoreAfter)
+
+			require.NoError(t, err)
+			assert.Equal(t, test.expectPhase, restoreAfter.Status.Phase)
+		})
+	}
+}

pkg/itemoperation/restore_operation.go

@@ -28,6 +28,21 @@ type RestoreOperation struct {
 	Status OperationStatus `json:"status"`
 }
 
+func (in *RestoreOperation) DeepCopy() *RestoreOperation {
+	if in == nil {
+		return nil
+	}
+	out := new(RestoreOperation)
+	in.DeepCopyInto(out)
+	return out
+}
+
+func (in *RestoreOperation) DeepCopyInto(out *RestoreOperation) {
+	*out = *in
+	in.Spec.DeepCopyInto(&out.Spec)
+	in.Status.DeepCopyInto(&out.Status)
+}
+
 type RestoreOperationSpec struct {
 	// RestoreName is the name of the Velero restore this item operation
 	// is associated with.
@@ -46,3 +61,17 @@ type RestoreOperationSpec struct {
 	// OperationID returned by the RIA plugin
 	OperationID string "json:operationID"
 }
+
+func (in *RestoreOperationSpec) DeepCopy() *RestoreOperationSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(RestoreOperationSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+func (in *RestoreOperationSpec) DeepCopyInto(out *RestoreOperationSpec) {
+	*out = *in
+	in.ResourceIdentifier.DeepCopyInto(&out.ResourceIdentifier)
+}

pkg/itemoperationmap/restore_operation_map.go

@@ -0,0 +1,170 @@
+/*
+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 itemoperationmap
+
+import (
+	"bytes"
+	"sync"
+
+	"github.com/pkg/errors"
+
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
+	"github.com/vmware-tanzu/velero/pkg/persistence"
+	"github.com/vmware-tanzu/velero/pkg/util/encode"
+)
+
+type RestoreItemOperationsMap struct {
+	opsMap  map[string]*OperationsForRestore
+	opsLock sync.Mutex
+}
+
+// Returns a pointer to a new RestoreItemOperationsMap
+func NewRestoreItemOperationsMap() *RestoreItemOperationsMap {
+	return &RestoreItemOperationsMap{opsMap: make(map[string]*OperationsForRestore)}
+}
+
+// returns a deep copy so we can minimize the time the map is locked
+func (m *RestoreItemOperationsMap) GetOperationsForRestore(
+	backupStore persistence.BackupStore,
+	restoreName string) (*OperationsForRestore, error) {
+	var err error
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+
+	operations, ok := m.opsMap[restoreName]
+	if !ok || len(operations.Operations) == 0 {
+		operations = &OperationsForRestore{}
+		operations.Operations, err = backupStore.GetRestoreItemOperations(restoreName)
+		if err == nil {
+			m.opsMap[restoreName] = operations
+		}
+	}
+	return operations.DeepCopy(), err
+}
+
+func (m *RestoreItemOperationsMap) PutOperationsForRestore(
+	operations *OperationsForRestore,
+	restoreName string) {
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+	if operations != nil {
+		m.opsMap[restoreName] = operations
+	}
+}
+
+func (m *RestoreItemOperationsMap) DeleteOperationsForRestore(restoreName string) {
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+	if _, ok := m.opsMap[restoreName]; ok {
+		delete(m.opsMap, restoreName)
+	}
+	return
+}
+
+// UploadProgressAndPutOperationsForRestore will upload the item operations for this restore to
+// the object store and update the map for this restore with the modified operations
+func (m *RestoreItemOperationsMap) UploadProgressAndPutOperationsForRestore(
+	backupStore persistence.BackupStore,
+	operations *OperationsForRestore,
+	restoreName string) error {
+
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+
+	if operations == nil {
+		return errors.New("nil operations passed in")
+	}
+	if err := operations.uploadProgress(backupStore, restoreName); err != nil {
+		return err
+	}
+	m.opsMap[restoreName] = operations
+	return nil
+}
+
+// UpdateForRestore will upload the item operations for this restore to
+// the object store, if it has changes not yet uploaded
+func (m *RestoreItemOperationsMap) UpdateForRestore(backupStore persistence.BackupStore, restoreName string) error {
+	// lock operations map
+	m.opsLock.Lock()
+	defer m.opsLock.Unlock()
+
+	operations, ok := m.opsMap[restoreName]
+	// if operations for this restore 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, restoreName); err != nil {
+		return err
+	}
+	return nil
+}
+
+type OperationsForRestore struct {
+	Operations         []*itemoperation.RestoreOperation
+	ChangesSinceUpdate bool
+	ErrsSinceUpdate    []string
+}
+
+func (in *OperationsForRestore) DeepCopy() *OperationsForRestore {
+	if in == nil {
+		return nil
+	}
+	out := new(OperationsForRestore)
+	in.DeepCopyInto(out)
+	return out
+}
+
+func (in *OperationsForRestore) DeepCopyInto(out *OperationsForRestore) {
+	*out = *in
+	if in.Operations != nil {
+		in, out := &in.Operations, &out.Operations
+		*out = make([]*itemoperation.RestoreOperation, len(*in))
+		for i := range *in {
+			if (*in)[i] != nil {
+				in, out := &(*in)[i], &(*out)[i]
+				*out = new(itemoperation.RestoreOperation)
+				(*in).DeepCopyInto(*out)
+			}
+		}
+	}
+	if in.ErrsSinceUpdate != nil {
+		in, out := &in.ErrsSinceUpdate, &out.ErrsSinceUpdate
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+}
+
+func (o *OperationsForRestore) uploadProgress(backupStore persistence.BackupStore, restoreName string) error {
+	if len(o.Operations) > 0 {
+		var restoreItemOperations *bytes.Buffer
+		restoreItemOperations, errs := encode.EncodeToJSONGzip(o.Operations, "restore item operations list")
+		if errs != nil {
+			return errors.Wrap(errs[0], "error encoding item operations json")
+		}
+		err := backupStore.PutRestoreItemOperations(restoreName, restoreItemOperations)
+		if err != nil {
+			return errors.Wrap(err, "error uploading item operations json")
+		}
+	}
+	o.ChangesSinceUpdate = false
+	o.ErrsSinceUpdate = nil
+	return nil
+}

pkg/persistence/mocks/backup_store.go

@@ -407,13 +407,13 @@ func (_m *BackupStore) PutBackupMetadata(backup string, backupMetadata io.Reader
 	return r0
 }
 
-// PutRestoreItemOperations provides a mock function with given fields: backup, restore, restoreItemOperations
-func (_m *BackupStore) PutRestoreItemOperations(backup string, restore string, restoreItemOperations io.Reader) error {
-	ret := _m.Called(backup, restore, restoreItemOperations)
+// PutRestoreItemOperations provides a mock function with given fields: restore, restoreItemOperations
+func (_m *BackupStore) PutRestoreItemOperations(restore string, restoreItemOperations io.Reader) error {
+	ret := _m.Called(restore, restoreItemOperations)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(string, string, io.Reader) error); ok {
-		r0 = rf(backup, restore, restoreItemOperations)
+	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
+		r0 = rf(restore, restoreItemOperations)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -449,6 +449,20 @@ func (_m *BackupStore) PutRestoreResults(backup string, restore string, results
 	return r0
 }
 
+// PutRestoredResourceList provides a mock function with given fields: restore, results
+func (_m *BackupStore) PutRestoredResourceList(restore string, results io.Reader) error {
+	ret := _m.Called(restore, results)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
+		r0 = rf(restore, results)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 type mockConstructorTestingTNewBackupStore interface {
 	mock.TestingT
 	Cleanup(func())
@@ -463,16 +477,3 @@ func NewBackupStore(t mockConstructorTestingTNewBackupStore) *BackupStore {
 
 	return mock
 }
-
-func (_m *BackupStore) PutRestoredResourceList(restore string, results io.Reader) error {
-	ret := _m.Called(restore, results)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(string, io.Reader) error); ok {
-		r0 = rf(restore, results)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}

pkg/persistence/object_store.go

@@ -80,7 +80,7 @@ type BackupStore interface {
 	PutRestoreLog(backup, restore string, log io.Reader) error
 	PutRestoreResults(backup, restore string, results io.Reader) error
 	PutRestoredResourceList(restore string, results io.Reader) error
-	PutRestoreItemOperations(backup, restore string, restoreItemOperations io.Reader) error
+	PutRestoreItemOperations(restore string, restoreItemOperations io.Reader) error
 	GetRestoreItemOperations(name string) ([]*itemoperation.RestoreOperation, error)
 	DeleteRestore(name string) error
 
@@ -547,7 +547,7 @@ func (s *objectBackupStore) PutRestoredResourceList(restore string, list io.Read
 	return s.objectStore.PutObject(s.bucket, s.layout.getRestoreResourceListKey(restore), list)
 }
 
-func (s *objectBackupStore) PutRestoreItemOperations(backup string, restore string, restoreItemOperations io.Reader) error {
+func (s *objectBackupStore) PutRestoreItemOperations(restore string, restoreItemOperations io.Reader) error {
 	return seekAndPutObject(s.objectStore, s.bucket, s.layout.getRestoreItemOperationsKey(restore), restoreItemOperations)
 }
 

pkg/plugin/clientmgmt/restoreitemaction/v2/restartable_restore_item_action.go

@@ -95,6 +95,11 @@ func (r *RestartableRestoreItemAction) getDelegate() (riav2.RestoreItemAction, e
 	return r.getRestoreItemAction()
 }
 
+// Name returns the plugin's name.
+func (r *RestartableRestoreItemAction) Name() string {
+	return r.Key.Name
+}
+
 // AppliesTo restarts the plugin's process if needed, then delegates the call.
 func (r RestartableRestoreItemAction) AppliesTo() (velero.ResourceSelector, error) {
 	delegate, err := r.getDelegate()
@@ -157,6 +162,11 @@ func NewAdaptedV1RestartableRestoreItemAction(v1Restartable *riav1cli.Restartabl
 	return r
 }
 
+// Name restarts the plugin's name.
+func (r *AdaptedV1RestartableRestoreItemAction) Name() string {
+	return r.V1Restartable.Key.Name
+}
+
 // AppliesTo delegates to the v1 AppliesTo call.
 func (r *AdaptedV1RestartableRestoreItemAction) AppliesTo() (velero.ResourceSelector, error) {
 	return r.V1Restartable.AppliesTo()

pkg/plugin/framework/restoreitemaction/v2/restore_item_action_client.go

@@ -199,3 +199,9 @@ func (c *RestoreItemActionGRPCClient) AreAdditionalItemsReady(additionalItems []
 
 	return res.Ready, nil
 }
+
+// This shouldn't be called on the GRPC client since the RestartableRestoreItemAction won't delegate
+// this method
+func (c *RestoreItemActionGRPCClient) Name() string {
+	return ""
+}

pkg/plugin/framework/restoreitemaction/v2/restore_item_action_server.go

@@ -263,3 +263,9 @@ func restoreResourceIdentifierToProto(id velero.ResourceIdentifier) *proto.Resou
 		Name:      id.Name,
 	}
 }
+
+// This shouldn't be called on the GRPC server since the server won't ever receive this request, as
+// the RestartableRestoreItemAction in Velero won't delegate this to the server
+func (c *RestoreItemActionGRPCServer) Name() string {
+	return ""
+}

pkg/plugin/velero/backupitemaction/v2/backup_item_action.go

@@ -29,7 +29,7 @@ import (
 
 // BackupItemAction is an actor that performs an operation on an individual item being backed up.
 type BackupItemAction interface {
-	// Name returns the name of this BIA. Plugins which implement this interface must defined Name,
+	// Name returns the name of this BIA. Plugins which implement this interface must define Name,
 	// but its content is unimportant, as it won't actually be called via RPC. Velero's plugin infrastructure
 	// will implement this directly rather than delegating to the RPC plugin in order to return the name
 	// that the plugin was registered under. The plugins must implement the method to complete the interface.

pkg/plugin/velero/mocks/restoreitemaction/v2/RestoreItemAction.go

@@ -13,7 +13,7 @@ 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.
 */
-// Code generated by mockery v1.0.0. DO NOT EDIT.
+// Code generated by mockery v2.16.0. DO NOT EDIT.
 
 package v2
 
@@ -51,19 +51,19 @@ func (_m *RestoreItemAction) AppliesTo() (velero.ResourceSelector, error) {
 }
 
 // AreAdditionalItemsReady provides a mock function with given fields: AdditionalItems, restore
-func (_m *RestoreItemAction) AreAdditionalItemsReady(AdditionalItems []velero.ResourceIdentifier, restore *v1.Restore) (bool, error) {
-	ret := _m.Called(AdditionalItems, restore)
+func (_m *RestoreItemAction) AreAdditionalItemsReady(additionalItems []velero.ResourceIdentifier, restore *v1.Restore) (bool, error) {
+	ret := _m.Called(additionalItems, restore)
 
 	var r0 bool
 	if rf, ok := ret.Get(0).(func([]velero.ResourceIdentifier, *v1.Restore) bool); ok {
-		r0 = rf(AdditionalItems, restore)
+		r0 = rf(additionalItems, restore)
 	} else {
 		r0 = ret.Get(0).(bool)
 	}
 
 	var r1 error
 	if rf, ok := ret.Get(1).(func([]velero.ResourceIdentifier, *v1.Restore) error); ok {
-		r1 = rf(AdditionalItems, restore)
+		r1 = rf(additionalItems, restore)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -108,6 +108,20 @@ func (_m *RestoreItemAction) Execute(input *velero.RestoreItemActionExecuteInput
 	return r0, r1
 }
 
+// Name provides a mock function with given fields:
+func (_m *RestoreItemAction) Name() string {
+	ret := _m.Called()
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func() string); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	return r0
+}
+
 // Progress provides a mock function with given fields: operationID, restore
 func (_m *RestoreItemAction) Progress(operationID string, restore *v1.Restore) (velero.OperationProgress, error) {
 	ret := _m.Called(operationID, restore)
@@ -128,3 +142,18 @@ func (_m *RestoreItemAction) Progress(operationID string, restore *v1.Restore) (
 
 	return r0, r1
 }
+
+type mockConstructorTestingTNewRestoreItemAction interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewRestoreItemAction creates a new instance of RestoreItemAction. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewRestoreItemAction(t mockConstructorTestingTNewRestoreItemAction) *RestoreItemAction {
+	mock := &RestoreItemAction{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/plugin/velero/restoreitemaction/v2/restore_item_action.go

@@ -27,6 +27,12 @@ import (
 
 // RestoreItemAction is an actor that performs an operation on an individual item being restored.
 type RestoreItemAction interface {
+	// Name returns the name of this RIA. Plugins which implement this interface must define Name,
+	// but its content is unimportant, as it won't actually be called via RPC. Velero's plugin infrastructure
+	// will implement this directly rather than delegating to the RPC plugin in order to return the name
+	// that the plugin was registered under. The plugins must implement the method to complete the interface.
+	Name() string
+
 	// AppliesTo returns information about which resources this action should be invoked for.
 	// A RestoreItemAction's Execute function will only be invoked on items that match the returned
 	// selector. A zero-valued ResourceSelector matches all resources.
@@ -58,7 +64,7 @@ type RestoreItemAction interface {
 	// slice of AdditionalItems (previously returned by Execute())
 	// are ready. Returns true if all items are ready, and false
 	// otherwise. The second return value is to report errors
-	AreAdditionalItemsReady(AdditionalItems []velero.ResourceIdentifier, restore *api.Restore) (bool, error)
+	AreAdditionalItemsReady(additionalItems []velero.ResourceIdentifier, restore *api.Restore) (bool, error)
 }
 
 func AsyncOperationsNotSupportedError() error {

pkg/restore/request.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/volume"
 )
 
@@ -49,12 +50,13 @@ func resourceKey(obj runtime.Object) string {
 type Request struct {
 	*velerov1api.Restore
 
-	Log              logrus.FieldLogger
-	Backup           *velerov1api.Backup
-	PodVolumeBackups []*velerov1api.PodVolumeBackup
-	VolumeSnapshots  []*volume.Snapshot
-	BackupReader     io.Reader
-	RestoredItems    map[itemKey]restoredItemStatus
+	Log                logrus.FieldLogger
+	Backup             *velerov1api.Backup
+	PodVolumeBackups   []*velerov1api.PodVolumeBackup
+	VolumeSnapshots    []*volume.Snapshot
+	BackupReader       io.Reader
+	RestoredItems      map[itemKey]restoredItemStatus
+	itemOperationsList *[]*itemoperation.RestoreOperation
 }
 
 type restoredItemStatus struct {
@@ -62,6 +64,15 @@ type restoredItemStatus struct {
 	itemExists bool
 }
 
+// GetItemOperationsList returns ItemOperationsList, initializing it if necessary
+func (r *Request) GetItemOperationsList() *[]*itemoperation.RestoreOperation {
+	if r.itemOperationsList == nil {
+		list := []*itemoperation.RestoreOperation{}
+		r.itemOperationsList = &list
+	}
+	return r.itemOperationsList
+}
+
 // RestoredResourceList returns the list of restored resources grouped by the API
 // Version and Kind
 func (r *Request) RestoredResourceList() map[string][]string {

pkg/restore/restore.go

@@ -53,6 +53,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/discovery"
 	"github.com/vmware-tanzu/velero/pkg/features"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
@@ -312,6 +313,7 @@ func (kr *kubernetesRestorer) RestoreWithResolvers(
 		hooksContext:                   hooksCtx,
 		hooksCancelFunc:                hooksCancelFunc,
 		kbClient:                       kr.kbClient,
+		itemOperationsList:             req.GetItemOperationsList(),
 	}
 
 	return restoreCtx.execute()
@@ -357,6 +359,7 @@ type restoreContext struct {
 	hooksContext                   go_context.Context
 	hooksCancelFunc                go_context.CancelFunc
 	kbClient                       crclient.Client
+	itemOperationsList             *[]*itemoperation.RestoreOperation
 }
 
 type resourceClientKey struct {
@@ -1211,6 +1214,30 @@ func (ctx *restoreContext) restoreItem(obj *unstructured.Unstructured, groupReso
 			return warnings, errs, itemExists
 		}
 
+		// If async plugin started async operation, add it to the ItemOperations list
+		if executeOutput.OperationID != "" {
+			resourceIdentifier := velero.ResourceIdentifier{
+				GroupResource: groupResource,
+				Namespace:     namespace,
+				Name:          name,
+			}
+			now := metav1.Now()
+			newOperation := itemoperation.RestoreOperation{
+				Spec: itemoperation.RestoreOperationSpec{
+					RestoreName:        ctx.restore.Name,
+					RestoreUID:         string(ctx.restore.UID),
+					RestoreItemAction:  action.RestoreItemAction.Name(),
+					ResourceIdentifier: resourceIdentifier,
+					OperationID:        executeOutput.OperationID,
+				},
+				Status: itemoperation.OperationStatus{
+					Phase:   itemoperation.OperationPhaseInProgress,
+					Created: &now,
+				},
+			}
+			itemOperList := ctx.itemOperationsList
+			*itemOperList = append(*itemOperList, &newOperation)
+		}
 		if executeOutput.SkipRestore {
 			ctx.log.Infof("Skipping restore of %s: %v because a registered plugin discarded it", obj.GroupVersionKind().Kind, name)
 			return warnings, errs, itemExists

pkg/restore/restore_test.go

@@ -45,6 +45,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/discovery"
+	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
@@ -1212,6 +1213,10 @@ type recordResourcesAction struct {
 	additionalItemsReadyTimeout time.Duration
 }
 
+func (a *recordResourcesAction) Name() string {
+	return ""
+}
+
 func (a *recordResourcesAction) AppliesTo() (velero.ResourceSelector, error) {
 	return a.selector, nil
 }
@@ -1438,8 +1443,9 @@ func TestRestoreActionsRunForCorrectItems(t *testing.T) {
 // pluggableAction is a restore item action that can be plugged with an Execute
 // function body at runtime.
 type pluggableAction struct {
-	selector    velero.ResourceSelector
-	executeFunc func(*velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error)
+	selector     velero.ResourceSelector
+	executeFunc  func(*velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error)
+	progressFunc func(string, *velerov1api.Restore) (velero.OperationProgress, error)
 }
 
 func (a *pluggableAction) Execute(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
@@ -1452,6 +1458,10 @@ func (a *pluggableAction) Execute(input *velero.RestoreItemActionExecuteInput) (
 	return a.executeFunc(input)
 }
 
+func (a *pluggableAction) Name() string {
+	return ""
+}
+
 func (a *pluggableAction) AppliesTo() (velero.ResourceSelector, error) {
 	return a.selector, nil
 }
@@ -1605,6 +1615,174 @@ func TestRestoreActionModifications(t *testing.T) {
 	}
 }
 
+// TestRestoreWithAsyncOperations runs restores which return operationIDs and
+// verifies that the itemoperations are tracked as appropriate. Verification is done by
+// looking at the restore request's itemOperationsList field.
+func TestRestoreWithAsyncOperations(t *testing.T) {
+	// completedOperationAction is a *pluggableAction, whose Execute(...)
+	// method returns an operationID which will always be done when calling Progress.
+	completedOperationAction := &pluggableAction{
+		executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+			obj, ok := input.Item.(*unstructured.Unstructured)
+			if !ok {
+				return nil, errors.Errorf("unexpected type %T", input.Item)
+			}
+			return &velero.RestoreItemActionExecuteOutput{
+				UpdatedItem: obj,
+				OperationID: obj.GetName() + "-1",
+			}, nil
+
+		},
+		progressFunc: func(operationID string, restore *velerov1api.Restore) (velero.OperationProgress, error) {
+			return velero.OperationProgress{
+				Completed:   true,
+				Description: "Done!",
+			}, nil
+		},
+	}
+
+	// incompleteOperationAction is a *pluggableAction, whose Execute(...)
+	// method returns an operationID which will never be done when calling Progress.
+	incompleteOperationAction := &pluggableAction{
+		executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+			obj, ok := input.Item.(*unstructured.Unstructured)
+			if !ok {
+				return nil, errors.Errorf("unexpected type %T", input.Item)
+			}
+			return &velero.RestoreItemActionExecuteOutput{
+				UpdatedItem: obj,
+				OperationID: obj.GetName() + "-1",
+			}, nil
+
+		},
+		progressFunc: func(operationID string, restore *velerov1api.Restore) (velero.OperationProgress, error) {
+			return velero.OperationProgress{
+				Completed:   false,
+				Description: "Working...",
+			}, nil
+		},
+	}
+
+	// noOperationAction is a *pluggableAction, whose Execute(...)
+	// method does not return an operationID.
+	noOperationAction := &pluggableAction{
+		executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+			obj, ok := input.Item.(*unstructured.Unstructured)
+			if !ok {
+				return nil, errors.Errorf("unexpected type %T", input.Item)
+			}
+			return &velero.RestoreItemActionExecuteOutput{
+				UpdatedItem: obj,
+			}, nil
+
+		},
+	}
+
+	tests := []struct {
+		name         string
+		restore      *velerov1api.Restore
+		backup       *velerov1api.Backup
+		apiResources []*test.APIResource
+		tarball      io.Reader
+		actions      []riav2.RestoreItemAction
+		want         []*itemoperation.RestoreOperation
+	}{
+		{
+			name:         "action that starts a short-running process records operation",
+			restore:      defaultRestore().Result(),
+			backup:       defaultBackup().Result(),
+			apiResources: []*test.APIResource{test.Pods()},
+			tarball:      test.NewTarWriter(t).AddItems("pods", builder.ForPod("ns-1", "pod-1").Result()).Done(),
+			actions: []riav2.RestoreItemAction{
+				completedOperationAction,
+			},
+			want: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName: "restore-1",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-1"},
+						OperationID: "pod-1-1",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase: "InProgress",
+					},
+				},
+			},
+		},
+		{
+			name:         "action that starts a long-running process records operation",
+			restore:      defaultRestore().Result(),
+			backup:       defaultBackup().Result(),
+			apiResources: []*test.APIResource{test.Pods()},
+			tarball:      test.NewTarWriter(t).AddItems("pods", builder.ForPod("ns-1", "pod-2").Result()).Done(),
+			actions: []riav2.RestoreItemAction{
+				incompleteOperationAction,
+			},
+			want: []*itemoperation.RestoreOperation{
+				{
+					Spec: itemoperation.RestoreOperationSpec{
+						RestoreName: "restore-1",
+						ResourceIdentifier: velero.ResourceIdentifier{
+							GroupResource: kuberesource.Pods,
+							Namespace:     "ns-1",
+							Name:          "pod-2"},
+						OperationID: "pod-2-1",
+					},
+					Status: itemoperation.OperationStatus{
+						Phase: "InProgress",
+					},
+				},
+			},
+		},
+		{
+			name:         "action that has no operation doesn't record one",
+			restore:      defaultRestore().Result(),
+			backup:       defaultBackup().Result(),
+			apiResources: []*test.APIResource{test.Pods()},
+			tarball:      test.NewTarWriter(t).AddItems("pods", builder.ForPod("ns-1", "pod-3").Result()).Done(),
+			actions: []riav2.RestoreItemAction{
+				noOperationAction,
+			},
+			want: []*itemoperation.RestoreOperation{},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t)
+
+			for _, r := range tc.apiResources {
+				h.AddItems(t, r)
+			}
+
+			data := &Request{
+				Log:              h.log,
+				Restore:          tc.restore,
+				Backup:           tc.backup,
+				PodVolumeBackups: nil,
+				VolumeSnapshots:  nil,
+				BackupReader:     tc.tarball,
+			}
+			warnings, errs := h.restorer.Restore(
+				data,
+				tc.actions,
+				nil, // volume snapshotter getter
+			)
+
+			assertEmptyResults(t, warnings, errs)
+			resultOper := *data.GetItemOperationsList()
+			// set want Created times so it won't fail the assert.Equal test
+			for i, wantOper := range tc.want {
+				wantOper.Status.Created = resultOper[i].Status.Created
+			}
+			assert.Equal(t, tc.want, *data.GetItemOperationsList())
+		})
+	}
+}
+
 // TestRestoreActionAdditionalItems runs restores with restore item actions that return additional items
 // to be restored, and verifies that that the correct set of items is created in the API. Verification is
 // done by looking at the namespaces/names of the items in the API; contents are not checked.

site/content/docs/main/api-types/restore.md

@@ -35,6 +35,10 @@ spec:
   # to restore from. If specified, and BackupName is empty, Velero will
   # restore from the most recent successful backup created from this schedule.
   scheduleName: my-scheduled-backup-name
+  # ItemOperationTimeout specifies the time used to wait for
+  # asynchronous BackupItemAction operations
+  # The default value is 1 hour.
+  itemOperationTimeout: 1h
   # Array of namespaces to include in the restore. If unspecified, all namespaces are included.
   # Optional.
   includedNamespaces:
@@ -185,6 +189,12 @@ status:
   phase: ""
   # An array of any validation errors encountered.
   validationErrors: null
+  # Number of attempted RestoreItemAction operations for this restore.
+  restoreItemOperationsAttempted: 2
+  # Number of RestoreItemAction operations that Velero successfully completed for this restore.
+  restoreItemOperationsCompleted: 1
+  # Number of RestoreItemAction operations that ended in failure for this restore.
+  restoreItemOperationsFailed: 0
   # Number of warnings that were logged by the restore.
   warnings: 2
   # Errors is a count of all error messages that were generated