Initial implementation of plugin version v2: Adding context object

pkg/plugin/framework/backup_item_action.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2019 the Velero contributors.
+Copyright 2021 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.

pkg/plugin/framework/backup_item_action_client.go

@@ -119,3 +119,51 @@ func (c *BackupItemActionGRPCClient) Execute(item runtime.Unstructured, backup *
 
 	return &updatedItem, additionalItems, nil
 }
+
+func (c *BackupItemActionGRPCClient) ExecuteV2(
+	ctx context.Context, item runtime.Unstructured, backup *api.Backup) (
+	runtime.Unstructured, []velero.ResourceIdentifier, error) {
+
+	itemJSON, err := json.Marshal(item.UnstructuredContent())
+	if err != nil {
+		return nil, nil, errors.WithStack(err)
+	}
+
+	backupJSON, err := json.Marshal(backup)
+	if err != nil {
+		return nil, nil, errors.WithStack(err)
+	}
+
+	req := &proto.ExecuteRequest{
+		Plugin: c.plugin,
+		Item:   itemJSON,
+		Backup: backupJSON,
+	}
+
+	res, err := c.grpcClient.Execute(ctx, req)
+	if err != nil {
+		return nil, nil, fromGRPCError(err)
+	}
+
+	var updatedItem unstructured.Unstructured
+	if err := json.Unmarshal(res.Item, &updatedItem); err != nil {
+		return nil, nil, errors.WithStack(err)
+	}
+
+	var additionalItems []velero.ResourceIdentifier
+
+	for _, itm := range res.AdditionalItems {
+		newItem := velero.ResourceIdentifier{
+			GroupResource: schema.GroupResource{
+				Group:    itm.Group,
+				Resource: itm.Resource,
+			},
+			Namespace: itm.Namespace,
+			Name:      itm.Name,
+		}
+
+		additionalItems = append(additionalItems, newItem)
+	}
+
+	return &updatedItem, additionalItems, nil
+}

pkg/plugin/framework/backup_item_action_server.go

@@ -26,6 +26,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	backupitemactionv2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/backupitemaction/v2"
 )
 
 // BackupItemActionGRPCServer implements the proto-generated BackupItemAction interface, and accepts
@@ -34,13 +35,13 @@ type BackupItemActionGRPCServer struct {
 	mux *serverMux
 }
 
