Merge branch 'main' into data-mover-ms-new-controller-1

changelogs/unreleased/8026-sseago

@@ -0,0 +1 @@
+Created new ItemBlockAction (IBA) plugin type

changelogs/unreleased/8061-Lyndon-Li

@@ -0,0 +1 @@
+Data mover micro service restore according to design #7576

design/backup-performance-improvements.md

@@ -91,13 +91,13 @@ message ItemBlockActionAppliesToResponse {
     ResourceSelector ResourceSelector = 1;
 }
 
-message ItemBlockActionRelatedItemsRequest {
+message ItemBlockActionGetRelatedItemsRequest {
     string plugin = 1;
     bytes item = 2;
     bytes backup = 3;
 }
 
-message ItemBlockActionRelatedItemsResponse {
+message ItemBlockActionGetRelatedItemsResponse {
     repeated generated.ResourceIdentifier relatedItems = 1;
 }
 ```

pkg/cmd/cli/datamover/restore.go

@@ -14,16 +14,37 @@ limitations under the License.
 package datamover
 
 import (
+	"context"
 	"fmt"
+	"os"
 	"strings"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
+	"github.com/vmware-tanzu/velero/internal/credentials"
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
 	"github.com/vmware-tanzu/velero/pkg/buildinfo"
 	"github.com/vmware-tanzu/velero/pkg/client"
+	"github.com/vmware-tanzu/velero/pkg/cmd/util/signals"
+	"github.com/vmware-tanzu/velero/pkg/datamover"
+	"github.com/vmware-tanzu/velero/pkg/datapath"
+	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 	"github.com/vmware-tanzu/velero/pkg/util/logging"
+
+	ctlcache "sigs.k8s.io/controller-runtime/pkg/cache"
+	ctlclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type dataMoverRestoreConfig struct {
@@ -52,7 +73,10 @@ func NewRestoreCommand(f client.Factory) *cobra.Command {
 			logger.Infof("Starting Velero data-mover restore %s (%s)", buildinfo.Version, buildinfo.FormattedGitSHA())
 
 			f.SetBasename(fmt.Sprintf("%s-%s", c.Parent().Name(), c.Name()))
-			s := newdataMoverRestore(logger, config)
+			s, err := newdataMoverRestore(logger, f, config)
+			if err != nil {
+				exitWithMessage(logger, false, "Failed to create data mover restore, %v", err)
+			}
 
 			s.run()
 		},
@@ -74,19 +98,174 @@ func NewRestoreCommand(f client.Factory) *cobra.Command {
 }
 
 type dataMoverRestore struct {
-	logger logrus.FieldLogger
-	config dataMoverRestoreConfig
+	logger      logrus.FieldLogger
+	ctx         context.Context
+	cancelFunc  context.CancelFunc
+	client      ctlclient.Client
+	cache       ctlcache.Cache
+	namespace   string
+	nodeName    string
+	config      dataMoverRestoreConfig
+	kubeClient  kubernetes.Interface
+	dataPathMgr *datapath.Manager
 }
 
-func newdataMoverRestore(logger logrus.FieldLogger, config dataMoverRestoreConfig) *dataMoverRestore {
-	s := &dataMoverRestore{
-		logger: logger,
-		config: config,
+func newdataMoverRestore(logger logrus.FieldLogger, factory client.Factory, config dataMoverRestoreConfig) (*dataMoverRestore, error) {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+
+	clientConfig, err := factory.ClientConfig()
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create client config")
 	}
 
-	return s
+	ctrl.SetLogger(zap.New(zap.UseDevMode(true)))
+
+	scheme := runtime.NewScheme()
+	if err := velerov1api.AddToScheme(scheme); err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to add velero v1 scheme")
+	}
+
+	if err := velerov2alpha1api.AddToScheme(scheme); err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to add velero v2alpha1 scheme")
+	}
+
+	if err := v1.AddToScheme(scheme); err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to add core v1 scheme")
+	}
+
+	nodeName := os.Getenv("NODE_NAME")
+
+	// use a field selector to filter to only pods scheduled on this node.
+	cacheOption := ctlcache.Options{
+		Scheme: scheme,
+		ByObject: map[ctlclient.Object]ctlcache.ByObject{
+			&v1.Pod{}: {
+				Field: fields.Set{"spec.nodeName": nodeName}.AsSelector(),
+			},
+			&velerov2alpha1api.DataDownload{}: {
+				Field: fields.Set{"metadata.namespace": factory.Namespace()}.AsSelector(),
+			},
+		},
+	}
+
+	cli, err := ctlclient.New(clientConfig, ctlclient.Options{
+		Scheme: scheme,
+	})
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create client")
+	}
+
+	cache, err := ctlcache.New(clientConfig, cacheOption)
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create client cache")
+	}
+
+	s := &dataMoverRestore{
+		logger:     logger,
+		ctx:        ctx,
+		cancelFunc: cancelFunc,
+		client:     cli,
+		cache:      cache,
+		config:     config,
+		namespace:  factory.Namespace(),
+		nodeName:   nodeName,
+	}
+
+	s.kubeClient, err = factory.KubeClient()
+	if err != nil {
+		cancelFunc()
+		return nil, errors.Wrap(err, "error to create kube client")
+	}
+
+	s.dataPathMgr = datapath.NewManager(1)
+
+	return s, nil
 }
 
+var funcCreateDataPathRestore = (*dataMoverRestore).createDataPathService
+
 func (s *dataMoverRestore) run() {
+	signals.CancelOnShutdown(s.cancelFunc, s.logger)
+	go func() {
+		if err := s.cache.Start(s.ctx); err != nil {
+			s.logger.WithError(err).Warn("error starting cache")
+		}
+	}()
+
+	// TODOOO: call s.runDataPath()
 	time.Sleep(time.Duration(1<<63 - 1))
 }
+
+func (s *dataMoverRestore) runDataPath() {
+	s.logger.Infof("Starting micro service in node %s for dd %s", s.nodeName, s.config.ddName)
+
+	dpService, err := funcCreateDataPathRestore(s)
+	if err != nil {
+		s.cancelFunc()
+		funcExitWithMessage(s.logger, false, "Failed to create data path service for DataDownload %s: %v", s.config.ddName, err)
+		return
+	}
+
+	s.logger.Infof("Starting data path service %s", s.config.ddName)
+
+	err = dpService.Init()
+	if err != nil {
+		s.cancelFunc()
+		funcExitWithMessage(s.logger, false, "Failed to init data path service for DataDownload %s: %v", s.config.ddName, err)
+		return
+	}
+
+	result, err := dpService.RunCancelableDataPath(s.ctx)
+	if err != nil {
+		s.cancelFunc()
+		funcExitWithMessage(s.logger, false, "Failed to run data path service for DataDownload %s: %v", s.config.ddName, err)
+		return
+	}
+
+	s.logger.WithField("dd", s.config.ddName).Info("Data path service completed")
+
+	dpService.Shutdown()
+
+	s.logger.WithField("dd", s.config.ddName).Info("Data path service is shut down")
+
+	s.cancelFunc()
+
+	funcExitWithMessage(s.logger, true, result)
+}
+
+func (s *dataMoverRestore) createDataPathService() (dataPathService, error) {
+	credentialFileStore, err := funcNewCredentialFileStore(
+		s.client,
+		s.namespace,
+		defaultCredentialsDirectory,
+		filesystem.NewFileSystem(),
+	)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error to create credential file store")
+	}
+
+	credSecretStore, err := funcNewCredentialSecretStore(s.client, s.namespace)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error to create credential secret store")
+	}
+
+	credGetter := &credentials.CredentialGetter{FromFile: credentialFileStore, FromSecret: credSecretStore}
+
+	duInformer, err := s.cache.GetInformer(s.ctx, &velerov2alpha1api.DataDownload{})
+	if err != nil {
+		return nil, errors.Wrap(err, "error to get controller-runtime informer from manager")
+	}
+
+	repoEnsurer := repository.NewEnsurer(s.client, s.logger, s.config.resourceTimeout)
+
+	return datamover.NewRestoreMicroService(s.ctx, s.client, s.kubeClient, s.config.ddName, s.namespace, s.nodeName, datapath.AccessPoint{
+		ByPath:  s.config.volumePath,
+		VolMode: uploader.PersistentVolumeMode(s.config.volumeMode),
+	}, s.dataPathMgr, repoEnsurer, credGetter, duInformer, s.logger), nil
+}

pkg/cmd/cli/datamover/restore_test.go

@@ -0,0 +1,166 @@
+/*
+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 datamover
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	cacheMock "github.com/vmware-tanzu/velero/pkg/cmd/cli/datamover/mocks"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func fakeCreateDataPathRestoreWithErr(_ *dataMoverRestore) (dataPathService, error) {
+	return nil, errors.New("fake-create-data-path-error")
+}
+
+func fakeCreateDataPathRestore(_ *dataMoverRestore) (dataPathService, error) {
+	return frHelper, nil
+}
+
+func TestRunDataPathRestore(t *testing.T) {
+	tests := []struct {
+		name                        string
+		ddName                      string
+		createDataPathFail          bool
+		initDataPathErr             error
+		runCancelableDataPathErr    error
+		runCancelableDataPathResult string
+		expectedMessage             string
+		expectedSucceed             bool
+	}{
+		{
+			name:               "create data path failed",
+			ddName:             "fake-name",
+			createDataPathFail: true,
+			expectedMessage:    "Failed to create data path service for DataDownload fake-name: fake-create-data-path-error",
+		},
+		{
+			name:            "init data path failed",
+			ddName:          "fake-name",
+			initDataPathErr: errors.New("fake-init-data-path-error"),
+			expectedMessage: "Failed to init data path service for DataDownload fake-name: fake-init-data-path-error",
+		},
+		{
+			name:                     "run data path failed",
+			ddName:                   "fake-name",
+			runCancelableDataPathErr: errors.New("fake-run-data-path-error"),
+			expectedMessage:          "Failed to run data path service for DataDownload fake-name: fake-run-data-path-error",
+		},
+		{
+			name:                        "succeed",
+			ddName:                      "fake-name",
+			runCancelableDataPathResult: "fake-run-data-path-result",
+			expectedMessage:             "fake-run-data-path-result",
+			expectedSucceed:             true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			frHelper = &fakeRunHelper{
+				initErr:                     test.initDataPathErr,
+				runCancelableDataPathErr:    test.runCancelableDataPathErr,
+				runCancelableDataPathResult: test.runCancelableDataPathResult,
+			}
+
+			if test.createDataPathFail {
+				funcCreateDataPathRestore = fakeCreateDataPathRestoreWithErr
+			} else {
+				funcCreateDataPathRestore = fakeCreateDataPathRestore
+			}
+
+			funcExitWithMessage = frHelper.ExitWithMessage
+
+			s := &dataMoverRestore{
+				logger:     velerotest.NewLogger(),
+				cancelFunc: func() {},
+				config: dataMoverRestoreConfig{
+					ddName: test.ddName,
+				},
+			}
+
+			s.runDataPath()
+
+			assert.Equal(t, test.expectedMessage, frHelper.exitMessage)
+			assert.Equal(t, test.expectedSucceed, frHelper.succeed)
+		})
+	}
+}
+
+func TestCreateDataPathRestore(t *testing.T) {
+	tests := []struct {
+		name            string
+		fileStoreErr    error
+		secretStoreErr  error
+		mockGetInformer bool
+		getInformerErr  error
+		expectedError   string
+	}{
+		{
+			name:          "create credential file store error",
+			fileStoreErr:  errors.New("fake-file-store-error"),
+			expectedError: "error to create credential file store: fake-file-store-error",
+		},
+		{
+			name:           "create credential secret store",
+			secretStoreErr: errors.New("fake-secret-store-error"),
+			expectedError:  "error to create credential secret store: fake-secret-store-error",
+		},
+		{
+			name:            "get informer error",
+			mockGetInformer: true,
+			getInformerErr:  errors.New("fake-get-informer-error"),
+			expectedError:   "error to get controller-runtime informer from manager: fake-get-informer-error",
+		},
+		{
+			name:            "succeed",
+			mockGetInformer: true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fcHelper := &fakeCreateDataPathServiceHelper{
+				fileStoreErr:   test.fileStoreErr,
+				secretStoreErr: test.secretStoreErr,
+			}
+
+			funcNewCredentialFileStore = fcHelper.NewNamespacedFileStore
+			funcNewCredentialSecretStore = fcHelper.NewNamespacedSecretStore
+
+			cache := cacheMock.NewCache(t)
+			if test.mockGetInformer {
+				cache.On("GetInformer", mock.Anything, mock.Anything).Return(nil, test.getInformerErr)
+			}
+
+			funcExitWithMessage = frHelper.ExitWithMessage
+
+			s := &dataMoverRestore{
+				cache: cache,
+			}
+
+			_, err := s.createDataPathService()
+
+			if test.expectedError != "" {
+				assert.EqualError(t, err, test.expectedError)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

pkg/controller/backup_deletion_controller.go

@@ -248,7 +248,11 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 
 	backupStore, err := r.backupStoreGetter.Get(location, pluginManager, log)
 	if err != nil {
-		return ctrl.Result{}, errors.Wrap(err, "error getting the backup store")
+		_, patchErr := r.patchDeleteBackupRequest(ctx, dbr, func(r *velerov1api.DeleteBackupRequest) {
+			r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
+			r.Status.Errors = append(r.Status.Errors, fmt.Sprintf("cannot delete backup because backup storage location %s is currently unavailable, error: %s", location.Name, err.Error()))
+		})
+		return ctrl.Result{}, patchErr
 	}
 
 	actions, err := pluginManager.GetDeleteItemActions()

pkg/controller/backup_deletion_controller_test.go

@@ -125,8 +125,13 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
 		td := setupBackupDeletionControllerTest(t, defaultTestDbr(), location, backup)
 		td.controller.backupStoreGetter = &fakeErrorBackupStoreGetter{}
 		_, err := td.controller.Reconcile(ctx, td.req)
-		assert.Error(t, err)
-		assert.True(t, strings.HasPrefix(err.Error(), "error getting the backup store"))
+		require.NoError(t, err)
+		res := &velerov1api.DeleteBackupRequest{}
+		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
+		require.NoError(t, err)
+		assert.Equal(t, "Processed", string(res.Status.Phase))
+		assert.Len(t, res.Status.Errors, 1)
+		assert.True(t, strings.HasPrefix(res.Status.Errors[0], fmt.Sprintf("cannot delete backup because backup storage location %s is currently unavailable", location.Name)))
 	})
 
 	t.Run("missing spec.backupName", func(t *testing.T) {

pkg/datamover/backup_micro_service.go

@@ -127,7 +127,7 @@ func (r *BackupMicroService) Init() error {
 	return err
 }
 
-var waitDuTimeout time.Duration = time.Minute * 2
+var waitControllerTimeout time.Duration = time.Minute * 2
 
 func (r *BackupMicroService) RunCancelableDataPath(ctx context.Context) (string, error) {
 	log := r.logger.WithFields(logrus.Fields{
@@ -135,7 +135,7 @@ func (r *BackupMicroService) RunCancelableDataPath(ctx context.Context) (string,
 	})
 
 	du := &velerov2alpha1api.DataUpload{}
-	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, waitDuTimeout, true, func(ctx context.Context) (bool, error) {
+	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, waitControllerTimeout, true, func(ctx context.Context) (bool, error) {
 		err := r.client.Get(ctx, types.NamespacedName{
 			Namespace: r.namespace,
 			Name:      r.dataUploadName,
@@ -313,7 +313,7 @@ func (r *BackupMicroService) closeDataPath(ctx context.Context, duName string) {
 func (r *BackupMicroService) cancelDataUpload(du *velerov2alpha1api.DataUpload) {
 	r.logger.WithField("DataUpload", du.Name).Info("Data upload is being canceled")
 
-	r.eventRecorder.Event(du, false, "Canceling", "Canceing for data upload %s", du.Name)
+	r.eventRecorder.Event(du, false, "Canceling", "Canceling for data upload %s", du.Name)
 
 	fsBackup := r.dataPathMgr.GetAsyncBR(du.Name)
 	if fsBackup == nil {

pkg/datamover/backup_micro_service_test.go

@@ -412,7 +412,7 @@ func TestRunCancelableDataPath(t *testing.T) {
 				return fsBR
 			}
 
-			waitDuTimeout = time.Second
+			waitControllerTimeout = time.Second
 
 			if test.result != nil {
 				go func() {

pkg/datamover/restore_micro_service.go

@@ -0,0 +1,295 @@
+/*
+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 datamover
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/vmware-tanzu/velero/internal/credentials"
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/datapath"
+	"github.com/vmware-tanzu/velero/pkg/repository"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
+
+	cachetool "k8s.io/client-go/tools/cache"
+)
+
+// RestoreMicroService process data mover restores inside the restore pod
+type RestoreMicroService struct {
+	ctx              context.Context
+	client           client.Client
+	kubeClient       kubernetes.Interface
+	repoEnsurer      *repository.Ensurer
+	credentialGetter *credentials.CredentialGetter
+	logger           logrus.FieldLogger
+	dataPathMgr      *datapath.Manager
+	eventRecorder    kube.EventRecorder
+
+	namespace        string
+	dataDownloadName string
+	dataDownload     *velerov2alpha1api.DataDownload
+	sourceTargetPath datapath.AccessPoint
+
+	resultSignal chan dataPathResult
+
+	ddInformer cache.Informer
+	ddHandler  cachetool.ResourceEventHandlerRegistration
+	nodeName   string
+}
+
+func NewRestoreMicroService(ctx context.Context, client client.Client, kubeClient kubernetes.Interface, dataDownloadName string, namespace string, nodeName string,
+	sourceTargetPath datapath.AccessPoint, dataPathMgr *datapath.Manager, repoEnsurer *repository.Ensurer, cred *credentials.CredentialGetter,
+	ddInformer cache.Informer, log logrus.FieldLogger) *RestoreMicroService {
+	return &RestoreMicroService{
+		ctx:              ctx,
+		client:           client,
+		kubeClient:       kubeClient,
+		credentialGetter: cred,
+		logger:           log,
+		repoEnsurer:      repoEnsurer,
+		dataPathMgr:      dataPathMgr,
+		namespace:        namespace,
+		dataDownloadName: dataDownloadName,
+		sourceTargetPath: sourceTargetPath,
+		nodeName:         nodeName,
+		resultSignal:     make(chan dataPathResult),
+		ddInformer:       ddInformer,
+	}
+}
+
+func (r *RestoreMicroService) Init() error {
+	r.eventRecorder = kube.NewEventRecorder(r.kubeClient, r.client.Scheme(), r.dataDownloadName, r.nodeName)
+
+	handler, err := r.ddInformer.AddEventHandler(
+		cachetool.ResourceEventHandlerFuncs{
+			UpdateFunc: func(oldObj interface{}, newObj interface{}) {
+				oldDd := oldObj.(*velerov2alpha1api.DataDownload)
+				newDd := newObj.(*velerov2alpha1api.DataDownload)
+
+				if newDd.Name != r.dataDownloadName {
+					return
+				}
+
+				if newDd.Status.Phase != velerov2alpha1api.DataDownloadPhaseInProgress {
+					return
+				}
+
+				if newDd.Spec.Cancel && !oldDd.Spec.Cancel {
+					r.cancelDataDownload(newDd)
+				}
+			},
+		},
+	)
+
+	if err != nil {
+		return errors.Wrap(err, "error adding dd handler")
+	}
+
+	r.ddHandler = handler
+
+	return err
+}
+
+func (r *RestoreMicroService) RunCancelableDataPath(ctx context.Context) (string, error) {
+	log := r.logger.WithFields(logrus.Fields{
+		"datadownload": r.dataDownloadName,
+	})
+
+	dd := &velerov2alpha1api.DataDownload{}
+	err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, waitControllerTimeout, true, func(ctx context.Context) (bool, error) {
+		err := r.client.Get(ctx, types.NamespacedName{
+			Namespace: r.namespace,
+			Name:      r.dataDownloadName,
+		}, dd)
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+
+		if err != nil {
+			return true, errors.Wrapf(err, "error to get dd %s", r.dataDownloadName)
+		}
+
+		if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseInProgress {
+			return true, nil
+		} else {
+			return false, nil
+		}
+	})
+	if err != nil {
+		log.WithError(err).Error("Failed to wait dd")
+		return "", errors.Wrap(err, "error waiting for dd")
+	}
+
+	r.dataDownload = dd
+
+	log.Info("Run cancelable dataDownload")
+
+	callbacks := datapath.Callbacks{
+		OnCompleted: r.OnDataDownloadCompleted,
+		OnFailed:    r.OnDataDownloadFailed,
+		OnCancelled: r.OnDataDownloadCancelled,
+		OnProgress:  r.OnDataDownloadProgress,
+	}
+
+	fsRestore, err := r.dataPathMgr.CreateFileSystemBR(dd.Name, dataUploadDownloadRequestor, ctx, r.client, dd.Namespace, callbacks, log)
+	if err != nil {
+		return "", errors.Wrap(err, "error to create data path")
+	}
+
+	log.Debug("Found volume path")
+	if err := fsRestore.Init(ctx,
+		&datapath.FSBRInitParam{
+			BSLName:           dd.Spec.BackupStorageLocation,
+			SourceNamespace:   dd.Spec.SourceNamespace,
+			UploaderType:      GetUploaderType(dd.Spec.DataMover),
+			RepositoryType:    velerov1api.BackupRepositoryTypeKopia,
+			RepoIdentifier:    "",
+			RepositoryEnsurer: r.repoEnsurer,
+			CredentialGetter:  r.credentialGetter,
+		}); err != nil {
+		return "", errors.Wrap(err, "error to initialize data path")
+	}
+	log.Info("fs init")
+
+	if err := fsRestore.StartRestore(dd.Spec.SnapshotID, r.sourceTargetPath, dd.Spec.DataMoverConfig); err != nil {
+		return "", errors.Wrap(err, "error starting data path restore")
+	}
+
+	log.Info("Async fs restore data path started")
+	r.eventRecorder.Event(dd, false, datapath.EventReasonStarted, "Data path for %s started", dd.Name)
+
+	result := ""
+	select {
+	case <-ctx.Done():
+		err = errors.New("timed out waiting for fs restore to complete")
+		break
+	case res := <-r.resultSignal:
+		err = res.err
+		result = res.result
+		break
+	}
+
+	if err != nil {
+		log.WithError(err).Error("Async fs restore was not completed")
+	}
+
+	return result, err
+}
+
+func (r *RestoreMicroService) Shutdown() {
+	r.eventRecorder.Shutdown()
+	r.closeDataPath(r.ctx, r.dataDownloadName)
+
+	if r.ddHandler != nil {
+		if err := r.ddInformer.RemoveEventHandler(r.ddHandler); err != nil {
+			r.logger.WithError(err).Warn("Failed to remove pod handler")
+		}
+	}
+}
+
+func (r *RestoreMicroService) OnDataDownloadCompleted(ctx context.Context, namespace string, ddName string, result datapath.Result) {
+	defer r.closeDataPath(ctx, ddName)
+
+	log := r.logger.WithField("datadownload", ddName)
+
+	restoreBytes, err := funcMarshal(result.Restore)
+	if err != nil {
+		log.WithError(err).Errorf("Failed to marshal restore result %v", result.Restore)
+		r.resultSignal <- dataPathResult{
+			err: errors.Wrapf(err, "Failed to marshal restore result %v", result.Restore),
+		}
+	} else {
+		r.eventRecorder.Event(r.dataDownload, false, datapath.EventReasonCompleted, string(restoreBytes))
+		r.resultSignal <- dataPathResult{
+			result: string(restoreBytes),
+		}
+	}
+
+	log.Info("Async fs restore data path completed")
+}
+
+func (r *RestoreMicroService) OnDataDownloadFailed(ctx context.Context, namespace string, ddName string, err error) {
+	defer r.closeDataPath(ctx, ddName)
+
+	log := r.logger.WithField("datadownload", ddName)
+	log.WithError(err).Error("Async fs restore data path failed")
+
+	r.eventRecorder.Event(r.dataDownload, false, datapath.EventReasonFailed, "Data path for data download %s failed, error %v", r.dataDownloadName, err)
+	r.resultSignal <- dataPathResult{
+		err: errors.Wrapf(err, "Data path for data download %s failed", r.dataDownloadName),
+	}
+}
+
+func (r *RestoreMicroService) OnDataDownloadCancelled(ctx context.Context, namespace string, ddName string) {
+	defer r.closeDataPath(ctx, ddName)
+
+	log := r.logger.WithField("datadownload", ddName)
+	log.Warn("Async fs restore data path canceled")
+
+	r.eventRecorder.Event(r.dataDownload, false, datapath.EventReasonCancelled, "Data path for data download %s canceled", ddName)
+	r.resultSignal <- dataPathResult{
+		err: errors.New(datapath.ErrCancelled),
+	}
+}
+
+func (r *RestoreMicroService) OnDataDownloadProgress(ctx context.Context, namespace string, ddName string, progress *uploader.Progress) {
+	log := r.logger.WithFields(logrus.Fields{
+		"datadownload": ddName,
+	})
+
+	progressBytes, err := funcMarshal(progress)
+	if err != nil {
+		log.WithError(err).Errorf("Failed to marshal progress %v", progress)
+		return
+	}
+
+	r.eventRecorder.Event(r.dataDownload, false, datapath.EventReasonProgress, string(progressBytes))
+}
+
+func (r *RestoreMicroService) closeDataPath(ctx context.Context, ddName string) {
+	fsRestore := r.dataPathMgr.GetAsyncBR(ddName)
+	if fsRestore != nil {
+		fsRestore.Close(ctx)
+	}
+
+	r.dataPathMgr.RemoveAsyncBR(ddName)
+}
+
+func (r *RestoreMicroService) cancelDataDownload(dd *velerov2alpha1api.DataDownload) {
+	r.logger.WithField("DataDownload", dd.Name).Info("Data download is being canceled")
+
+	r.eventRecorder.Event(dd, false, "Canceling", "Canceling for data download %s", dd.Name)
+
+	fsBackup := r.dataPathMgr.GetAsyncBR(dd.Name)
+	if fsBackup == nil {
+		r.OnDataDownloadCancelled(r.ctx, dd.GetNamespace(), dd.GetName())
+	} else {
+		fsBackup.Cancel()
+	}
+}

pkg/datamover/restore_micro_service_test.go

@@ -0,0 +1,394 @@
+/*
+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 datamover
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"k8s.io/apimachinery/pkg/runtime"
+	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+	clientFake "sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/datapath"
+	datapathmockes "github.com/vmware-tanzu/velero/pkg/datapath/mocks"
+	"github.com/vmware-tanzu/velero/pkg/uploader"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func TestOnDataDownloadFailed(t *testing.T) {
+	dataDownloadName := "fake-data-download"
+	bt := &backupMsTestHelper{}
+
+	bs := &RestoreMicroService{
+		dataDownloadName: dataDownloadName,
+		dataPathMgr:      datapath.NewManager(1),
+		eventRecorder:    bt,
+		resultSignal:     make(chan dataPathResult),
+		logger:           velerotest.NewLogger(),
+	}
+
+	expectedErr := "Data path for data download fake-data-download failed: fake-error"
+	expectedEventReason := datapath.EventReasonFailed
+	expectedEventMsg := "Data path for data download fake-data-download failed, error fake-error"
+
+	go bs.OnDataDownloadFailed(context.TODO(), velerov1api.DefaultNamespace, dataDownloadName, errors.New("fake-error"))
+
+	result := <-bs.resultSignal
+	assert.EqualError(t, result.err, expectedErr)
+	assert.Equal(t, expectedEventReason, bt.EventReason())
+	assert.Equal(t, expectedEventMsg, bt.EventMessage())
+}
+
+func TestOnDataDownloadCancelled(t *testing.T) {
+	dataDownloadName := "fake-data-download"
+	bt := &backupMsTestHelper{}
+
+	bs := &RestoreMicroService{
+		dataDownloadName: dataDownloadName,
+		dataPathMgr:      datapath.NewManager(1),
+		eventRecorder:    bt,
+		resultSignal:     make(chan dataPathResult),
+		logger:           velerotest.NewLogger(),
+	}
+
+	expectedErr := datapath.ErrCancelled
+	expectedEventReason := datapath.EventReasonCancelled
+	expectedEventMsg := "Data path for data download fake-data-download canceled"
+
+	go bs.OnDataDownloadCancelled(context.TODO(), velerov1api.DefaultNamespace, dataDownloadName)
+
+	result := <-bs.resultSignal
+	assert.EqualError(t, result.err, expectedErr)
+	assert.Equal(t, expectedEventReason, bt.EventReason())
+	assert.Equal(t, expectedEventMsg, bt.EventMessage())
+}
+
+func TestOnDataDownloadCompleted(t *testing.T) {
+	tests := []struct {
+		name                string
+		expectedErr         string
+		expectedEventReason string
+		expectedEventMsg    string
+		marshalErr          error
+		marshallStr         string
+	}{
+		{
+			name:        "marshal fail",
+			marshalErr:  errors.New("fake-marshal-error"),
+			expectedErr: "Failed to marshal restore result {{ }}: fake-marshal-error",
+		},
+		{
+			name:                "succeed",
+			marshallStr:         "fake-complete-string",
+			expectedEventReason: datapath.EventReasonCompleted,
+			expectedEventMsg:    "fake-complete-string",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			dataDownloadName := "fake-data-download"
+
+			bt := &backupMsTestHelper{
+				marshalErr:   test.marshalErr,
+				marshalBytes: []byte(test.marshallStr),
+			}
+
+			bs := &RestoreMicroService{
+				dataPathMgr:   datapath.NewManager(1),
+				eventRecorder: bt,
+				resultSignal:  make(chan dataPathResult),
+				logger:        velerotest.NewLogger(),
+			}
+
+			funcMarshal = bt.Marshal
+
+			go bs.OnDataDownloadCompleted(context.TODO(), velerov1api.DefaultNamespace, dataDownloadName, datapath.Result{})
+
+			result := <-bs.resultSignal
+			if test.marshalErr != nil {
+				assert.EqualError(t, result.err, test.expectedErr)
+			} else {
+				assert.NoError(t, result.err)
+				assert.Equal(t, test.expectedEventReason, bt.EventReason())
+				assert.Equal(t, test.expectedEventMsg, bt.EventMessage())
+			}
+		})
+	}
+}
+
+func TestOnDataDownloadProgress(t *testing.T) {
+	tests := []struct {
+		name                string
+		expectedEventReason string
+		expectedEventMsg    string
+		marshalErr          error
+		marshallStr         string
+	}{
+		{
+			name:       "marshal fail",
+			marshalErr: errors.New("fake-marshal-error"),
+		},
+		{
+			name:                "succeed",
+			marshallStr:         "fake-progress-string",
+			expectedEventReason: datapath.EventReasonProgress,
+			expectedEventMsg:    "fake-progress-string",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			dataDownloadName := "fake-data-download"
+
+			bt := &backupMsTestHelper{
+				marshalErr:   test.marshalErr,
+				marshalBytes: []byte(test.marshallStr),
+			}
+
+			bs := &RestoreMicroService{
+				dataPathMgr:   datapath.NewManager(1),
+				eventRecorder: bt,
+				logger:        velerotest.NewLogger(),
+			}
+
+			funcMarshal = bt.Marshal
+
+			bs.OnDataDownloadProgress(context.TODO(), velerov1api.DefaultNamespace, dataDownloadName, &uploader.Progress{})
+
+			if test.marshalErr != nil {
+				assert.False(t, bt.withEvent)
+			} else {
+				assert.True(t, bt.withEvent)
+				assert.Equal(t, test.expectedEventReason, bt.EventReason())
+				assert.Equal(t, test.expectedEventMsg, bt.EventMessage())
+			}
+		})
+	}
+}
+
+func TestCancelDataDownload(t *testing.T) {
+	tests := []struct {
+		name                string
+		expectedEventReason string
+		expectedEventMsg    string
+		expectedErr         string
+	}{
+		{
+			name:                "no fs restore",
+			expectedEventReason: datapath.EventReasonCancelled,
+			expectedEventMsg:    "Data path for data download fake-data-download canceled",
+			expectedErr:         datapath.ErrCancelled,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			dataDownloadName := "fake-data-download"
+			dd := builder.ForDataDownload(velerov1api.DefaultNamespace, dataDownloadName).Result()
+
+			bt := &backupMsTestHelper{}
+
+			bs := &RestoreMicroService{
+				dataPathMgr:   datapath.NewManager(1),
+				eventRecorder: bt,
+				resultSignal:  make(chan dataPathResult),
+				logger:        velerotest.NewLogger(),
+			}
+
+			go bs.cancelDataDownload(dd)
+
+			result := <-bs.resultSignal
+
+			assert.EqualError(t, result.err, test.expectedErr)
+			assert.True(t, bt.withEvent)
+			assert.Equal(t, test.expectedEventReason, bt.EventReason())
+			assert.Equal(t, test.expectedEventMsg, bt.EventMessage())
+		})
+	}
+}
+
+func TestRunCancelableRestore(t *testing.T) {
+	dataDownloadName := "fake-data-download"
+	dd := builder.ForDataDownload(velerov1api.DefaultNamespace, dataDownloadName).Phase(velerov2alpha1api.DataDownloadPhaseNew).Result()
+	ddInProgress := builder.ForDataDownload(velerov1api.DefaultNamespace, dataDownloadName).Phase(velerov2alpha1api.DataDownloadPhaseInProgress).Result()
+	ctxTimeout, cancel := context.WithTimeout(context.Background(), time.Second)
+
+	tests := []struct {
+		name             string
+		ctx              context.Context
+		result           *dataPathResult
+		dataPathMgr      *datapath.Manager
+		kubeClientObj    []runtime.Object
+		initErr          error
+		startErr         error
+		dataPathStarted  bool
+		expectedEventMsg string
+		expectedErr      string
+	}{
+		{
+			name:        "no dd",
+			ctx:         context.Background(),
+			expectedErr: "error waiting for dd: context deadline exceeded",
+		},
+		{
+			name:          "dd not in in-progress",
+			ctx:           context.Background(),
+			kubeClientObj: []runtime.Object{dd},
+			expectedErr:   "error waiting for dd: context deadline exceeded",
+		},
+		{
+			name:          "create data path fail",
+			ctx:           context.Background(),
+			kubeClientObj: []runtime.Object{ddInProgress},
+			dataPathMgr:   datapath.NewManager(0),
+			expectedErr:   "error to create data path: Concurrent number exceeds",
+		},
+		{
+			name:          "init data path fail",
+			ctx:           context.Background(),
+			kubeClientObj: []runtime.Object{ddInProgress},
+			initErr:       errors.New("fake-init-error"),
+			expectedErr:   "error to initialize data path: fake-init-error",
+		},
+		{
+			name:          "start data path fail",
+			ctx:           context.Background(),
+			kubeClientObj: []runtime.Object{ddInProgress},
+			startErr:      errors.New("fake-start-error"),
+			expectedErr:   "error starting data path restore: fake-start-error",
+		},
+		{
+			name:             "data path timeout",
+			ctx:              ctxTimeout,
+			kubeClientObj:    []runtime.Object{ddInProgress},
+			dataPathStarted:  true,
+			expectedEventMsg: fmt.Sprintf("Data path for %s started", dataDownloadName),
+			expectedErr:      "timed out waiting for fs restore to complete",
+		},
+		{
+			name:            "data path returns error",
+			ctx:             context.Background(),
+			kubeClientObj:   []runtime.Object{ddInProgress},
+			dataPathStarted: true,
+			result: &dataPathResult{
+				err: errors.New("fake-data-path-error"),
+			},
+			expectedEventMsg: fmt.Sprintf("Data path for %s started", dataDownloadName),
+			expectedErr:      "fake-data-path-error",
+		},
+		{
+			name:            "succeed",
+			ctx:             context.Background(),
+			kubeClientObj:   []runtime.Object{ddInProgress},
+			dataPathStarted: true,
+			result: &dataPathResult{
+				result: "fake-succeed-result",
+			},
+			expectedEventMsg: fmt.Sprintf("Data path for %s started", dataDownloadName),
+		},
+	}
+
+	scheme := runtime.NewScheme()
+	velerov2alpha1api.AddToScheme(scheme)
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			fakeClientBuilder := clientFake.NewClientBuilder()
+			fakeClientBuilder = fakeClientBuilder.WithScheme(scheme)
+
+			fakeClient := fakeClientBuilder.WithRuntimeObjects(test.kubeClientObj...).Build()
+
+			bt := &backupMsTestHelper{}
+
+			rs := &RestoreMicroService{
+				namespace:        velerov1api.DefaultNamespace,
+				dataDownloadName: dataDownloadName,
+				ctx:              context.Background(),
+				client:           fakeClient,
+				dataPathMgr:      datapath.NewManager(1),
+				eventRecorder:    bt,
+				resultSignal:     make(chan dataPathResult),
+				logger:           velerotest.NewLogger(),
+			}
+
+			if test.ctx != nil {
+				rs.ctx = test.ctx
+			}
+
+			if test.dataPathMgr != nil {
+				rs.dataPathMgr = test.dataPathMgr
+			}
+
+			datapath.FSBRCreator = func(string, string, kbclient.Client, string, datapath.Callbacks, logrus.FieldLogger) datapath.AsyncBR {
+				fsBR := datapathmockes.NewAsyncBR(t)
+				if test.initErr != nil {
+					fsBR.On("Init", mock.Anything, mock.Anything).Return(test.initErr)
+				}
+
+				if test.startErr != nil {
+					fsBR.On("Init", mock.Anything, mock.Anything).Return(nil)
+					fsBR.On("StartRestore", mock.Anything, mock.Anything, mock.Anything).Return(test.startErr)
+				}
+
+				if test.dataPathStarted {
+					fsBR.On("Init", mock.Anything, mock.Anything).Return(nil)
+					fsBR.On("StartRestore", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+				}
+
+				return fsBR
+			}
+
+			waitControllerTimeout = time.Second
+
+			if test.result != nil {
+				go func() {
+					time.Sleep(time.Millisecond * 500)
+					rs.resultSignal <- *test.result
+				}()
+			}
+
+			result, err := rs.RunCancelableDataPath(test.ctx)
+
+			if test.expectedErr != "" {
+				assert.EqualError(t, err, test.expectedErr)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, test.result.result, result)
+			}
+
+			if test.expectedEventMsg != "" {
+				assert.True(t, bt.withEvent)
+				assert.Equal(t, test.expectedEventMsg, bt.EventMessage())
+			}
+		})
+	}
+
+	cancel()
+}

pkg/plugin/clientmgmt/itemblockaction/v1/restartable_item_block_action.go

@@ -0,0 +1,115 @@
+/*
+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 v1
+
+import (
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/process"
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	ibav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/itemblockaction/v1"
+)
+
+// AdaptedItemBlockAction is an ItemBlock action adapted to the v1 ItemBlockAction API
+type AdaptedItemBlockAction struct {
+	Kind common.PluginKind
+
+	// Get returns a restartable ItemBlockAction for the given name and process, wrapping if necessary
+	GetRestartable func(name string, restartableProcess process.RestartableProcess) ibav1.ItemBlockAction
+}
+
+func AdaptedItemBlockActions() []AdaptedItemBlockAction {
+	return []AdaptedItemBlockAction{
+		{
+			Kind: common.PluginKindItemBlockAction,
+			GetRestartable: func(name string, restartableProcess process.RestartableProcess) ibav1.ItemBlockAction {
+				return NewRestartableItemBlockAction(name, restartableProcess)
+			},
+		},
+	}
+}
+
+// RestartableItemBlockAction is an ItemBlock action for a given implementation (such as "pod"). It is associated with
+// a restartableProcess, which may be shared and used to run multiple plugins. At the beginning of each method
+// call, the restartableItemBlockAction asks its restartableProcess to restart itself if needed (e.g. if the
+// process terminated for any reason), then it proceeds with the actual call.
+type RestartableItemBlockAction struct {
+	Key                 process.KindAndName
+	SharedPluginProcess process.RestartableProcess
+}
+
+// NewRestartableItemBlockAction returns a new RestartableItemBlockAction.
+func NewRestartableItemBlockAction(name string, sharedPluginProcess process.RestartableProcess) *RestartableItemBlockAction {
+	r := &RestartableItemBlockAction{
+		Key:                 process.KindAndName{Kind: common.PluginKindItemBlockAction, Name: name},
+		SharedPluginProcess: sharedPluginProcess,
+	}
+	return r
+}
+
+// getItemBlockAction returns the ItemBlock action for this restartableItemBlockAction. It does *not* restart the
+// plugin process.
+func (r *RestartableItemBlockAction) getItemBlockAction() (ibav1.ItemBlockAction, error) {
+	plugin, err := r.SharedPluginProcess.GetByKindAndName(r.Key)
+	if err != nil {
+		return nil, err
+	}
+
+	itemBlockAction, ok := plugin.(ibav1.ItemBlockAction)
+	if !ok {
+		return nil, errors.Errorf("plugin %T is not an ItemBlockAction", plugin)
+	}
+
+	return itemBlockAction, nil
+}
+
+// getDelegate restarts the plugin process (if needed) and returns the ItemBlock action for this restartableItemBlockAction.
+func (r *RestartableItemBlockAction) getDelegate() (ibav1.ItemBlockAction, error) {
+	if err := r.SharedPluginProcess.ResetIfNeeded(); err != nil {
+		return nil, err
+	}
+
+	return r.getItemBlockAction()
+}
+
+// Name returns the plugin's name.
+func (r *RestartableItemBlockAction) Name() string {
+	return r.Key.Name
+}
+
+// AppliesTo restarts the plugin's process if needed, then delegates the call.
+func (r *RestartableItemBlockAction) AppliesTo() (velero.ResourceSelector, error) {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return velero.ResourceSelector{}, err
+	}
+
+	return delegate.AppliesTo()
+}
+
+// GetRelatedItems restarts the plugin's process if needed, then delegates the call.
+func (r *RestartableItemBlockAction) GetRelatedItems(item runtime.Unstructured, backup *api.Backup) ([]velero.ResourceIdentifier, error) {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return nil, err
+	}
+
+	return delegate.GetRelatedItems(item, backup)
+}

pkg/plugin/clientmgmt/itemblockaction/v1/restartable_item_block_action_test.go

@@ -0,0 +1,144 @@
+/*
+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 v1
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/vmware-tanzu/velero/internal/restartabletest"
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/process"
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	mocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks/itemblockaction/v1"
+)
+
+func TestRestartableGetItemBlockAction(t *testing.T) {
+	tests := []struct {
+		name          string
+		plugin        interface{}
+		getError      error
+		expectedError string
+	}{
+		{
+			name:          "error getting by kind and name",
+			getError:      errors.Errorf("get error"),
+			expectedError: "get error",
+		},
+		{
+			name:          "wrong type",
+			plugin:        3,
+			expectedError: "plugin int is not an ItemBlockAction",
+		},
+		{
+			name:   "happy path",
+			plugin: new(mocks.ItemBlockAction),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			p := new(restartabletest.MockRestartableProcess)
+			defer p.AssertExpectations(t)
+
+			name := "pod"
+			key := process.KindAndName{Kind: common.PluginKindItemBlockAction, Name: name}
+			p.On("GetByKindAndName", key).Return(tc.plugin, tc.getError)
+
+			r := NewRestartableItemBlockAction(name, p)
+			a, err := r.getItemBlockAction()
+			if tc.expectedError != "" {
+				assert.EqualError(t, err, tc.expectedError)
+				return
+			}
+			require.NoError(t, err)
+
+			assert.Equal(t, tc.plugin, a)
+		})
+	}
+}
+
+func TestRestartableItemBlockActionGetDelegate(t *testing.T) {
+	p := new(restartabletest.MockRestartableProcess)
+	defer p.AssertExpectations(t)
+
+	// Reset error
+	p.On("ResetIfNeeded").Return(errors.Errorf("reset error")).Once()
+	name := "pod"
+	r := NewRestartableItemBlockAction(name, p)
+	a, err := r.getDelegate()
+	assert.Nil(t, a)
+	assert.EqualError(t, err, "reset error")
+
+	// Happy path
+	p.On("ResetIfNeeded").Return(nil)
+	expected := new(mocks.ItemBlockAction)
+	key := process.KindAndName{Kind: common.PluginKindItemBlockAction, Name: name}
+	p.On("GetByKindAndName", key).Return(expected, nil)
+
+	a, err = r.getDelegate()
+	assert.NoError(t, err)
+	assert.Equal(t, expected, a)
+}
+
+func TestRestartableItemBlockActionDelegatedFunctions(t *testing.T) {
+	b := new(v1.Backup)
+
+	pv := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"color": "blue",
+		},
+	}
+
+	relatedItems := []velero.ResourceIdentifier{
+		{
+			GroupResource: schema.GroupResource{Group: "velero.io", Resource: "backups"},
+		},
+	}
+
+	restartabletest.RunRestartableDelegateTests(
+		t,
+		common.PluginKindItemBlockAction,
+		func(key process.KindAndName, p process.RestartableProcess) interface{} {
+			return &RestartableItemBlockAction{
+				Key:                 key,
+				SharedPluginProcess: p,
+			}
+		},
+		func() restartabletest.Mockable {
+			return new(mocks.ItemBlockAction)
+		},
+		restartabletest.RestartableDelegateTest{
+			Function:                "AppliesTo",
+			Inputs:                  []interface{}{},
+			ExpectedErrorOutputs:    []interface{}{velero.ResourceSelector{}, errors.Errorf("reset error")},
+			ExpectedDelegateOutputs: []interface{}{velero.ResourceSelector{IncludedNamespaces: []string{"a"}}, errors.Errorf("delegate error")},
+		},
+		restartabletest.RestartableDelegateTest{
+			Function:                "GetRelatedItems",
+			Inputs:                  []interface{}{pv, b},
+			ExpectedErrorOutputs:    []interface{}{([]velero.ResourceIdentifier)(nil), errors.Errorf("reset error")},
+			ExpectedDelegateOutputs: []interface{}{relatedItems, errors.Errorf("delegate error")},
+		},
+	)
+}

pkg/plugin/clientmgmt/manager.go

@@ -26,6 +26,7 @@ import (
 
 	biav1cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/backupitemaction/v1"
 	biav2cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/backupitemaction/v2"
+	ibav1cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/itemblockaction/v1"
 	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/process"
 	riav1cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/restoreitemaction/v1"
 	riav2cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/restoreitemaction/v2"
@@ -34,6 +35,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	biav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v1"
 	biav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v2"
+	ibav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/itemblockaction/v1"
 	riav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v1"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
 	vsv1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/volumesnapshotter/v1"
@@ -77,6 +79,12 @@ type Manager interface {
 	// GetDeleteItemAction returns the delete item action plugin for name.
 	GetDeleteItemAction(name string) (velero.DeleteItemAction, error)
 
+	// GetItemBlockActions returns all v1 ItemBlock action plugins.
+	GetItemBlockActions() ([]ibav1.ItemBlockAction, error)
+
+	// GetItemBlockAction returns the ItemBlock action plugin for name.
+	GetItemBlockAction(name string) (ibav1.ItemBlockAction, error)
+
 	// CleanupClients terminates all of the Manager's running plugin processes.
 	CleanupClients()
 }
@@ -374,6 +382,44 @@ func (m *manager) GetDeleteItemAction(name string) (velero.DeleteItemAction, err
 	return r, nil
 }
 
+// GetItemBlockActions returns all ItemBlock actions as restartableItemBlockActions.
+func (m *manager) GetItemBlockActions() ([]ibav1.ItemBlockAction, error) {
+	list := m.registry.List(common.PluginKindItemBlockAction)
+
+	actions := make([]ibav1.ItemBlockAction, 0, len(list))
+
+	for i := range list {
+		id := list[i]
+
+		r, err := m.GetItemBlockAction(id.Name)
+		if err != nil {
+			return nil, err
+		}
+
+		actions = append(actions, r)
+	}
+
+	return actions, nil
+}
+
+// GetItemBlockAction returns a restartableItemBlockAction for name.
+func (m *manager) GetItemBlockAction(name string) (ibav1.ItemBlockAction, error) {
+	name = sanitizeName(name)
+
+	for _, adaptedItemBlockAction := range ibav1cli.AdaptedItemBlockActions() {
+		restartableProcess, err := m.getRestartableProcess(adaptedItemBlockAction.Kind, name)
+		// Check if plugin was not found
+		if errors.As(err, &pluginNotFoundErrType) {
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		return adaptedItemBlockAction.GetRestartable(name, restartableProcess), nil
+	}
+	return nil, fmt.Errorf("unable to get valid ItemBlockAction for %q", name)
+}
+
 // sanitizeName adds "velero.io" to legacy plugins that weren't namespaced.
 func sanitizeName(name string) string {
 	// Backwards compatibility with non-namespaced Velero plugins, following principle of least surprise

pkg/plugin/clientmgmt/manager_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/vmware-tanzu/velero/internal/restartabletest"
 	biav1cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/backupitemaction/v1"
 	biav2cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/backupitemaction/v2"
+	ibav1cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/itemblockaction/v1"
 	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/process"
 	riav1cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/restoreitemaction/v1"
 	riav2cli "github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt/restoreitemaction/v2"
@@ -256,6 +257,23 @@ func TestGetRestoreItemActionV2(t *testing.T) {
 	)
 }
 
+func TestGetItemBlockAction(t *testing.T) {
+	getPluginTest(t,
+		common.PluginKindItemBlockAction,
+		"velero.io/pod",
+		func(m Manager, name string) (interface{}, error) {
+			return m.GetItemBlockAction(name)
+		},
+		func(name string, sharedPluginProcess process.RestartableProcess) interface{} {
+			return &ibav1cli.RestartableItemBlockAction{
+				Key:                 process.KindAndName{Kind: common.PluginKindItemBlockAction, Name: name},
+				SharedPluginProcess: sharedPluginProcess,
+			}
+		},
+		false,
+	)
+}
+
 func getPluginTest(
 	t *testing.T,
 	kind common.PluginKind,
@@ -784,6 +802,98 @@ func TestGetDeleteItemActions(t *testing.T) {
 	}
 }
 
+func TestGetItemBlockActions(t *testing.T) {
+	tests := []struct {
+		name                       string
+		names                      []string
+		newRestartableProcessError error
+		expectedError              string
+	}{
+		{
+			name:  "No items",
+			names: []string{},
+		},
+		{
+			name:                       "Error getting restartable process",
+			names:                      []string{"velero.io/a", "velero.io/b", "velero.io/c"},
+			newRestartableProcessError: errors.Errorf("NewRestartableProcess"),
+			expectedError:              "NewRestartableProcess",
+		},
+		{
+			name:  "Happy path",
+			names: []string{"velero.io/a", "velero.io/b", "velero.io/c"},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			logger := test.NewLogger()
+			logLevel := logrus.InfoLevel
+
+			registry := &mockRegistry{}
+			defer registry.AssertExpectations(t)
+
+			m := NewManager(logger, logLevel, registry).(*manager)
+			factory := &mockRestartableProcessFactory{}
+			defer factory.AssertExpectations(t)
+			m.restartableProcessFactory = factory
+
+			pluginKind := common.PluginKindItemBlockAction
+			var pluginIDs []framework.PluginIdentifier
+			for i := range tc.names {
+				pluginID := framework.PluginIdentifier{
+					Command: "/command",
+					Kind:    pluginKind,
+					Name:    tc.names[i],
+				}
+				pluginIDs = append(pluginIDs, pluginID)
+			}
+			registry.On("List", pluginKind).Return(pluginIDs)
+
+			var expectedActions []interface{}
+			for i := range pluginIDs {
+				pluginID := pluginIDs[i]
+				pluginName := pluginID.Name
+
+				registry.On("Get", pluginKind, pluginName).Return(pluginID, nil)
+
+				restartableProcess := &restartabletest.MockRestartableProcess{}
+				defer restartableProcess.AssertExpectations(t)
+
+				expected := &ibav1cli.RestartableItemBlockAction{
+					Key:                 process.KindAndName{Kind: pluginKind, Name: pluginName},
+					SharedPluginProcess: restartableProcess,
+				}
+
+				if tc.newRestartableProcessError != nil {
+					// Test 1: error getting restartable process
+					factory.On("NewRestartableProcess", pluginID.Command, logger, logLevel).Return(nil, errors.Errorf("NewRestartableProcess")).Once()
+					break
+				}
+
+				// Test 2: happy path
+				if i == 0 {
+					factory.On("NewRestartableProcess", pluginID.Command, logger, logLevel).Return(restartableProcess, nil).Once()
+				}
+
+				expectedActions = append(expectedActions, expected)
+			}
+
+			itemBlockActions, err := m.GetItemBlockActions()
+			if tc.newRestartableProcessError != nil {
+				assert.Nil(t, itemBlockActions)
+				assert.EqualError(t, err, "NewRestartableProcess")
+			} else {
+				require.NoError(t, err)
+				var actual []interface{}
+				for i := range itemBlockActions {
+					actual = append(actual, itemBlockActions[i])
+				}
+				assert.Equal(t, expectedActions, actual)
+			}
+		})
+	}
+}
+
 func TestSanitizeName(t *testing.T) {
 	tests := []struct {
 		name, pluginName, expectedName string

pkg/plugin/clientmgmt/process/client_builder.go

@@ -29,6 +29,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
 	biav2 "github.com/vmware-tanzu/velero/pkg/plugin/framework/backupitemaction/v2"
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	ibav1 "github.com/vmware-tanzu/velero/pkg/plugin/framework/itemblockaction/v1"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/framework/restoreitemaction/v2"
 )
 
@@ -78,6 +79,7 @@ func (b *clientBuilder) clientConfig() *hcplugin.ClientConfig {
 			string(common.PluginKindRestoreItemAction):   framework.NewRestoreItemActionPlugin(common.ClientLogger(b.clientLogger)),
 			string(common.PluginKindRestoreItemActionV2): riav2.NewRestoreItemActionPlugin(common.ClientLogger(b.clientLogger)),
 			string(common.PluginKindDeleteItemAction):    framework.NewDeleteItemActionPlugin(common.ClientLogger(b.clientLogger)),
+			string(common.PluginKindItemBlockAction):     ibav1.NewItemBlockActionPlugin(common.ClientLogger(b.clientLogger)),
 		},
 		Logger: b.pluginLogger,
 		Cmd:    exec.Command(b.commandName, b.commandArgs...), //nolint:gosec // Internal call. No need to check the command line.

pkg/plugin/clientmgmt/process/client_builder_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
 	biav2 "github.com/vmware-tanzu/velero/pkg/plugin/framework/backupitemaction/v2"
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	ibav1 "github.com/vmware-tanzu/velero/pkg/plugin/framework/itemblockaction/v1"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/framework/restoreitemaction/v2"
 	"github.com/vmware-tanzu/velero/pkg/test"
 )
@@ -70,6 +71,7 @@ func TestClientConfig(t *testing.T) {
 			string(common.PluginKindRestoreItemAction):   framework.NewRestoreItemActionPlugin(common.ClientLogger(logger)),
 			string(common.PluginKindRestoreItemActionV2): riav2.NewRestoreItemActionPlugin(common.ClientLogger(logger)),
 			string(common.PluginKindDeleteItemAction):    framework.NewDeleteItemActionPlugin(common.ClientLogger(logger)),
+			string(common.PluginKindItemBlockAction):     ibav1.NewItemBlockActionPlugin(common.ClientLogger(logger)),
 		},
 		Logger: cb.pluginLogger,
 		Cmd:    exec.Command(cb.commandName, cb.commandArgs...),

pkg/plugin/framework/action_resolver.go

@@ -27,6 +27,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	biav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v1"
 	biav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v2"
+	ibav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/itemblockaction/v1"
 	riav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v1"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
 	"github.com/vmware-tanzu/velero/pkg/util/collections"
@@ -280,3 +281,38 @@ func (recv DeleteItemActionResolver) ResolveActions(helper discovery.Helper, log
 	}
 	return resolved, nil
 }
+
+type ItemBlockResolvedAction struct {
+	ibav1.ItemBlockAction
+	resolvedAction
+}
+
+type ItemBlockActionResolver struct {
+	actions []ibav1.ItemBlockAction
+}
+
+func NewItemBlockActionResolver(actions []ibav1.ItemBlockAction) ItemBlockActionResolver {
+	return ItemBlockActionResolver{
+		actions: actions,
+	}
+}
+
+func (recv ItemBlockActionResolver) ResolveActions(helper discovery.Helper, log logrus.FieldLogger) ([]ItemBlockResolvedAction, error) {
+	var resolved []ItemBlockResolvedAction
+	for _, action := range recv.actions {
+		resources, namespaces, selector, err := resolveAction(helper, action)
+		if err != nil {
+			return nil, err
+		}
+		res := ItemBlockResolvedAction{
+			ItemBlockAction: action,
+			resolvedAction: resolvedAction{
+				ResourceIncludesExcludes:  resources,
+				NamespaceIncludesExcludes: namespaces,
+				Selector:                  selector,
+			},
+		}
+		resolved = append(resolved, res)
+	}
+	return resolved, nil
+}

pkg/plugin/framework/common/plugin_kinds.go

@@ -47,6 +47,9 @@ const (
 	// PluginKindDeleteItemAction represents a delete item action plugin.
 	PluginKindDeleteItemAction PluginKind = "DeleteItemAction"
 
+	// PluginKindItemBlockAction represents a v1 ItemBlock action plugin.
+	PluginKindItemBlockAction PluginKind = "ItemBlockAction"
+
 	// PluginKindPluginLister represents a plugin lister plugin.
 	PluginKindPluginLister PluginKind = "PluginLister"
 )
@@ -70,5 +73,6 @@ func AllPluginKinds() map[string]PluginKind {
 	allPluginKinds[PluginKindRestoreItemAction.String()] = PluginKindRestoreItemAction
 	allPluginKinds[PluginKindRestoreItemActionV2.String()] = PluginKindRestoreItemActionV2
 	allPluginKinds[PluginKindDeleteItemAction.String()] = PluginKindDeleteItemAction
+	allPluginKinds[PluginKindItemBlockAction.String()] = PluginKindItemBlockAction
 	return allPluginKinds
 }

pkg/plugin/framework/itemblockaction/v1/item_block_action.go

@@ -0,0 +1,45 @@
+/*
+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 v1
+
+import (
+	plugin "github.com/hashicorp/go-plugin"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	protoibav1 "github.com/vmware-tanzu/velero/pkg/plugin/generated/itemblockaction/v1"
+)
+
+// ItemBlockActionPlugin is an implementation of go-plugin's Plugin
+// interface with support for gRPC for the backup/ItemAction
+// interface.
+type ItemBlockActionPlugin struct {
+	plugin.NetRPCUnsupportedPlugin
+	*common.PluginBase
+}
+
+// GRPCClient returns a clientDispenser for ItemBlockAction gRPC clients.
+func (p *ItemBlockActionPlugin) GRPCClient(_ context.Context, _ *plugin.GRPCBroker, clientConn *grpc.ClientConn) (interface{}, error) {
+	return common.NewClientDispenser(p.ClientLogger, clientConn, newItemBlockActionGRPCClient), nil
+}
+
+// GRPCServer registers a ItemBlockAction gRPC server.
+func (p *ItemBlockActionPlugin) GRPCServer(_ *plugin.GRPCBroker, server *grpc.Server) error {
+	protoibav1.RegisterItemBlockActionServer(server, &ItemBlockActionGRPCServer{mux: p.ServerMux})
+	return nil
+}

pkg/plugin/framework/itemblockaction/v1/item_block_action_client.go

@@ -0,0 +1,122 @@
+/*
+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 v1
+
+import (
+	"encoding/json"
+
+	"github.com/pkg/errors"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	protoibav1 "github.com/vmware-tanzu/velero/pkg/plugin/generated/itemblockaction/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+// NewItemBlockActionPlugin constructs a ItemBlockActionPlugin.
+func NewItemBlockActionPlugin(options ...common.PluginOption) *ItemBlockActionPlugin {
+	return &ItemBlockActionPlugin{
+		PluginBase: common.NewPluginBase(options...),
+	}
+}
+
+// ItemBlockActionGRPCClient implements the backup/ItemAction interface and uses a
+// gRPC client to make calls to the plugin server.
+type ItemBlockActionGRPCClient struct {
+	*common.ClientBase
+	grpcClient protoibav1.ItemBlockActionClient
+}
+
+func newItemBlockActionGRPCClient(base *common.ClientBase, clientConn *grpc.ClientConn) interface{} {
+	return &ItemBlockActionGRPCClient{
+		ClientBase: base,
+		grpcClient: protoibav1.NewItemBlockActionClient(clientConn),
+	}
+}
+
+func (c *ItemBlockActionGRPCClient) AppliesTo() (velero.ResourceSelector, error) {
+	req := &protoibav1.ItemBlockActionAppliesToRequest{
+		Plugin: c.Plugin,
+	}
+
+	res, err := c.grpcClient.AppliesTo(context.Background(), req)
+	if err != nil {
+		return velero.ResourceSelector{}, common.FromGRPCError(err)
+	}
+
+	if res.ResourceSelector == nil {
+		return velero.ResourceSelector{}, nil
+	}
+
+	return velero.ResourceSelector{
+		IncludedNamespaces: res.ResourceSelector.IncludedNamespaces,
+		ExcludedNamespaces: res.ResourceSelector.ExcludedNamespaces,
+		IncludedResources:  res.ResourceSelector.IncludedResources,
+		ExcludedResources:  res.ResourceSelector.ExcludedResources,
+		LabelSelector:      res.ResourceSelector.Selector,
+	}, nil
+}
+
+func (c *ItemBlockActionGRPCClient) GetRelatedItems(item runtime.Unstructured, backup *api.Backup) ([]velero.ResourceIdentifier, error) {
+	itemJSON, err := json.Marshal(item.UnstructuredContent())
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	backupJSON, err := json.Marshal(backup)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	req := &protoibav1.ItemBlockActionGetRelatedItemsRequest{
+		Plugin: c.Plugin,
+		Item:   itemJSON,
+		Backup: backupJSON,
+	}
+
+	res, err := c.grpcClient.GetRelatedItems(context.Background(), req)
+	if err != nil {
+		return nil, common.FromGRPCError(err)
+	}
+
+	var relatedItems []velero.ResourceIdentifier
+
+	for _, itm := range res.RelatedItems {
+		newItem := velero.ResourceIdentifier{
+			GroupResource: schema.GroupResource{
+				Group:    itm.Group,
+				Resource: itm.Resource,
+			},
+			Namespace: itm.Namespace,
+			Name:      itm.Name,
+		}
+
+		relatedItems = append(relatedItems, newItem)
+	}
+
+	return relatedItems, nil
+}
+
+// This shouldn't be called on the GRPC client since the RestartableItemBlockAction won't delegate
+// this method
+func (c *ItemBlockActionGRPCClient) Name() string {
+	return ""
+}

pkg/plugin/framework/itemblockaction/v1/item_block_action_server.go

@@ -0,0 +1,134 @@
+/*
+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 v1
+
+import (
+	"encoding/json"
+
+	"github.com/pkg/errors"
+	"golang.org/x/net/context"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+
+	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
+	protoibav1 "github.com/vmware-tanzu/velero/pkg/plugin/generated/itemblockaction/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	ibav1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/itemblockaction/v1"
+)
+
+// ItemBlockActionGRPCServer implements the proto-generated ItemBlockAction interface, and accepts
+// gRPC calls and forwards them to an implementation of the pluggable interface.
+type ItemBlockActionGRPCServer struct {
+	mux *common.ServerMux
+}
+
+func (s *ItemBlockActionGRPCServer) getImpl(name string) (ibav1.ItemBlockAction, error) {
+	impl, err := s.mux.GetHandler(name)
+	if err != nil {
+		return nil, err
+	}
+
+	itemAction, ok := impl.(ibav1.ItemBlockAction)
+	if !ok {
+		return nil, errors.Errorf("%T is not a backup item action", impl)
+	}
+
+	return itemAction, nil
+}
+
+func (s *ItemBlockActionGRPCServer) AppliesTo(
+	ctx context.Context, req *protoibav1.ItemBlockActionAppliesToRequest) (
+	response *protoibav1.ItemBlockActionAppliesToResponse, err error) {
+	defer func() {
+		if recoveredErr := common.HandlePanic(recover()); recoveredErr != nil {
+			err = recoveredErr
+		}
+	}()
+
+	impl, err := s.getImpl(req.Plugin)
+	if err != nil {
+		return nil, common.NewGRPCError(err)
+	}
+
+	resourceSelector, err := impl.AppliesTo()
+	if err != nil {
+		return nil, common.NewGRPCError(err)
+	}
+
+	return &protoibav1.ItemBlockActionAppliesToResponse{
+		ResourceSelector: &proto.ResourceSelector{
+			IncludedNamespaces: resourceSelector.IncludedNamespaces,
+			ExcludedNamespaces: resourceSelector.ExcludedNamespaces,
+			IncludedResources:  resourceSelector.IncludedResources,
+			ExcludedResources:  resourceSelector.ExcludedResources,
+			Selector:           resourceSelector.LabelSelector,
+		},
+	}, nil
+}
+
+func (s *ItemBlockActionGRPCServer) GetRelatedItems(
+	ctx context.Context, req *protoibav1.ItemBlockActionGetRelatedItemsRequest) (response *protoibav1.ItemBlockActionGetRelatedItemsResponse, err error) {
+	defer func() {
+		if recoveredErr := common.HandlePanic(recover()); recoveredErr != nil {
+			err = recoveredErr
+		}
+	}()
+
+	impl, err := s.getImpl(req.Plugin)
+	if err != nil {
+		return nil, common.NewGRPCError(err)
+	}
+
+	var item unstructured.Unstructured
+	var backup api.Backup
+
+	if err := json.Unmarshal(req.Item, &item); err != nil {
+		return nil, common.NewGRPCError(errors.WithStack(err))
+	}
+	if err := json.Unmarshal(req.Backup, &backup); err != nil {
+		return nil, common.NewGRPCError(errors.WithStack(err))
+	}
+
+	relatedItems, err := impl.GetRelatedItems(&item, &backup)
+	if err != nil {
+		return nil, common.NewGRPCError(err)
+	}
+
+	res := &protoibav1.ItemBlockActionGetRelatedItemsResponse{}
+
+	for _, item := range relatedItems {
+		res.RelatedItems = append(res.RelatedItems, backupResourceIdentifierToProto(item))
+	}
+
+	return res, nil
+}
+
+func backupResourceIdentifierToProto(id velero.ResourceIdentifier) *proto.ResourceIdentifier {
+	return &proto.ResourceIdentifier{
+		Group:     id.Group,
+		Resource:  id.Resource,
+		Namespace: id.Namespace,
+		Name:      id.Name,
+	}
+}
+
+// This shouldn't be called on the GRPC server since the server won't ever receive this request, as
+// the RestartableItemBlockAction in Velero won't delegate this to the server
+func (s *ItemBlockActionGRPCServer) Name() string {
+	return ""
+}

pkg/plugin/framework/itemblockaction/v1/item_block_action_test.go

@@ -0,0 +1,163 @@
+/*
+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 v1
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/context"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
+	protoibav1 "github.com/vmware-tanzu/velero/pkg/plugin/generated/itemblockaction/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	mocks "github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks/itemblockaction/v1"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func TestItemBlockActionGRPCServerGetRelatedItems(t *testing.T) {
+	invalidItem := []byte("this is gibberish json")
+	validItem := []byte(`
+	{
+		"apiVersion": "v1",
+		"kind": "ConfigMap",
+		"metadata": {
+			"namespace": "myns",
+			"name": "myconfigmap"
+		},
+		"data": {
+			"key": "value"
+		}
+	}`)
+	var validItemObject unstructured.Unstructured
+	err := json.Unmarshal(validItem, &validItemObject)
+	require.NoError(t, err)
+
+	invalidBackup := []byte("this is gibberish json")
+	validBackup := []byte(`
+	{
+		"apiVersion": "velero.io/v1",
+		"kind": "Backup",
+		"metadata": {
+			"namespace": "myns",
+			"name": "mybackup"
+		},
+		"spec": {
+			"includedNamespaces": ["*"],
+			"includedResources": ["*"],
+			"ttl": "60m"
+		}
+	}`)
+	var validBackupObject v1.Backup
+	err = json.Unmarshal(validBackup, &validBackupObject)
+	require.NoError(t, err)
+
+	tests := []struct {
+		name             string
+		backup           []byte
+		item             []byte
+		implRelatedItems []velero.ResourceIdentifier
+		implError        error
+		expectError      bool
+		skipMock         bool
+	}{
+		{
+			name:        "error unmarshaling item",
+			item:        invalidItem,
+			backup:      validBackup,
+			expectError: true,
+			skipMock:    true,
+		},
+		{
+			name:        "error unmarshaling backup",
+			item:        validItem,
+			backup:      invalidBackup,
+			expectError: true,
+			skipMock:    true,
+		},
+		{
+			name:        "error running impl",
+			item:        validItem,
+			backup:      validBackup,
+			implError:   errors.New("impl error"),
+			expectError: true,
+		},
+		{
+			name:   "no relatedItems",
+			item:   validItem,
+			backup: validBackup,
+		},
+		{
+			name:   "some relatedItems",
+			item:   validItem,
+			backup: validBackup,
+			implRelatedItems: []velero.ResourceIdentifier{
+				{
+					GroupResource: schema.GroupResource{Group: "v1", Resource: "pods"},
+					Namespace:     "myns",
+					Name:          "mypod",
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			itemAction := &mocks.ItemBlockAction{}
+			defer itemAction.AssertExpectations(t)
+
+			if !test.skipMock {
+				itemAction.On("GetRelatedItems", &validItemObject, &validBackupObject).Return(test.implRelatedItems, test.implError)
+			}
+
+			s := &ItemBlockActionGRPCServer{mux: &common.ServerMux{
+				ServerLog: velerotest.NewLogger(),
+				Handlers: map[string]interface{}{
+					"xyz": itemAction,
+				},
+			}}
+
+			req := &protoibav1.ItemBlockActionGetRelatedItemsRequest{
+				Plugin: "xyz",
+				Item:   test.item,
+				Backup: test.backup,
+			}
+
+			resp, err := s.GetRelatedItems(context.Background(), req)
+
+			// Verify error
+			assert.Equal(t, test.expectError, err != nil)
+			if err != nil {
+				return
+			}
+			require.NotNil(t, resp)
+
+			// Verify related items
+			var expectedRelatedItems []*proto.ResourceIdentifier
+			for _, item := range test.implRelatedItems {
+				expectedRelatedItems = append(expectedRelatedItems, backupResourceIdentifierToProto(item))
+			}
+			assert.Equal(t, expectedRelatedItems, resp.RelatedItems)
+		})
+	}
+}

pkg/plugin/framework/server.go

@@ -27,6 +27,7 @@ import (
 
 	biav2 "github.com/vmware-tanzu/velero/pkg/plugin/framework/backupitemaction/v2"
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+	ibav1 "github.com/vmware-tanzu/velero/pkg/plugin/framework/itemblockaction/v1"
 	riav2 "github.com/vmware-tanzu/velero/pkg/plugin/framework/restoreitemaction/v2"
 	"github.com/vmware-tanzu/velero/pkg/util/logging"
 )
@@ -90,6 +91,13 @@ type Server interface {
 	// RegisterDeleteItemActions registers multiple Delete item actions.
 	RegisterDeleteItemActions(map[string]common.HandlerInitializer) Server
 
+	// RegisterItemBlockAction registers a ItemBlock action. Accepted format
+	// for the plugin name is <DNS subdomain>/<non-empty name>.
+	RegisterItemBlockAction(pluginName string, initializer common.HandlerInitializer) Server
+
+	// RegisterItemBlockActions registers multiple ItemBlock actions.
+	RegisterItemBlockActions(map[string]common.HandlerInitializer) Server
+
 	// Server runs the plugin server.
 	Serve()
 }
@@ -106,6 +114,7 @@ type server struct {
 	restoreItemAction   *RestoreItemActionPlugin
 	restoreItemActionV2 *riav2.RestoreItemActionPlugin
 	deleteItemAction    *DeleteItemActionPlugin
+	itemBlockAction     *ibav1.ItemBlockActionPlugin
 }
 
 // NewServer returns a new Server
@@ -122,6 +131,7 @@ func NewServer() Server {
 		restoreItemAction:   NewRestoreItemActionPlugin(common.ServerLogger(log)),
 		restoreItemActionV2: riav2.NewRestoreItemActionPlugin(common.ServerLogger(log)),
 		deleteItemAction:    NewDeleteItemActionPlugin(common.ServerLogger(log)),
+		itemBlockAction:     ibav1.NewItemBlockActionPlugin(common.ServerLogger(log)),
 	}
 }
 
@@ -217,6 +227,18 @@ func (s *server) RegisterDeleteItemActions(m map[string]common.HandlerInitialize
 	return s
 }
 
+func (s *server) RegisterItemBlockAction(name string, initializer common.HandlerInitializer) Server {
+	s.itemBlockAction.Register(name, initializer)
+	return s
+}
+
+func (s *server) RegisterItemBlockActions(m map[string]common.HandlerInitializer) Server {
+	for name := range m {
+		s.RegisterItemBlockAction(name, m[name])
+	}
+	return s
+}
+
 // getNames returns a list of PluginIdentifiers registered with plugin.
 func getNames(command string, kind common.PluginKind, plugin Interface) []PluginIdentifier {
 	var pluginIdentifiers []PluginIdentifier
@@ -251,6 +273,7 @@ func (s *server) Serve() {
 	pluginIdentifiers = append(pluginIdentifiers, getNames(command, common.PluginKindRestoreItemAction, s.restoreItemAction)...)
 	pluginIdentifiers = append(pluginIdentifiers, getNames(command, common.PluginKindRestoreItemActionV2, s.restoreItemActionV2)...)
 	pluginIdentifiers = append(pluginIdentifiers, getNames(command, common.PluginKindDeleteItemAction, s.deleteItemAction)...)
+	pluginIdentifiers = append(pluginIdentifiers, getNames(command, common.PluginKindItemBlockAction, s.itemBlockAction)...)
 
 	pluginLister := NewPluginLister(pluginIdentifiers...)
 
@@ -265,6 +288,7 @@ func (s *server) Serve() {
 			string(common.PluginKindRestoreItemAction):   s.restoreItemAction,
 			string(common.PluginKindRestoreItemActionV2): s.restoreItemActionV2,
 			string(common.PluginKindDeleteItemAction):    s.deleteItemAction,
+			string(common.PluginKindItemBlockAction):     s.itemBlockAction,
 		},
 		GRPCServer: plugin.DefaultGRPCServer,
 	})

pkg/plugin/generated/itemblockaction/v1/ItemBlockAction.pb.go

@@ -0,0 +1,388 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.33.0
+// 	protoc        v4.25.2
+// source: itemblockaction/v1/ItemBlockAction.proto
+
+package v1
+
+import (
+	generated "github.com/vmware-tanzu/velero/pkg/plugin/generated"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type ItemBlockActionAppliesToRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Plugin string `protobuf:"bytes,1,opt,name=plugin,proto3" json:"plugin,omitempty"`
+}
+
+func (x *ItemBlockActionAppliesToRequest) Reset() {
+	*x = ItemBlockActionAppliesToRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ItemBlockActionAppliesToRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ItemBlockActionAppliesToRequest) ProtoMessage() {}
+
+func (x *ItemBlockActionAppliesToRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ItemBlockActionAppliesToRequest.ProtoReflect.Descriptor instead.
+func (*ItemBlockActionAppliesToRequest) Descriptor() ([]byte, []int) {
+	return file_itemblockaction_v1_ItemBlockAction_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *ItemBlockActionAppliesToRequest) GetPlugin() string {
+	if x != nil {
+		return x.Plugin
+	}
+	return ""
+}
+
+type ItemBlockActionAppliesToResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	ResourceSelector *generated.ResourceSelector `protobuf:"bytes,1,opt,name=ResourceSelector,proto3" json:"ResourceSelector,omitempty"`
+}
+
+func (x *ItemBlockActionAppliesToResponse) Reset() {
+	*x = ItemBlockActionAppliesToResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ItemBlockActionAppliesToResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ItemBlockActionAppliesToResponse) ProtoMessage() {}
+
+func (x *ItemBlockActionAppliesToResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ItemBlockActionAppliesToResponse.ProtoReflect.Descriptor instead.
+func (*ItemBlockActionAppliesToResponse) Descriptor() ([]byte, []int) {
+	return file_itemblockaction_v1_ItemBlockAction_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *ItemBlockActionAppliesToResponse) GetResourceSelector() *generated.ResourceSelector {
+	if x != nil {
+		return x.ResourceSelector
+	}
+	return nil
+}
+
+type ItemBlockActionGetRelatedItemsRequest struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Plugin string `protobuf:"bytes,1,opt,name=plugin,proto3" json:"plugin,omitempty"`
+	Item   []byte `protobuf:"bytes,2,opt,name=item,proto3" json:"item,omitempty"`
+	Backup []byte `protobuf:"bytes,3,opt,name=backup,proto3" json:"backup,omitempty"`
+}
+
+func (x *ItemBlockActionGetRelatedItemsRequest) Reset() {
+	*x = ItemBlockActionGetRelatedItemsRequest{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ItemBlockActionGetRelatedItemsRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ItemBlockActionGetRelatedItemsRequest) ProtoMessage() {}
+
+func (x *ItemBlockActionGetRelatedItemsRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ItemBlockActionGetRelatedItemsRequest.ProtoReflect.Descriptor instead.
+func (*ItemBlockActionGetRelatedItemsRequest) Descriptor() ([]byte, []int) {
+	return file_itemblockaction_v1_ItemBlockAction_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *ItemBlockActionGetRelatedItemsRequest) GetPlugin() string {
+	if x != nil {
+		return x.Plugin
+	}
+	return ""
+}
+
+func (x *ItemBlockActionGetRelatedItemsRequest) GetItem() []byte {
+	if x != nil {
+		return x.Item
+	}
+	return nil
+}
+
+func (x *ItemBlockActionGetRelatedItemsRequest) GetBackup() []byte {
+	if x != nil {
+		return x.Backup
+	}
+	return nil
+}
+
+type ItemBlockActionGetRelatedItemsResponse struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	RelatedItems []*generated.ResourceIdentifier `protobuf:"bytes,1,rep,name=relatedItems,proto3" json:"relatedItems,omitempty"`
+}
+
+func (x *ItemBlockActionGetRelatedItemsResponse) Reset() {
+	*x = ItemBlockActionGetRelatedItemsResponse{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *ItemBlockActionGetRelatedItemsResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ItemBlockActionGetRelatedItemsResponse) ProtoMessage() {}
+
+func (x *ItemBlockActionGetRelatedItemsResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ItemBlockActionGetRelatedItemsResponse.ProtoReflect.Descriptor instead.
+func (*ItemBlockActionGetRelatedItemsResponse) Descriptor() ([]byte, []int) {
+	return file_itemblockaction_v1_ItemBlockAction_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *ItemBlockActionGetRelatedItemsResponse) GetRelatedItems() []*generated.ResourceIdentifier {
+	if x != nil {
+		return x.RelatedItems
+	}
+	return nil
+}
+
+var File_itemblockaction_v1_ItemBlockAction_proto protoreflect.FileDescriptor
+
+var file_itemblockaction_v1_ItemBlockAction_proto_rawDesc = []byte{
+	0x0a, 0x28, 0x69, 0x74, 0x65, 0x6d, 0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x61, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x2f, 0x76, 0x31, 0x2f, 0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63,
+	0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x02, 0x76, 0x31, 0x1a, 0x0c,
+	0x53, 0x68, 0x61, 0x72, 0x65, 0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x39, 0x0a, 0x1f,
+	0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x41,
+	0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12,
+	0x16, 0x0a, 0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x22, 0x6b, 0x0a, 0x20, 0x49, 0x74, 0x65, 0x6d, 0x42,
+	0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65,
+	0x73, 0x54, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x47, 0x0a, 0x10, 0x52,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65, 0x63, 0x74, 0x6f, 0x72, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1b, 0x2e, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65,
+	0x64, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65, 0x63, 0x74,
+	0x6f, 0x72, 0x52, 0x10, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x53, 0x65, 0x6c, 0x65,
+	0x63, 0x74, 0x6f, 0x72, 0x22, 0x6b, 0x0a, 0x25, 0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63,
+	0x6b, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x6c, 0x61, 0x74, 0x65,
+	0x64, 0x49, 0x74, 0x65, 0x6d, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x16, 0x0a,
+	0x06, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x70,
+	0x6c, 0x75, 0x67, 0x69, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x0c, 0x52, 0x04, 0x69, 0x74, 0x65, 0x6d, 0x12, 0x16, 0x0a, 0x06, 0x62, 0x61, 0x63,
+	0x6b, 0x75, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x62, 0x61, 0x63, 0x6b, 0x75,
+	0x70, 0x22, 0x6b, 0x0a, 0x26, 0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63,
+	0x74, 0x69, 0x6f, 0x6e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x49, 0x74,
+	0x65, 0x6d, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x41, 0x0a, 0x0c, 0x72,
+	0x65, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x49, 0x74, 0x65, 0x6d, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28,
+	0x0b, 0x32, 0x1d, 0x2e, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2e, 0x52, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x64, 0x65, 0x6e, 0x74, 0x69, 0x66, 0x69, 0x65, 0x72,
+	0x52, 0x0c, 0x72, 0x65, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x49, 0x74, 0x65, 0x6d, 0x73, 0x32, 0xd3,
+	0x01, 0x0a, 0x0f, 0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63, 0x74, 0x69,
+	0x6f, 0x6e, 0x12, 0x56, 0x0a, 0x09, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x12,
+	0x23, 0x2e, 0x76, 0x31, 0x2e, 0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63,
+	0x74, 0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73, 0x54, 0x6f, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x1a, 0x24, 0x2e, 0x76, 0x31, 0x2e, 0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c,
+	0x6f, 0x63, 0x6b, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x41, 0x70, 0x70, 0x6c, 0x69, 0x65, 0x73,
+	0x54, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x68, 0x0a, 0x0f, 0x47, 0x65,
+	0x74, 0x52, 0x65, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x49, 0x74, 0x65, 0x6d, 0x73, 0x12, 0x29, 0x2e,
+	0x76, 0x31, 0x2e, 0x49, 0x74, 0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63, 0x74, 0x69,
+	0x6f, 0x6e, 0x47, 0x65, 0x74, 0x52, 0x65, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x49, 0x74, 0x65, 0x6d,
+	0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2a, 0x2e, 0x76, 0x31, 0x2e, 0x49, 0x74,
+	0x65, 0x6d, 0x42, 0x6c, 0x6f, 0x63, 0x6b, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x47, 0x65, 0x74,
+	0x52, 0x65, 0x6c, 0x61, 0x74, 0x65, 0x64, 0x49, 0x74, 0x65, 0x6d, 0x73, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x42, 0x48, 0x5a, 0x46, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
+	0x6f, 0x6d, 0x2f, 0x76, 0x6d, 0x77, 0x61, 0x72, 0x65, 0x2d, 0x74, 0x61, 0x6e, 0x7a, 0x75, 0x2f,
+	0x76, 0x65, 0x6c, 0x65, 0x72, 0x6f, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x70, 0x6c, 0x75, 0x67, 0x69,
+	0x6e, 0x2f, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2f, 0x69, 0x74, 0x65, 0x6d,
+	0x62, 0x6c, 0x6f, 0x63, 0x6b, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x2f, 0x76, 0x31, 0x62, 0x06,
+	0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_itemblockaction_v1_ItemBlockAction_proto_rawDescOnce sync.Once
+	file_itemblockaction_v1_ItemBlockAction_proto_rawDescData = file_itemblockaction_v1_ItemBlockAction_proto_rawDesc
+)
+
+func file_itemblockaction_v1_ItemBlockAction_proto_rawDescGZIP() []byte {
+	file_itemblockaction_v1_ItemBlockAction_proto_rawDescOnce.Do(func() {
+		file_itemblockaction_v1_ItemBlockAction_proto_rawDescData = protoimpl.X.CompressGZIP(file_itemblockaction_v1_ItemBlockAction_proto_rawDescData)
+	})
+	return file_itemblockaction_v1_ItemBlockAction_proto_rawDescData
+}
+
+var file_itemblockaction_v1_ItemBlockAction_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
+var file_itemblockaction_v1_ItemBlockAction_proto_goTypes = []interface{}{
+	(*ItemBlockActionAppliesToRequest)(nil),        // 0: v1.ItemBlockActionAppliesToRequest
+	(*ItemBlockActionAppliesToResponse)(nil),       // 1: v1.ItemBlockActionAppliesToResponse
+	(*ItemBlockActionGetRelatedItemsRequest)(nil),  // 2: v1.ItemBlockActionGetRelatedItemsRequest
+	(*ItemBlockActionGetRelatedItemsResponse)(nil), // 3: v1.ItemBlockActionGetRelatedItemsResponse
+	(*generated.ResourceSelector)(nil),             // 4: generated.ResourceSelector
+	(*generated.ResourceIdentifier)(nil),           // 5: generated.ResourceIdentifier
+}
+var file_itemblockaction_v1_ItemBlockAction_proto_depIdxs = []int32{
+	4, // 0: v1.ItemBlockActionAppliesToResponse.ResourceSelector:type_name -> generated.ResourceSelector
+	5, // 1: v1.ItemBlockActionGetRelatedItemsResponse.relatedItems:type_name -> generated.ResourceIdentifier
+	0, // 2: v1.ItemBlockAction.AppliesTo:input_type -> v1.ItemBlockActionAppliesToRequest
+	2, // 3: v1.ItemBlockAction.GetRelatedItems:input_type -> v1.ItemBlockActionGetRelatedItemsRequest
+	1, // 4: v1.ItemBlockAction.AppliesTo:output_type -> v1.ItemBlockActionAppliesToResponse
+	3, // 5: v1.ItemBlockAction.GetRelatedItems:output_type -> v1.ItemBlockActionGetRelatedItemsResponse
+	4, // [4:6] is the sub-list for method output_type
+	2, // [2:4] is the sub-list for method input_type
+	2, // [2:2] is the sub-list for extension type_name
+	2, // [2:2] is the sub-list for extension extendee
+	0, // [0:2] is the sub-list for field type_name
+}
+
+func init() { file_itemblockaction_v1_ItemBlockAction_proto_init() }
+func file_itemblockaction_v1_ItemBlockAction_proto_init() {
+	if File_itemblockaction_v1_ItemBlockAction_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ItemBlockActionAppliesToRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ItemBlockActionAppliesToResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ItemBlockActionGetRelatedItemsRequest); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_itemblockaction_v1_ItemBlockAction_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*ItemBlockActionGetRelatedItemsResponse); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_itemblockaction_v1_ItemBlockAction_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   4,
+			NumExtensions: 0,
+			NumServices:   1,
+		},
+		GoTypes:           file_itemblockaction_v1_ItemBlockAction_proto_goTypes,
+		DependencyIndexes: file_itemblockaction_v1_ItemBlockAction_proto_depIdxs,
+		MessageInfos:      file_itemblockaction_v1_ItemBlockAction_proto_msgTypes,
+	}.Build()
+	File_itemblockaction_v1_ItemBlockAction_proto = out.File
+	file_itemblockaction_v1_ItemBlockAction_proto_rawDesc = nil
+	file_itemblockaction_v1_ItemBlockAction_proto_goTypes = nil
+	file_itemblockaction_v1_ItemBlockAction_proto_depIdxs = nil
+}

pkg/plugin/generated/itemblockaction/v1/ItemBlockAction_grpc.pb.go

@@ -0,0 +1,144 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.3.0
+// - protoc             v4.25.2
+// source: itemblockaction/v1/ItemBlockAction.proto
+
+package v1
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.32.0 or later.
+const _ = grpc.SupportPackageIsVersion7
+
+const (
+	ItemBlockAction_AppliesTo_FullMethodName       = "/v1.ItemBlockAction/AppliesTo"
+	ItemBlockAction_GetRelatedItems_FullMethodName = "/v1.ItemBlockAction/GetRelatedItems"
+)
+
+// ItemBlockActionClient is the client API for ItemBlockAction service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+type ItemBlockActionClient interface {
+	AppliesTo(ctx context.Context, in *ItemBlockActionAppliesToRequest, opts ...grpc.CallOption) (*ItemBlockActionAppliesToResponse, error)
+	GetRelatedItems(ctx context.Context, in *ItemBlockActionGetRelatedItemsRequest, opts ...grpc.CallOption) (*ItemBlockActionGetRelatedItemsResponse, error)
+}
+
+type itemBlockActionClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewItemBlockActionClient(cc grpc.ClientConnInterface) ItemBlockActionClient {
+	return &itemBlockActionClient{cc}
+}
+
+func (c *itemBlockActionClient) AppliesTo(ctx context.Context, in *ItemBlockActionAppliesToRequest, opts ...grpc.CallOption) (*ItemBlockActionAppliesToResponse, error) {
+	out := new(ItemBlockActionAppliesToResponse)
+	err := c.cc.Invoke(ctx, ItemBlockAction_AppliesTo_FullMethodName, in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *itemBlockActionClient) GetRelatedItems(ctx context.Context, in *ItemBlockActionGetRelatedItemsRequest, opts ...grpc.CallOption) (*ItemBlockActionGetRelatedItemsResponse, error) {
+	out := new(ItemBlockActionGetRelatedItemsResponse)
+	err := c.cc.Invoke(ctx, ItemBlockAction_GetRelatedItems_FullMethodName, in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// ItemBlockActionServer is the server API for ItemBlockAction service.
+// All implementations should embed UnimplementedItemBlockActionServer
+// for forward compatibility
+type ItemBlockActionServer interface {
+	AppliesTo(context.Context, *ItemBlockActionAppliesToRequest) (*ItemBlockActionAppliesToResponse, error)
+	GetRelatedItems(context.Context, *ItemBlockActionGetRelatedItemsRequest) (*ItemBlockActionGetRelatedItemsResponse, error)
+}
+
+// UnimplementedItemBlockActionServer should be embedded to have forward compatible implementations.
+type UnimplementedItemBlockActionServer struct {
+}
+
+func (UnimplementedItemBlockActionServer) AppliesTo(context.Context, *ItemBlockActionAppliesToRequest) (*ItemBlockActionAppliesToResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method AppliesTo not implemented")
+}
+func (UnimplementedItemBlockActionServer) GetRelatedItems(context.Context, *ItemBlockActionGetRelatedItemsRequest) (*ItemBlockActionGetRelatedItemsResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetRelatedItems not implemented")
+}
+
+// UnsafeItemBlockActionServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to ItemBlockActionServer will
+// result in compilation errors.
+type UnsafeItemBlockActionServer interface {
+	mustEmbedUnimplementedItemBlockActionServer()
+}
+
+func RegisterItemBlockActionServer(s grpc.ServiceRegistrar, srv ItemBlockActionServer) {
+	s.RegisterService(&ItemBlockAction_ServiceDesc, srv)
+}
+
+func _ItemBlockAction_AppliesTo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ItemBlockActionAppliesToRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ItemBlockActionServer).AppliesTo(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: ItemBlockAction_AppliesTo_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ItemBlockActionServer).AppliesTo(ctx, req.(*ItemBlockActionAppliesToRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _ItemBlockAction_GetRelatedItems_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ItemBlockActionGetRelatedItemsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ItemBlockActionServer).GetRelatedItems(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: ItemBlockAction_GetRelatedItems_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ItemBlockActionServer).GetRelatedItems(ctx, req.(*ItemBlockActionGetRelatedItemsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// ItemBlockAction_ServiceDesc is the grpc.ServiceDesc for ItemBlockAction service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var ItemBlockAction_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "v1.ItemBlockAction",
+	HandlerType: (*ItemBlockActionServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "AppliesTo",
+			Handler:    _ItemBlockAction_AppliesTo_Handler,
+		},
+		{
+			MethodName: "GetRelatedItems",
+			Handler:    _ItemBlockAction_GetRelatedItems_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "itemblockaction/v1/ItemBlockAction.proto",
+}

pkg/plugin/mocks/manager.go

@@ -1,9 +1,11 @@
-// Code generated by mockery v1.0.0. DO NOT EDIT.
+// Code generated by mockery v2.43.2. DO NOT EDIT.
 
 package mocks
 
 import (
 	mock "github.com/stretchr/testify/mock"
+	itemblockactionv1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/itemblockaction/v1"
+
 	restoreitemactionv1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v1"
 
 	restoreitemactionv2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
@@ -31,7 +33,15 @@ func (_m *Manager) CleanupClients() {
 func (_m *Manager) GetBackupItemAction(name string) (v1.BackupItemAction, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupItemAction")
+	}
+
 	var r0 v1.BackupItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (v1.BackupItemAction, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) v1.BackupItemAction); ok {
 		r0 = rf(name)
 	} else {
@@ -40,7 +50,6 @@ func (_m *Manager) GetBackupItemAction(name string) (v1.BackupItemAction, error)
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -54,7 +63,15 @@ func (_m *Manager) GetBackupItemAction(name string) (v1.BackupItemAction, error)
 func (_m *Manager) GetBackupItemActionV2(name string) (v2.BackupItemAction, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupItemActionV2")
+	}
+
 	var r0 v2.BackupItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (v2.BackupItemAction, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) v2.BackupItemAction); ok {
 		r0 = rf(name)
 	} else {
@@ -63,7 +80,6 @@ func (_m *Manager) GetBackupItemActionV2(name string) (v2.BackupItemAction, erro
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -77,7 +93,15 @@ func (_m *Manager) GetBackupItemActionV2(name string) (v2.BackupItemAction, erro
 func (_m *Manager) GetBackupItemActions() ([]v1.BackupItemAction, error) {
 	ret := _m.Called()
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupItemActions")
+	}
+
 	var r0 []v1.BackupItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]v1.BackupItemAction, error)); ok {
+		return rf()
+	}
 	if rf, ok := ret.Get(0).(func() []v1.BackupItemAction); ok {
 		r0 = rf()
 	} else {
@@ -86,7 +110,6 @@ func (_m *Manager) GetBackupItemActions() ([]v1.BackupItemAction, error) {
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func() error); ok {
 		r1 = rf()
 	} else {
@@ -100,7 +123,15 @@ func (_m *Manager) GetBackupItemActions() ([]v1.BackupItemAction, error) {
 func (_m *Manager) GetBackupItemActionsV2() ([]v2.BackupItemAction, error) {
 	ret := _m.Called()
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetBackupItemActionsV2")
+	}
+
 	var r0 []v2.BackupItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]v2.BackupItemAction, error)); ok {
+		return rf()
+	}
 	if rf, ok := ret.Get(0).(func() []v2.BackupItemAction); ok {
 		r0 = rf()
 	} else {
@@ -109,7 +140,6 @@ func (_m *Manager) GetBackupItemActionsV2() ([]v2.BackupItemAction, error) {
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func() error); ok {
 		r1 = rf()
 	} else {
@@ -123,7 +153,15 @@ func (_m *Manager) GetBackupItemActionsV2() ([]v2.BackupItemAction, error) {
 func (_m *Manager) GetDeleteItemAction(name string) (velero.DeleteItemAction, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetDeleteItemAction")
+	}
+
 	var r0 velero.DeleteItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (velero.DeleteItemAction, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) velero.DeleteItemAction); ok {
 		r0 = rf(name)
 	} else {
@@ -132,7 +170,6 @@ func (_m *Manager) GetDeleteItemAction(name string) (velero.DeleteItemAction, er
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -146,7 +183,15 @@ func (_m *Manager) GetDeleteItemAction(name string) (velero.DeleteItemAction, er
 func (_m *Manager) GetDeleteItemActions() ([]velero.DeleteItemAction, error) {
 	ret := _m.Called()
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetDeleteItemActions")
+	}
+
 	var r0 []velero.DeleteItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]velero.DeleteItemAction, error)); ok {
+		return rf()
+	}
 	if rf, ok := ret.Get(0).(func() []velero.DeleteItemAction); ok {
 		r0 = rf()
 	} else {
@@ -155,7 +200,66 @@ func (_m *Manager) GetDeleteItemActions() ([]velero.DeleteItemAction, error) {
 		}
 	}
 
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetItemBlockAction provides a mock function with given fields: name
+func (_m *Manager) GetItemBlockAction(name string) (itemblockactionv1.ItemBlockAction, error) {
+	ret := _m.Called(name)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetItemBlockAction")
+	}
+
+	var r0 itemblockactionv1.ItemBlockAction
 	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (itemblockactionv1.ItemBlockAction, error)); ok {
+		return rf(name)
+	}
+	if rf, ok := ret.Get(0).(func(string) itemblockactionv1.ItemBlockAction); ok {
+		r0 = rf(name)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(itemblockactionv1.ItemBlockAction)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(name)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetItemBlockActions provides a mock function with given fields:
+func (_m *Manager) GetItemBlockActions() ([]itemblockactionv1.ItemBlockAction, error) {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetItemBlockActions")
+	}
+
+	var r0 []itemblockactionv1.ItemBlockAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]itemblockactionv1.ItemBlockAction, error)); ok {
+		return rf()
+	}
+	if rf, ok := ret.Get(0).(func() []itemblockactionv1.ItemBlockAction); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]itemblockactionv1.ItemBlockAction)
+		}
+	}
+
 	if rf, ok := ret.Get(1).(func() error); ok {
 		r1 = rf()
 	} else {
@@ -169,7 +273,15 @@ func (_m *Manager) GetDeleteItemActions() ([]velero.DeleteItemAction, error) {
 func (_m *Manager) GetObjectStore(name string) (velero.ObjectStore, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetObjectStore")
+	}
+
 	var r0 velero.ObjectStore
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (velero.ObjectStore, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) velero.ObjectStore); ok {
 		r0 = rf(name)
 	} else {
@@ -178,7 +290,6 @@ func (_m *Manager) GetObjectStore(name string) (velero.ObjectStore, error) {
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -192,7 +303,15 @@ func (_m *Manager) GetObjectStore(name string) (velero.ObjectStore, error) {
 func (_m *Manager) GetRestoreItemAction(name string) (restoreitemactionv1.RestoreItemAction, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetRestoreItemAction")
+	}
+
 	var r0 restoreitemactionv1.RestoreItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (restoreitemactionv1.RestoreItemAction, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) restoreitemactionv1.RestoreItemAction); ok {
 		r0 = rf(name)
 	} else {
@@ -201,7 +320,6 @@ func (_m *Manager) GetRestoreItemAction(name string) (restoreitemactionv1.Restor
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -215,7 +333,15 @@ func (_m *Manager) GetRestoreItemAction(name string) (restoreitemactionv1.Restor
 func (_m *Manager) GetRestoreItemActionV2(name string) (restoreitemactionv2.RestoreItemAction, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetRestoreItemActionV2")
+	}
+
 	var r0 restoreitemactionv2.RestoreItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (restoreitemactionv2.RestoreItemAction, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) restoreitemactionv2.RestoreItemAction); ok {
 		r0 = rf(name)
 	} else {
@@ -224,7 +350,6 @@ func (_m *Manager) GetRestoreItemActionV2(name string) (restoreitemactionv2.Rest
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -238,7 +363,15 @@ func (_m *Manager) GetRestoreItemActionV2(name string) (restoreitemactionv2.Rest
 func (_m *Manager) GetRestoreItemActions() ([]restoreitemactionv1.RestoreItemAction, error) {
 	ret := _m.Called()
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetRestoreItemActions")
+	}
+
 	var r0 []restoreitemactionv1.RestoreItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]restoreitemactionv1.RestoreItemAction, error)); ok {
+		return rf()
+	}
 	if rf, ok := ret.Get(0).(func() []restoreitemactionv1.RestoreItemAction); ok {
 		r0 = rf()
 	} else {
@@ -247,7 +380,6 @@ func (_m *Manager) GetRestoreItemActions() ([]restoreitemactionv1.RestoreItemAct
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func() error); ok {
 		r1 = rf()
 	} else {
@@ -261,7 +393,15 @@ func (_m *Manager) GetRestoreItemActions() ([]restoreitemactionv1.RestoreItemAct
 func (_m *Manager) GetRestoreItemActionsV2() ([]restoreitemactionv2.RestoreItemAction, error) {
 	ret := _m.Called()
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetRestoreItemActionsV2")
+	}
+
 	var r0 []restoreitemactionv2.RestoreItemAction
+	var r1 error
+	if rf, ok := ret.Get(0).(func() ([]restoreitemactionv2.RestoreItemAction, error)); ok {
+		return rf()
+	}
 	if rf, ok := ret.Get(0).(func() []restoreitemactionv2.RestoreItemAction); ok {
 		r0 = rf()
 	} else {
@@ -270,7 +410,6 @@ func (_m *Manager) GetRestoreItemActionsV2() ([]restoreitemactionv2.RestoreItemA
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func() error); ok {
 		r1 = rf()
 	} else {
@@ -284,7 +423,15 @@ func (_m *Manager) GetRestoreItemActionsV2() ([]restoreitemactionv2.RestoreItemA
 func (_m *Manager) GetVolumeSnapshotter(name string) (volumesnapshotterv1.VolumeSnapshotter, error) {
 	ret := _m.Called(name)
 
+	if len(ret) == 0 {
+		panic("no return value specified for GetVolumeSnapshotter")
+	}
+
 	var r0 volumesnapshotterv1.VolumeSnapshotter
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (volumesnapshotterv1.VolumeSnapshotter, error)); ok {
+		return rf(name)
+	}
 	if rf, ok := ret.Get(0).(func(string) volumesnapshotterv1.VolumeSnapshotter); ok {
 		r0 = rf(name)
 	} else {
@@ -293,7 +440,6 @@ func (_m *Manager) GetVolumeSnapshotter(name string) (volumesnapshotterv1.Volume
 		}
 	}
 
-	var r1 error
 	if rf, ok := ret.Get(1).(func(string) error); ok {
 		r1 = rf(name)
 	} else {
@@ -302,3 +448,17 @@ func (_m *Manager) GetVolumeSnapshotter(name string) (volumesnapshotterv1.Volume
 
 	return r0, r1
 }
+
+// NewManager creates a new instance of Manager. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewManager(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *Manager {
+	mock := &Manager{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/plugin/proto/itemblockaction/v1/ItemBlockAction.proto

@@ -0,0 +1,29 @@
+syntax = "proto3";
+package v1;
+option go_package = "github.com/vmware-tanzu/velero/pkg/plugin/generated/itemblockaction/v1";
+
+import "Shared.proto";
+
+
+service ItemBlockAction {
+    rpc AppliesTo(ItemBlockActionAppliesToRequest) returns (ItemBlockActionAppliesToResponse);
+    rpc GetRelatedItems(ItemBlockActionGetRelatedItemsRequest) returns (ItemBlockActionGetRelatedItemsResponse);
+}
+
+message ItemBlockActionAppliesToRequest {
+    string plugin = 1;
+}
+
+message ItemBlockActionAppliesToResponse {
+    generated.ResourceSelector ResourceSelector = 1;
+}
+
+message ItemBlockActionGetRelatedItemsRequest {
+    string plugin = 1;
+    bytes item = 2;
+    bytes backup = 3;
+}
+
+message ItemBlockActionGetRelatedItemsResponse {
+    repeated generated.ResourceIdentifier relatedItems = 1;
+}

pkg/plugin/velero/itemblockaction/v1/item_block_action.go

@@ -0,0 +1,46 @@
+/*
+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 v1
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+
+	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+// ItemBlockAction is an action that returns a list of related items that must be backed up
+// along with the current item (and not in a separate parallel backup thread).
+type ItemBlockAction interface {
+	// Name returns the name of this IBA. 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 ItemBlockAction's GetRelatedItems function will only be invoked on items that match the returned
+	// selector. A zero-valued ResourceSelector matches all resources.
+	AppliesTo() (velero.ResourceSelector, error)
+
+	// GetRelatedItems allows the ItemBlockAction to identify related items which must be backed up
+	// along with the current item. In many cases, these will be the same items that a related
+	// BackupItemAction's Execute method will return as additionalItems, but there may be differences.
+	// For example, items that are newly-created in the BIA Execute and don't yet exist at backup
+	// start will *not* be returned here.
+	GetRelatedItems(item runtime.Unstructured, backup *api.Backup) ([]velero.ResourceIdentifier, error)
+}

pkg/plugin/velero/mocks/itemblockaction/v1/ItemBlockAction.go

@@ -0,0 +1,122 @@
+/*
+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.
+*/
+// Code generated by mockery v2.43.2. DO NOT EDIT.
+
+package v1
+
+import (
+	mock "github.com/stretchr/testify/mock"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+
+	velero "github.com/vmware-tanzu/velero/pkg/plugin/velero"
+
+	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+// ItemBlockAction is an autogenerated mock type for the ItemBlockAction type
+type ItemBlockAction struct {
+	mock.Mock
+}
+
+// AppliesTo provides a mock function with given fields:
+func (_m *ItemBlockAction) AppliesTo() (velero.ResourceSelector, error) {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for AppliesTo")
+	}
+
+	var r0 velero.ResourceSelector
+	var r1 error
+	if rf, ok := ret.Get(0).(func() (velero.ResourceSelector, error)); ok {
+		return rf()
+	}
+	if rf, ok := ret.Get(0).(func() velero.ResourceSelector); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(velero.ResourceSelector)
+	}
+
+	if rf, ok := ret.Get(1).(func() error); ok {
+		r1 = rf()
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetRelatedItems provides a mock function with given fields: item, backup
+func (_m *ItemBlockAction) GetRelatedItems(item runtime.Unstructured, backup *velerov1.Backup) ([]velero.ResourceIdentifier, error) {
+	ret := _m.Called(item, backup)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetRelatedItems")
+	}
+
+	var r0 []velero.ResourceIdentifier
+	var r1 error
+	if rf, ok := ret.Get(0).(func(runtime.Unstructured, *velerov1.Backup) ([]velero.ResourceIdentifier, error)); ok {
+		return rf(item, backup)
+	}
+	if rf, ok := ret.Get(0).(func(runtime.Unstructured, *velerov1.Backup) []velero.ResourceIdentifier); ok {
+		r0 = rf(item, backup)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]velero.ResourceIdentifier)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(runtime.Unstructured, *velerov1.Backup) error); ok {
+		r1 = rf(item, backup)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Name provides a mock function with given fields:
+func (_m *ItemBlockAction) Name() string {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for Name")
+	}
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func() string); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	return r0
+}
+
+// NewItemBlockAction creates a new instance of ItemBlockAction. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewItemBlockAction(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *ItemBlockAction {
+	mock := &ItemBlockAction{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}