-func (s *BackupItemActionGRPCServer) getImpl(name string) (velero.BackupItemAction, error) {
+func (s *BackupItemActionGRPCServer) getImpl(name string) (backupitemactionv2.BackupItemAction, error) {
 	impl, err := s.mux.getHandler(name)
 	if err != nil {
 		return nil, err
 	}
 
-	itemAction, ok := impl.(velero.BackupItemAction)
+	itemAction, ok := impl.(backupitemactionv2.BackupItemAction)
 	if !ok {
 		return nil, errors.Errorf("%T is not a backup item action", impl)
 	}
@@ -98,7 +99,7 @@ func (s *BackupItemActionGRPCServer) Execute(ctx context.Context, req *proto.Exe
 		return nil, newGRPCError(errors.WithStack(err))
 	}
 
-	updatedItem, additionalItems, err := impl.Execute(&item, &backup)
+	updatedItem, additionalItems, err := impl.ExecuteV2(ctx, &item, &backup)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}

pkg/plugin/framework/delete_item_action_client.go

@@ -25,9 +25,10 @@ import (
 
 	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	deleteitemactionv2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/deleteitemaction/v2"
 )
 
-var _ velero.DeleteItemAction = &DeleteItemActionGRPCClient{}
+var _ deleteitemactionv2.DeleteItemAction = &DeleteItemActionGRPCClient{}
 
 // NewDeleteItemActionPlugin constructs a DeleteItemActionPlugin.
 func NewDeleteItemActionPlugin(options ...PluginOption) *DeleteItemActionPlugin {
@@ -93,3 +94,28 @@ func (c *DeleteItemActionGRPCClient) Execute(input *velero.DeleteItemActionExecu
 
 	return nil
 }
+
+func (c *DeleteItemActionGRPCClient) ExecuteV2(ctx context.Context, input *velero.DeleteItemActionExecuteInput) error {
+	itemJSON, err := json.Marshal(input.Item.UnstructuredContent())
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	backupJSON, err := json.Marshal(input.Backup)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	req := &proto.DeleteItemActionExecuteRequest{
+		Plugin: c.plugin,
+		Item:   itemJSON,
+		Backup: backupJSON,
+	}
+
+	// First return item is just an empty struct no matter what.
+	if _, err = c.grpcClient.Execute(ctx, req); err != nil {
+		return fromGRPCError(err)
+	}
+
+	return nil
+}

pkg/plugin/framework/delete_item_action_server.go

@@ -26,6 +26,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	deleteitemactionv2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/deleteitemaction/v2"
 )
 
 // DeleteItemActionGRPCServer implements the proto-generated DeleteItemActionServer interface, and accepts
@@ -34,13 +35,13 @@ type DeleteItemActionGRPCServer struct {
 	mux *serverMux
 }
 
-func (s *DeleteItemActionGRPCServer) getImpl(name string) (velero.DeleteItemAction, error) {
+func (s *DeleteItemActionGRPCServer) getImpl(name string) (deleteitemactionv2.DeleteItemAction, error) {
 	impl, err := s.mux.getHandler(name)
 	if err != nil {
 		return nil, err
 	}
 
-	itemAction, ok := impl.(velero.DeleteItemAction)
+	itemAction, ok := impl.(deleteitemactionv2.DeleteItemAction)
 	if !ok {
 		return nil, errors.Errorf("%T is not a delete item action", impl)
 	}
@@ -76,7 +77,8 @@ func (s *DeleteItemActionGRPCServer) AppliesTo(ctx context.Context, req *proto.D
 	}, nil
 }
 
-func (s *DeleteItemActionGRPCServer) Execute(ctx context.Context, req *proto.DeleteItemActionExecuteRequest) (_ *proto.Empty, err error) {
+func (s *DeleteItemActionGRPCServer) Execute(
+	ctx context.Context, req *proto.DeleteItemActionExecuteRequest) (_ *proto.Empty, err error) {
 	defer func() {
 		if recoveredErr := handlePanic(recover()); recoveredErr != nil {
 			err = recoveredErr
@@ -101,7 +103,7 @@ func (s *DeleteItemActionGRPCServer) Execute(ctx context.Context, req *proto.Del
 		return nil, newGRPCError(errors.WithStack(err))
 	}
 
-	if err := impl.Execute(&velero.DeleteItemActionExecuteInput{
+	if err := impl.ExecuteV2(ctx, &velero.DeleteItemActionExecuteInput{
 		Item:   &item,
 		Backup: &backup,
 	}); err != nil {

pkg/plugin/framework/object_store_client.go

@@ -218,3 +218,161 @@ func (c *ObjectStoreGRPCClient) CreateSignedURL(bucket, key string, ttl time.Dur
 
 	return res.Url, nil
 }
+
+// Version 2
+// PutObjectV2 creates a new object using the data in body within the specified
+// object storage bucket with the given key.
+func (c *ObjectStoreGRPCClient) PutObjectV2(ctx context.Context, bucket, key string, body io.Reader) error {
+	stream, err := c.grpcClient.PutObject(ctx)
+	if err != nil {
+		return fromGRPCError(err)
+	}
+
+	// read from the provider io.Reader into chunks, and send each one over
+	// the gRPC stream
+	chunk := make([]byte, byteChunkSize)
+	for {
+		n, err := body.Read(chunk)
+		if err == io.EOF {
+			if _, resErr := stream.CloseAndRecv(); resErr != nil {
+				return fromGRPCError(resErr)
+			}
+			return nil
+		}
+		if err != nil {
+			stream.CloseSend()
+			return errors.WithStack(err)
+		}
+
+		if err := stream.Send(&proto.PutObjectRequest{Plugin: c.plugin, Bucket: bucket, Key: key, Body: chunk[0:n]}); err != nil {
+			return fromGRPCError(err)
+		}
+	}
+}
+
+// ObjectExistsV2 checks if there is an object with the given key in the object storage bucket.
+func (c *ObjectStoreGRPCClient) ObjectExistsV2(ctx context.Context, bucket, key string) (bool, error) {
+	req := &proto.ObjectExistsRequest{
+		Plugin: c.plugin,
+		Bucket: bucket,
+		Key:    key,
+	}
+
+	res, err := c.grpcClient.ObjectExists(ctx, req)
+	if err != nil {
+		return false, err
+	}
+
+	return res.Exists, nil
+}
+
+// GetObjectV2 retrieves the object with the given key from the specified
+// bucket in object storage.
+func (c *ObjectStoreGRPCClient) GetObjectV2(ctx context.Context, bucket, key string) (io.ReadCloser, error) {
+	req := &proto.GetObjectRequest{
+		Plugin: c.plugin,
+		Bucket: bucket,
+		Key:    key,
+	}
+
+	stream, err := c.grpcClient.GetObject(ctx, req)
+	if err != nil {
+		return nil, fromGRPCError(err)
+	}
+
+	receive := func() ([]byte, error) {
+		data, err := stream.Recv()
+		if err == io.EOF {
+			// we need to return io.EOF errors unwrapped so that
+			// calling code sees them as io.EOF and knows to stop
+			// reading.
+			return nil, err
+		}
+		if err != nil {
+			return nil, fromGRPCError(err)
+		}
+
+		return data.Data, nil
+	}
+
+	close := func() error {
+		if err := stream.CloseSend(); err != nil {
+			return fromGRPCError(err)
+		}
+		return nil
+	}
+
+	return &StreamReadCloser{receive: receive, close: close}, nil
+}
+
+// ListCommonPrefixesV2 gets a list of all object key prefixes that come
+// after the provided prefix and before the provided delimiter (this is
+// often used to simulate a directory hierarchy in object storage).
+func (c *ObjectStoreGRPCClient) ListCommonPrefixesV2(
+	ctx context.Context, bucket, prefix, delimiter string) ([]string, error) {
+	req := &proto.ListCommonPrefixesRequest{
+		Plugin:    c.plugin,
+		Bucket:    bucket,
+		Prefix:    prefix,
+		Delimiter: delimiter,
+	}
+
+	res, err := c.grpcClient.ListCommonPrefixes(ctx, req)
+	if err != nil {
+		return nil, fromGRPCError(err)
+	}
+
+	return res.Prefixes, nil
+}
+
+// ListObjectsV2 gets a list of all objects in bucket that have the same prefix.
+func (c *ObjectStoreGRPCClient) ListObjectsV2(
+	ctx context.Context, bucket, prefix string) ([]string, error) {
+	req := &proto.ListObjectsRequest{
+		Plugin: c.plugin,
+		Bucket: bucket,
+		Prefix: prefix,
+	}
+
+	res, err := c.grpcClient.ListObjects(ctx, req)
+	if err != nil {
+		return nil, fromGRPCError(err)
+	}
+
+	return res.Keys, nil
+}
+
+// DeleteObjectV2 removes object with the specified key from the given
+// bucket.
+func (c *ObjectStoreGRPCClient) DeleteObjectV2(
+	ctx context.Context, bucket, key string) error {
+	req := &proto.DeleteObjectRequest{
+		Plugin: c.plugin,
+		Bucket: bucket,
+		Key:    key,
+	}
+
+	if _, err := c.grpcClient.DeleteObject(ctx, req); err != nil {
+		return fromGRPCError(err)
+	}
+
+	return nil
+}
+
+// CreateSignedURLV2 creates a pre-signed URL for the given bucket and key that expires after ttl.
+func (c *ObjectStoreGRPCClient) CreateSignedURLV2(
+	ctx context.Context, bucket, key string, ttl time.Duration) (string, error) {
+	req := &proto.CreateSignedURLRequest{
+		Plugin: c.plugin,
+		Bucket: bucket,
+		Key:    key,
+		Ttl:    int64(ttl),
+	}
+
+	res, err := c.grpcClient.CreateSignedURL(ctx, req)
+	if err != nil {
+		return "", fromGRPCError(err)
+	}
+
+	return res.Url, nil
+}

pkg/plugin/framework/object_store_server.go

@@ -24,7 +24,7 @@ import (
 	"golang.org/x/net/context"
 
 	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
-	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	objectstorev2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/objectstore/v2"
 )
 
 // ObjectStoreGRPCServer implements the proto-generated ObjectStoreServer interface, and accepts
@@ -33,13 +33,13 @@ type ObjectStoreGRPCServer struct {
 	mux *serverMux
 }
 
-func (s *ObjectStoreGRPCServer) getImpl(name string) (velero.ObjectStore, error) {
+func (s *ObjectStoreGRPCServer) getImpl(name string) (objectstorev2.ObjectStore, error) {
 	impl, err := s.mux.getHandler(name)
 	if err != nil {
 		return nil, err
 	}
 
-	itemAction, ok := impl.(velero.ObjectStore)
+	itemAction, ok := impl.(objectstorev2.ObjectStore)
 	if !ok {
 		return nil, errors.Errorf("%T is not an object store", impl)
 	}
@@ -62,7 +62,7 @@ func (s *ObjectStoreGRPCServer) Init(ctx context.Context, req *proto.ObjectStore
 		return nil, newGRPCError(err)
 	}
 
-	if err := impl.Init(req.Config); err != nil {
+	if err := impl.InitV2(ctx, req.Config); err != nil {
 		return nil, newGRPCError(err)
 	}
 
@@ -141,7 +141,7 @@ func (s *ObjectStoreGRPCServer) ObjectExists(ctx context.Context, req *proto.Obj
 		return nil, newGRPCError(err)
 	}
 
-	exists, err := impl.ObjectExists(req.Bucket, req.Key)
+	exists, err := impl.ObjectExistsV2(ctx, req.Bucket, req.Key)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}
@@ -200,7 +200,7 @@ func (s *ObjectStoreGRPCServer) ListCommonPrefixes(ctx context.Context, req *pro
 		return nil, newGRPCError(err)
 	}
 
-	prefixes, err := impl.ListCommonPrefixes(req.Bucket, req.Prefix, req.Delimiter)
+	prefixes, err := impl.ListCommonPrefixesV2(ctx, req.Bucket, req.Prefix, req.Delimiter)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}
@@ -221,7 +221,7 @@ func (s *ObjectStoreGRPCServer) ListObjects(ctx context.Context, req *proto.List
 		return nil, newGRPCError(err)
 	}
 
-	keys, err := impl.ListObjects(req.Bucket, req.Prefix)
+	keys, err := impl.ListObjectsV2(ctx, req.Bucket, req.Prefix)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}
@@ -243,7 +243,7 @@ func (s *ObjectStoreGRPCServer) DeleteObject(ctx context.Context, req *proto.Del
 		return nil, newGRPCError(err)
 	}
 
-	if err := impl.DeleteObject(req.Bucket, req.Key); err != nil {
+	if err := impl.DeleteObjectV2(ctx, req.Bucket, req.Key); err != nil {
 		return nil, newGRPCError(err)
 	}
 
@@ -263,7 +263,7 @@ func (s *ObjectStoreGRPCServer) CreateSignedURL(ctx context.Context, req *proto.
 		return nil, newGRPCError(err)
 	}
 
-	url, err := impl.CreateSignedURL(req.Bucket, req.Key, time.Duration(req.Ttl))
+	url, err := impl.CreateSignedURLV2(ctx, req.Bucket, req.Key, time.Duration(req.Ttl))
 	if err != nil {
 		return nil, newGRPCError(err)
 	}

pkg/plugin/framework/plugin_kinds.go

@@ -45,6 +45,27 @@ const (
 	PluginKindPluginLister PluginKind = "PluginLister"
 )
 
+const (
+	// PluginKindObjectStoreV2 represents an object store plugin version 2.
+	PluginKindObjectStoreV2 PluginKind = "ObjectStoreV2"
+
+	// PluginKindVolumeSnapshotterV2 represents a volume snapshotter plugin version 2.
+	PluginKindVolumeSnapshotterV2 PluginKind = "VolumeSnapshotterV2"
+
+	// PluginKindBackupItemActionV2 represents a backup item action plugin version 2.
+	PluginKindBackupItemActionV2 PluginKind = "BackupItemActionV2"
+
+	// PluginKindRestoreItemActionV2 represents a restore item action plugin version 2.
+	PluginKindRestoreItemActionV2 PluginKind = "RestoreItemActionV2"
+
+	// PluginKindDeleteItemActionV2 represents a delete item action plugin version 2.
+	PluginKindDeleteItemActionV2 PluginKind = "DeleteItemActionV2"
+
+	// TODO: we may not need this
+	// PluginKindPluginListerV2 represents a plugin lister plugin version 2.
+	PluginKindPluginListerV2 PluginKind = "PluginListerV2"
+)
+
 // AllPluginKinds contains all the valid plugin kinds that Velero supports, excluding PluginLister because that is not a
 // kind that a developer would ever need to implement (it's handled by Velero and the Velero plugin library code).
 func AllPluginKinds() map[string]PluginKind {

pkg/plugin/framework/restore_item_action_client.go

@@ -27,9 +27,10 @@ import (
 
 	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	restoreitemactionv2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
 )
 
-var _ velero.RestoreItemAction = &RestoreItemActionGRPCClient{}
+var _ restoreitemactionv2.RestoreItemAction = &RestoreItemActionGRPCClient{}
 
 // NewRestoreItemActionPlugin constructs a RestoreItemActionPlugin.
 func NewRestoreItemActionPlugin(options ...PluginOption) *RestoreItemActionPlugin {
@@ -71,7 +72,14 @@ func (c *RestoreItemActionGRPCClient) AppliesTo() (velero.ResourceSelector, erro
 	}, nil
 }
 
-func (c *RestoreItemActionGRPCClient) Execute(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+func (c *RestoreItemActionGRPCClient) Execute(
+	input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+	return c.ExecuteV2(context.Background(), input)
+}
+
+func (c *RestoreItemActionGRPCClient) ExecuteV2(
+	ctx context.Context, input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+
 	itemJSON, err := json.Marshal(input.Item.UnstructuredContent())
 	if err != nil {
 		return nil, errors.WithStack(err)
@@ -94,7 +102,7 @@ func (c *RestoreItemActionGRPCClient) Execute(input *velero.RestoreItemActionExe
 		Restore:        restoreJSON,
 	}
 
-	res, err := c.grpcClient.Execute(context.Background(), req)
+	res, err := c.grpcClient.Execute(ctx, req)
 	if err != nil {
 		return nil, fromGRPCError(err)
 	}

pkg/plugin/framework/restore_item_action_server.go

@@ -26,6 +26,7 @@ import (
 	api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	restoreitemactionv2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
 )
 
 // RestoreItemActionGRPCServer implements the proto-generated RestoreItemActionServer interface, and accepts
@@ -34,13 +35,13 @@ type RestoreItemActionGRPCServer struct {
 	mux *serverMux
 }
 
-func (s *RestoreItemActionGRPCServer) getImpl(name string) (velero.RestoreItemAction, error) {
+func (s *RestoreItemActionGRPCServer) getImpl(name string) (restoreitemactionv2.RestoreItemAction, error) {
 	impl, err := s.mux.getHandler(name)
 	if err != nil {
 		return nil, err
 	}
 
-	itemAction, ok := impl.(velero.RestoreItemAction)
+	itemAction, ok := impl.(restoreitemactionv2.RestoreItemAction)
 	if !ok {
 		return nil, errors.Errorf("%T is not a restore item action", impl)
 	}
@@ -76,7 +77,9 @@ func (s *RestoreItemActionGRPCServer) AppliesTo(ctx context.Context, req *proto.
 	}, nil
 }
 
-func (s *RestoreItemActionGRPCServer) Execute(ctx context.Context, req *proto.RestoreItemActionExecuteRequest) (response *proto.RestoreItemActionExecuteResponse, err error) {
+func (s *RestoreItemActionGRPCServer) Execute(
+	ctx context.Context, req *proto.RestoreItemActionExecuteRequest) (response *proto.RestoreItemActionExecuteResponse, err error) {
+
 	defer func() {
 		if recoveredErr := handlePanic(recover()); recoveredErr != nil {
 			err = recoveredErr
@@ -106,11 +109,12 @@ func (s *RestoreItemActionGRPCServer) Execute(ctx context.Context, req *proto.Re
 		return nil, newGRPCError(errors.WithStack(err))
 	}
 
-	executeOutput, err := impl.Execute(&velero.RestoreItemActionExecuteInput{
-		Item:           &item,
-		ItemFromBackup: &itemFromBackup,
-		Restore:        &restoreObj,
-	})
+	executeOutput, err := impl.ExecuteV2(ctx,
+		&velero.RestoreItemActionExecuteInput{
+			Item:           &item,
+			ItemFromBackup: &itemFromBackup,
+			Restore:        &restoreObj,
+		})
 	if err != nil {
 		return nil, newGRPCError(err)
 	}

pkg/plugin/framework/server.go

@@ -74,21 +74,65 @@ type Server interface {
 	// RegisterDeleteItemActions registers multiple Delete item actions.
 	RegisterDeleteItemActions(map[string]HandlerInitializer) Server
 
+	// Version 2
+
+	// RegisterVolumeSnapshottersV2 registers multiple volume snapshotters.
+	RegisterVolumeSnapshottersV2(map[string]HandlerInitializer) Server
+
+	// RegisterObjectStoreV2 registers an object store. Accepted format
+	// for the plugin name is <DNS subdomain>/<non-empty name>.
+	RegisterObjectStoreV2(pluginName string, initializer HandlerInitializer) Server
+
+	// RegisterBackupItemActionV2 registers a backup item action. Accepted format
+	// for the plugin name is <DNS subdomain>/<non-empty name>.
+	RegisterBackupItemActionV2(pluginName string, initializer HandlerInitializer) Server
+
+	// RegisterBackupItemActionsV2 registers multiple backup item actions.
+	RegisterBackupItemActionsV2(map[string]HandlerInitializer) Server
+
+	// RegisterVolumeSnapshotterV2 registers a volume snapshotter. Accepted format
+	// for the plugin name is <DNS subdomain>/<non-empty name>.
+	RegisterVolumeSnapshotterV2(pluginName string, initializer HandlerInitializer) Server
+
+	// RegisterObjectStoresV2 registers multiple object stores.
+	RegisterObjectStoresV2(map[string]HandlerInitializer) Server
+
+	// RegisterRestoreItemActionV2 registers a restore item action. Accepted format
+	// for the plugin name is <DNS subdomain>/<non-empty name>.
+	RegisterRestoreItemActionV2(pluginName string, initializer HandlerInitializer) Server
+
+	// RegisterRestoreItemActionsV2 registers multiple restore item actions.
+	RegisterRestoreItemActionsV2(map[string]HandlerInitializer) Server
+
+	// RegisterDeleteItemActionV2 registers a delete item action. Accepted format
+	// for the plugin name is <DNS subdomain>/<non-empty name>.
+	RegisterDeleteItemActionV2(pluginName string, initializer HandlerInitializer) Server
+
+	// RegisterDeleteItemActionsV2 registers multiple Delete item actions.
+	RegisterDeleteItemActionsV2(map[string]HandlerInitializer) Server
+
 	// Server runs the plugin server.
 	Serve()
 }
 
 // server implements Server.
 type server struct {
-	log               *logrus.Logger
-	logLevelFlag      *logging.LevelFlag
-	flagSet           *pflag.FlagSet
-	featureSet        *veleroflag.StringArray
+	log          *logrus.Logger
+	logLevelFlag *logging.LevelFlag
+	flagSet      *pflag.FlagSet
+	featureSet   *veleroflag.StringArray
+	// Version 1
 	backupItemAction  *BackupItemActionPlugin
 	volumeSnapshotter *VolumeSnapshotterPlugin
 	objectStore       *ObjectStorePlugin
 	restoreItemAction *RestoreItemActionPlugin
 	deleteItemAction  *DeleteItemActionPlugin
+	// Version 2
+	backupItemActionV2  *BackupItemActionPlugin
+	volumeSnapshotterV2 *VolumeSnapshotterPlugin
+	objectStoreV2       *ObjectStorePlugin
+	restoreItemActionV2 *RestoreItemActionPlugin
+	deleteItemActionV2  *DeleteItemActionPlugin
 }
 
 // NewServer returns a new Server
@@ -177,6 +221,67 @@ func (s *server) RegisterDeleteItemActions(m map[string]HandlerInitializer) Serv
 	return s
 }
 
+// Version 2
+func (s *server) RegisterBackupItemActionV2(name string, initializer HandlerInitializer) Server {
+	s.backupItemActionV2.register(name, initializer)
+	return s
+}
+
+func (s *server) RegisterBackupItemActionsV2(m map[string]HandlerInitializer) Server {
+	for name := range m {
+		s.RegisterBackupItemActionV2(name, m[name])
+	}
+	return s
+}
+
+func (s *server) RegisterVolumeSnapshotterV2(name string, initializer HandlerInitializer) Server {
+	s.volumeSnapshotterV2.register(name, initializer)
+	return s
+}
+
+func (s *server) RegisterVolumeSnapshottersV2(m map[string]HandlerInitializer) Server {
+	for name := range m {
+		s.RegisterVolumeSnapshotterV2(name, m[name])
+	}
+	return s
+}
+
+func (s *server) RegisterObjectStoreV2(name string, initializer HandlerInitializer) Server {
+	s.objectStoreV2.register(name, initializer)
+	return s
+}
+
+func (s *server) RegisterObjectStoresV2(m map[string]HandlerInitializer) Server {
+	for name := range m {
+		s.RegisterObjectStoreV2(name, m[name])
+	}
+	return s
+}
+
+func (s *server) RegisterRestoreItemActionV2(name string, initializer HandlerInitializer) Server {
+	s.restoreItemActionV2.register(name, initializer)
+	return s
+}
+
+func (s *server) RegisterRestoreItemActionsV2(m map[string]HandlerInitializer) Server {
+	for name := range m {
+		s.RegisterRestoreItemActionV2(name, m[name])
+	}
+	return s
+}
+
+func (s *server) RegisterDeleteItemActionV2(name string, initializer HandlerInitializer) Server {
+	s.deleteItemActionV2.register(name, initializer)
+	return s
+}
+
+func (s *server) RegisterDeleteItemActionsV2(m map[string]HandlerInitializer) Server {
+	for name := range m {
+		s.RegisterDeleteItemActionV2(name, m[name])
+	}
+	return s
+}
+
 // getNames returns a list of PluginIdentifiers registered with plugin.
 func getNames(command string, kind PluginKind, plugin Interface) []PluginIdentifier {
 	var pluginIdentifiers []PluginIdentifier
@@ -206,6 +311,12 @@ func (s *server) Serve() {
 	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindObjectStore, s.objectStore)...)
 	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindRestoreItemAction, s.restoreItemAction)...)
 	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindDeleteItemAction, s.deleteItemAction)...)
+	// Version 2
+	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindBackupItemActionV2, s.backupItemActionV2)...)
+	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindVolumeSnapshotterV2, s.volumeSnapshotterV2)...)
+	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindObjectStoreV2, s.objectStoreV2)...)
+	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindRestoreItemActionV2, s.restoreItemActionV2)...)
+	pluginIdentifiers = append(pluginIdentifiers, getNames(command, PluginKindDeleteItemActionV2, s.deleteItemActionV2)...)
 
 	pluginLister := NewPluginLister(pluginIdentifiers...)
 
@@ -218,6 +329,14 @@ func (s *server) Serve() {
 			string(PluginKindPluginLister):      NewPluginListerPlugin(pluginLister),
 			string(PluginKindRestoreItemAction): s.restoreItemAction,
 			string(PluginKindDeleteItemAction):  s.deleteItemAction,
+			// Version 2
+			// TODO: check to see if need pluginLister for V2
+			// string(PluginKindPluginLister):      NewPluginListerPlugin(pluginLister),
+			string(PluginKindBackupItemActionV2):  s.backupItemActionV2,
+			string(PluginKindVolumeSnapshotterV2): s.volumeSnapshotterV2,
+			string(PluginKindObjectStoreV2):       s.objectStoreV2,
+			string(PluginKindRestoreItemActionV2): s.restoreItemActionV2,
+			string(PluginKindDeleteItemActionV2):  s.deleteItemActionV2,
 		},
 		GRPCServer: plugin.DefaultGRPCServer,
 	})

pkg/plugin/framework/volume_snapshotter_client.go

@@ -53,12 +53,19 @@ func newVolumeSnapshotterGRPCClient(base *clientBase, clientConn *grpc.ClientCon
 // configuration key-value pairs. It returns an error if the VolumeSnapshotter
 // cannot be initialized from the provided config.
 func (c *VolumeSnapshotterGRPCClient) Init(config map[string]string) error {
+	return c.InitV2(context.Background(), config)
+}
+
+// InitV2 prepares the VolumeSnapshotter for usage using the provided map of
+// configuration key-value pairs. It returns an error if the VolumeSnapshotter
+// cannot be initialized from the provided config.
+func (c *VolumeSnapshotterGRPCClient) InitV2(ctx context.Context, config map[string]string) error {
 	req := &proto.VolumeSnapshotterInitRequest{
 		Plugin: c.plugin,
 		Config: config,
 	}
 
-	if _, err := c.grpcClient.Init(context.Background(), req); err != nil {
+	if _, err := c.grpcClient.Init(ctx, req); err != nil {
 		return fromGRPCError(err)
 	}
 
@@ -67,7 +74,14 @@ func (c *VolumeSnapshotterGRPCClient) Init(config map[string]string) error {
 
 // CreateVolumeFromSnapshot creates a new block volume, initialized from the provided snapshot,
 // and with the specified type and IOPS (if using provisioned IOPS).
-func (c *VolumeSnapshotterGRPCClient) CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ string, iops *int64) (string, error) {
+func (c *VolumeSnapshotterGRPCClient) CreateVolumeFromSnapshot(
+	snapshotID, volumeType, volumeAZ string, iops *int64) (string, error) {
+	return c.CreateVolumeFromSnapshotV2(context.Background(), snapshotID, volumeType, volumeAZ, iops)
+}
+
+func (c *VolumeSnapshotterGRPCClient) CreateVolumeFromSnapshotV2(
+	ctx context.Context, snapshotID, volumeType, volumeAZ string, iops *int64) (string, error) {
+
 	req := &proto.CreateVolumeRequest{
 		Plugin:     c.plugin,
 		SnapshotID: snapshotID,
@@ -81,7 +95,7 @@ func (c *VolumeSnapshotterGRPCClient) CreateVolumeFromSnapshot(snapshotID, volum
 		req.Iops = *iops
 	}
 
-	res, err := c.grpcClient.CreateVolumeFromSnapshot(context.Background(), req)
+	res, err := c.grpcClient.CreateVolumeFromSnapshot(ctx, req)
 	if err != nil {
 		return "", fromGRPCError(err)
 	}
@@ -92,13 +106,19 @@ func (c *VolumeSnapshotterGRPCClient) CreateVolumeFromSnapshot(snapshotID, volum
 // GetVolumeInfo returns the type and IOPS (if using provisioned IOPS) for a specified block
 // volume.
 func (c *VolumeSnapshotterGRPCClient) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
+	return c.GetVolumeInfoV2(context.Background(), volumeID, volumeAZ)
+}
+
+func (c *VolumeSnapshotterGRPCClient) GetVolumeInfoV2(
+	ctx context.Context, volumeID, volumeAZ string) (string, *int64, error) {
+
 	req := &proto.GetVolumeInfoRequest{
 		Plugin:   c.plugin,
 		VolumeID: volumeID,
 		VolumeAZ: volumeAZ,
 	}
 
-	res, err := c.grpcClient.GetVolumeInfo(context.Background(), req)
+	res, err := c.grpcClient.GetVolumeInfo(ctx, req)
 	if err != nil {
 		return "", nil, fromGRPCError(err)
 	}
@@ -114,6 +134,11 @@ func (c *VolumeSnapshotterGRPCClient) GetVolumeInfo(volumeID, volumeAZ string) (
 // CreateSnapshot creates a snapshot of the specified block volume, and applies the provided
 // set of tags to the snapshot.
 func (c *VolumeSnapshotterGRPCClient) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
+	return c.CreateSnapshotV2(context.Background(), volumeID, volumeID, tags)
+}
+
+func (c *VolumeSnapshotterGRPCClient) CreateSnapshotV2(
+	ctx context.Context, volumeID, volumeAZ string, tags map[string]string) (string, error) {
 	req := &proto.CreateSnapshotRequest{
 		Plugin:   c.plugin,
 		VolumeID: volumeID,
@@ -121,7 +146,7 @@ func (c *VolumeSnapshotterGRPCClient) CreateSnapshot(volumeID, volumeAZ string,
 		Tags:     tags,
 	}
 
-	res, err := c.grpcClient.CreateSnapshot(context.Background(), req)
+	res, err := c.grpcClient.CreateSnapshot(ctx, req)
 	if err != nil {
 		return "", fromGRPCError(err)
 	}
@@ -131,12 +156,17 @@ func (c *VolumeSnapshotterGRPCClient) CreateSnapshot(volumeID, volumeAZ string,
 
 // DeleteSnapshot deletes the specified volume snapshot.
 func (c *VolumeSnapshotterGRPCClient) DeleteSnapshot(snapshotID string) error {
+	return c.DeleteSnapshotV2(context.Background(), snapshotID)
+}
+
+func (c *VolumeSnapshotterGRPCClient) DeleteSnapshotV2(
+	ctx context.Context, snapshotID string) error {
 	req := &proto.DeleteSnapshotRequest{
 		Plugin:     c.plugin,
 		SnapshotID: snapshotID,
 	}
 
-	if _, err := c.grpcClient.DeleteSnapshot(context.Background(), req); err != nil {
+	if _, err := c.grpcClient.DeleteSnapshot(ctx, req); err != nil {
 		return fromGRPCError(err)
 	}
 
@@ -144,6 +174,11 @@ func (c *VolumeSnapshotterGRPCClient) DeleteSnapshot(snapshotID string) error {
 }
 
 func (c *VolumeSnapshotterGRPCClient) GetVolumeID(pv runtime.Unstructured) (string, error) {
+	return c.GetVolumeIDV2(context.Background(), pv)
+}
+
+func (c *VolumeSnapshotterGRPCClient) GetVolumeIDV2(
+	ctx context.Context, pv runtime.Unstructured) (string, error) {
 	encodedPV, err := json.Marshal(pv.UnstructuredContent())
 	if err != nil {
 		return "", errors.WithStack(err)
@@ -154,7 +189,7 @@ func (c *VolumeSnapshotterGRPCClient) GetVolumeID(pv runtime.Unstructured) (stri
 		PersistentVolume: encodedPV,
 	}
 
-	resp, err := c.grpcClient.GetVolumeID(context.Background(), req)
+	resp, err := c.grpcClient.GetVolumeID(ctx, req)
 	if err != nil {
 		return "", fromGRPCError(err)
 	}
@@ -163,6 +198,11 @@ func (c *VolumeSnapshotterGRPCClient) GetVolumeID(pv runtime.Unstructured) (stri
 }
 
 func (c *VolumeSnapshotterGRPCClient) SetVolumeID(pv runtime.Unstructured, volumeID string) (runtime.Unstructured, error) {
+	return c.SetVolumeIDV2(context.Background(), pv, volumeID)
+}
+
+func (c *VolumeSnapshotterGRPCClient) SetVolumeIDV2(
+	ctx context.Context, pv runtime.Unstructured, volumeID string) (runtime.Unstructured, error) {
 	encodedPV, err := json.Marshal(pv.UnstructuredContent())
 	if err != nil {
 		return nil, errors.WithStack(err)
@@ -174,7 +214,7 @@ func (c *VolumeSnapshotterGRPCClient) SetVolumeID(pv runtime.Unstructured, volum
 		VolumeID:         volumeID,
 	}
 
-	resp, err := c.grpcClient.SetVolumeID(context.Background(), req)
+	resp, err := c.grpcClient.SetVolumeID(ctx, req)
 	if err != nil {
 		return nil, fromGRPCError(err)
 	}

pkg/plugin/framework/volume_snapshotter_server.go

@@ -24,7 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 
 	proto "github.com/vmware-tanzu/velero/pkg/plugin/generated"
-	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	volumesnapshotterv2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/volumesnapshotter/v2"
 )
 
 // VolumeSnapshotterGRPCServer implements the proto-generated VolumeSnapshotterServer interface, and accepts
@@ -33,13 +33,13 @@ type VolumeSnapshotterGRPCServer struct {
 	mux *serverMux
 }
 
-func (s *VolumeSnapshotterGRPCServer) getImpl(name string) (velero.VolumeSnapshotter, error) {
+func (s *VolumeSnapshotterGRPCServer) getImpl(name string) (volumesnapshotterv2.VolumeSnapshotter, error) {
 	impl, err := s.mux.getHandler(name)
 	if err != nil {
 		return nil, err
 	}
 
-	volumeSnapshotter, ok := impl.(velero.VolumeSnapshotter)
+	volumeSnapshotter, ok := impl.(volumesnapshotterv2.VolumeSnapshotter)
 	if !ok {
 		return nil, errors.Errorf("%T is not a volume snapshotter", impl)
 	}
@@ -62,7 +62,7 @@ func (s *VolumeSnapshotterGRPCServer) Init(ctx context.Context, req *proto.Volum
 		return nil, newGRPCError(err)
 	}
 
-	if err := impl.Init(req.Config); err != nil {
+	if err := impl.InitV2(ctx, req.Config); err != nil {
 		return nil, newGRPCError(err)
 	}
 
@@ -92,7 +92,7 @@ func (s *VolumeSnapshotterGRPCServer) CreateVolumeFromSnapshot(ctx context.Conte
 		iops = &req.Iops
 	}
 
-	volumeID, err := impl.CreateVolumeFromSnapshot(snapshotID, volumeType, volumeAZ, iops)
+	volumeID, err := impl.CreateVolumeFromSnapshotV2(ctx, snapshotID, volumeType, volumeAZ, iops)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}
@@ -114,7 +114,7 @@ func (s *VolumeSnapshotterGRPCServer) GetVolumeInfo(ctx context.Context, req *pr
 		return nil, newGRPCError(err)
 	}
 
-	volumeType, iops, err := impl.GetVolumeInfo(req.VolumeID, req.VolumeAZ)
+	volumeType, iops, err := impl.GetVolumeInfoV2(ctx, req.VolumeID, req.VolumeAZ)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}
@@ -144,7 +144,7 @@ func (s *VolumeSnapshotterGRPCServer) CreateSnapshot(ctx context.Context, req *p
 		return nil, newGRPCError(err)
 	}
 
-	snapshotID, err := impl.CreateSnapshot(req.VolumeID, req.VolumeAZ, req.Tags)
+	snapshotID, err := impl.CreateSnapshotV2(ctx, req.VolumeID, req.VolumeAZ, req.Tags)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}
@@ -165,7 +165,7 @@ func (s *VolumeSnapshotterGRPCServer) DeleteSnapshot(ctx context.Context, req *p
 		return nil, newGRPCError(err)
 	}
 
-	if err := impl.DeleteSnapshot(req.SnapshotID); err != nil {
+	if err := impl.DeleteSnapshotV2(ctx, req.SnapshotID); err != nil {
 		return nil, newGRPCError(err)
 	}
 
@@ -190,7 +190,7 @@ func (s *VolumeSnapshotterGRPCServer) GetVolumeID(ctx context.Context, req *prot
 		return nil, newGRPCError(errors.WithStack(err))
 	}
 
-	volumeID, err := impl.GetVolumeID(&pv)
+	volumeID, err := impl.GetVolumeIDV2(ctx, &pv)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}
@@ -215,7 +215,7 @@ func (s *VolumeSnapshotterGRPCServer) SetVolumeID(ctx context.Context, req *prot
 		return nil, newGRPCError(errors.WithStack(err))
 	}
 
-	updatedPV, err := impl.SetVolumeID(&pv, req.VolumeID)
+	updatedPV, err := impl.SetVolumeIDV2(ctx, &pv, req.VolumeID)
 	if err != nil {
 		return nil, newGRPCError(err)
 	}