ItemSnapshotter plugin APIs. Addresses #3753 (#4077)

Added ItemSnapshotter.proto
Added item_snapshotter Go interface
Added framework components for item_snapshotter
Updated plugins doc with ItemSnapshotter info
Added SnapshotPhase to item_snapshotter.go
ProgressOutputOutput now includes a phase as well as an error string for problems that occured

Signed-off-by: Dave Smith-Uchida <dsmithuchida@vmware.com>

pkg/plugin/clientmgmt/client_builder.go

@@ -73,6 +73,7 @@ func (b *clientBuilder) clientConfig() *hcplugin.ClientConfig {
 			string(framework.PluginKindPluginLister):      &framework.PluginListerPlugin{},
 			string(framework.PluginKindRestoreItemAction): framework.NewRestoreItemActionPlugin(framework.ClientLogger(b.clientLogger)),
 			string(framework.PluginKindDeleteItemAction):  framework.NewDeleteItemActionPlugin(framework.ClientLogger(b.clientLogger)),
+			string(framework.PluginKindItemSnapshotter):   framework.NewItemSnapshotterPlugin(framework.ClientLogger(b.clientLogger)),
 		},
 		Logger: b.pluginLogger,
 		Cmd:    exec.Command(b.commandName, b.commandArgs...),

pkg/plugin/clientmgmt/client_builder_test.go

@@ -66,6 +66,7 @@ func TestClientConfig(t *testing.T) {
 			string(framework.PluginKindPluginLister):      &framework.PluginListerPlugin{},
 			string(framework.PluginKindRestoreItemAction): framework.NewRestoreItemActionPlugin(framework.ClientLogger(logger)),
 			string(framework.PluginKindDeleteItemAction):  framework.NewDeleteItemActionPlugin(framework.ClientLogger(logger)),
+			string(framework.PluginKindItemSnapshotter):   framework.NewItemSnapshotterPlugin(framework.ClientLogger(logger)),
 		},
 		Logger: cb.pluginLogger,
 		Cmd:    exec.Command(cb.commandName, cb.commandArgs...),

pkg/plugin/clientmgmt/manager.go

@@ -20,6 +20,8 @@ import (
 	"strings"
 	"sync"
 
+	v1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/item_snapshotter/v1"
+
 	"github.com/sirupsen/logrus"
 
 	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
@@ -52,6 +54,12 @@ type Manager interface {
 	// GetDeleteItemAction returns the delete item action plugin for name.
 	GetDeleteItemAction(name string) (velero.DeleteItemAction, error)
 
+	// GetItemSnapshotter returns the item snapshotter plugin for name
+	GetItemSnapshotter(name string) (v1.ItemSnapshotter, error)
+
+	// GetItemSnapshotters returns all item snapshotter plugins
+	GetItemSnapshotters() ([]v1.ItemSnapshotter, error)
+
 	// CleanupClients terminates all of the Manager's running plugin processes.
 	CleanupClients()
 }
@@ -256,6 +264,37 @@ func (m *manager) GetDeleteItemAction(name string) (velero.DeleteItemAction, err
 	return r, nil
 }
 
+func (m *manager) GetItemSnapshotter(name string) (v1.ItemSnapshotter, error) {
+	name = sanitizeName(name)
+
+	restartableProcess, err := m.getRestartableProcess(framework.PluginKindItemSnapshotter, name)
+	if err != nil {
+		return nil, err
+	}
+
+	r := newRestartableItemSnapshotter(name, restartableProcess)
+	return r, nil
+}
+
+func (m *manager) GetItemSnapshotters() ([]v1.ItemSnapshotter, error) {
+	list := m.registry.List(framework.PluginKindItemSnapshotter)
+
+	actions := make([]v1.ItemSnapshotter, 0, len(list))
+
+	for i := range list {
+		id := list[i]
+
+		r, err := m.GetItemSnapshotter(id.Name)
+		if err != nil {
+			return nil, err
+		}
+
+		actions = append(actions, r)
+	}
+
+	return actions, nil
+}
+
 // 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/restartable_item_snapshotter.go

@@ -0,0 +1,131 @@
+/*
+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 clientmgmt
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+
+	isv1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/item_snapshotter/v1"
+
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+type restartableItemSnapshotter struct {
+	key                 kindAndName
+	sharedPluginProcess RestartableProcess
+}
+
+// newRestartableItemSnapshotter returns a new newRestartableItemSnapshotter.
+func newRestartableItemSnapshotter(name string, sharedPluginProcess RestartableProcess) *restartableItemSnapshotter {
+	r := &restartableItemSnapshotter{
+		key:                 kindAndName{kind: framework.PluginKindItemSnapshotter, name: name},
+		sharedPluginProcess: sharedPluginProcess,
+	}
+	return r
+}
+
+// getItemSnapshotter returns the item snapshotter for this restartableItemSnapshotter. It does *not* restart the
+// plugin process.
+func (r *restartableItemSnapshotter) getItemSnapshotter() (isv1.ItemSnapshotter, error) {
+	plugin, err := r.sharedPluginProcess.getByKindAndName(r.key)
+	if err != nil {
+		return nil, err
+	}
+
+	itemSnapshotter, ok := plugin.(isv1.ItemSnapshotter)
+	if !ok {
+		return nil, errors.Errorf("%T is not an ItemSnapshotter!", plugin)
+	}
+
+	return itemSnapshotter, nil
+}
+
+// getDelegate restarts the plugin process (if needed) and returns the item snapshotter for this restartableItemSnapshotter.
+func (r *restartableItemSnapshotter) getDelegate() (isv1.ItemSnapshotter, error) {
+	if err := r.sharedPluginProcess.resetIfNeeded(); err != nil {
+		return nil, err
+	}
+
+	return r.getItemSnapshotter()
+}
+
+func (r *restartableItemSnapshotter) Init(config map[string]string) error {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return err
+	}
+
+	return delegate.Init(config)
+}
+
+// AppliesTo restarts the plugin's process if needed, then delegates the call.
+func (r *restartableItemSnapshotter) AppliesTo() (velero.ResourceSelector, error) {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return velero.ResourceSelector{}, err
+	}
+
+	return delegate.AppliesTo()
+}
+
+func (r *restartableItemSnapshotter) AlsoHandles(input *isv1.AlsoHandlesInput) ([]velero.ResourceIdentifier, error) {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return nil, err
+	}
+
+	return delegate.AlsoHandles(input)
+}
+
+func (r *restartableItemSnapshotter) SnapshotItem(ctx context.Context, input *isv1.SnapshotItemInput) (*isv1.SnapshotItemOutput, error) {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return nil, err
+	}
+
+	return delegate.SnapshotItem(ctx, input)
+}
+
+func (r *restartableItemSnapshotter) Progress(input *isv1.ProgressInput) (*isv1.ProgressOutput, error) {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return nil, err
+	}
+
+	return delegate.Progress(input)
+}
+
+func (r *restartableItemSnapshotter) DeleteSnapshot(ctx context.Context, input *isv1.DeleteSnapshotInput) error {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return err
+	}
+
+	return delegate.DeleteSnapshot(ctx, input)
+}
+
+func (r *restartableItemSnapshotter) CreateItemFromSnapshot(ctx context.Context, input *isv1.CreateItemInput) (*isv1.CreateItemOutput, error) {
+	delegate, err := r.getDelegate()
+	if err != nil {
+		return nil, err
+	}
+
+	return delegate.CreateItemFromSnapshot(ctx, input)
+}

pkg/plugin/clientmgmt/restartable_item_snapshotter_test.go

@@ -0,0 +1,233 @@
+/*
+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 clientmgmt
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	isv1 "github.com/vmware-tanzu/velero/pkg/plugin/velero/item_snapshotter/v1"
+
+	"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"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero/item_snapshotter/v1/mocks"
+
+	"github.com/vmware-tanzu/velero/pkg/plugin/framework"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+)
+
+func TestRestartableGetItemSnapshotter(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: "int is not an ItemSnapshotter!",
+		},
+		{
+			name:   "happy path",
+			plugin: new(mocks.ItemSnapshotter),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			p := new(mockRestartableProcess)
+			defer p.AssertExpectations(t)
+
+			name := "pvc"
+			key := kindAndName{kind: framework.PluginKindItemSnapshotter, name: name}
+			p.On("getByKindAndName", key).Return(tc.plugin, tc.getError)
+
+			r := newRestartableItemSnapshotter(name, p)
+			a, err := r.getItemSnapshotter()
+			if tc.expectedError != "" {
+				assert.EqualError(t, err, tc.expectedError)
+				return
+			}
+			require.NoError(t, err)
+
+			assert.Equal(t, tc.plugin, a)
+		})
+	}
+}
+
+func TestRestartableItemSnapshotterGetDelegate(t *testing.T) {
+	p := new(mockRestartableProcess)
+	defer p.AssertExpectations(t)
+
+	// Reset error
+	p.On("resetIfNeeded").Return(errors.Errorf("reset error")).Once()
+	name := "pvc"
+	r := newRestartableItemSnapshotter(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.ItemSnapshotter)
+	key := kindAndName{kind: framework.PluginKindItemSnapshotter, name: name}
+	p.On("getByKindAndName", key).Return(expected, nil)
+
+	a, err = r.getDelegate()
+	assert.NoError(t, err)
+	assert.Equal(t, expected, a)
+}
+
+func TestRestartableItemSnasphotterDelegatedFunctions(t *testing.T) {
+	b := new(v1.Backup)
+
+	pv := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"color": "blue",
+		},
+	}
+
+	sii := &isv1.SnapshotItemInput{
+		Item:   pv,
+		Params: nil,
+		Backup: b,
+	}
+
+	ctx := context.Background()
+
+	pvToReturn := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"color": "green",
+		},
+	}
+
+	additionalItems := []velero.ResourceIdentifier{
+		{
+			GroupResource: schema.GroupResource{Group: "velero.io", Resource: "backups"},
+		},
+	}
+
+	sio := &isv1.SnapshotItemOutput{
+		UpdatedItem:      pvToReturn,
+		SnapshotID:       "",
+		SnapshotMetadata: nil,
+		AdditionalItems:  additionalItems,
+		HandledItems:     nil,
+	}
+
+	cii := &isv1.CreateItemInput{
+		SnapshottedItem:  nil,
+		SnapshotID:       "",
+		ItemFromBackup:   nil,
+		SnapshotMetadata: nil,
+		Params:           nil,
+		Restore:          nil,
+	}
+
+	cio := &isv1.CreateItemOutput{
+		UpdatedItem:     nil,
+		AdditionalItems: nil,
+		SkipRestore:     false,
+	}
+
+	pi := &isv1.ProgressInput{
+		ItemID:     velero.ResourceIdentifier{},
+		SnapshotID: "",
+		Backup:     nil,
+	}
+	po := &isv1.ProgressOutput{
+		Phase:           isv1.SnapshotPhaseInProgress,
+		Err:             "",
+		ItemsCompleted:  0,
+		ItemsToComplete: 0,
+		Started:         time.Time{},
+		Updated:         time.Time{},
+	}
+	dsi := &isv1.DeleteSnapshotInput{
+		SnapshotID:       "",
+		ItemFromBackup:   nil,
+		SnapshotMetadata: nil,
+		Params:           nil,
+	}
+	runRestartableDelegateTests(
+		t,
+		framework.PluginKindItemSnapshotter,
+		func(key kindAndName, p RestartableProcess) interface{} {
+			return &restartableItemSnapshotter{
+				key:                 key,
+				sharedPluginProcess: p,
+			}
+		},
+		func() mockable {
+			return new(mocks.ItemSnapshotter)
+		},
+		restartableDelegateTest{
+			function:                "Init",
+			inputs:                  []interface{}{map[string]string{}},
+			expectedErrorOutputs:    []interface{}{errors.Errorf("reset error")},
+			expectedDelegateOutputs: []interface{}{errors.Errorf("delegate error")},
+		},
+		restartableDelegateTest{
+			function:                "AppliesTo",
+			inputs:                  []interface{}{},
+			expectedErrorOutputs:    []interface{}{velero.ResourceSelector{}, errors.Errorf("reset error")},
+			expectedDelegateOutputs: []interface{}{velero.ResourceSelector{IncludedNamespaces: []string{"a"}}, errors.Errorf("delegate error")},
+		},
+		restartableDelegateTest{
+			function:                "AlsoHandles",
+			inputs:                  []interface{}{&isv1.AlsoHandlesInput{}},
+			expectedErrorOutputs:    []interface{}{[]velero.ResourceIdentifier([]velero.ResourceIdentifier(nil)), errors.Errorf("reset error")},
+			expectedDelegateOutputs: []interface{}{[]velero.ResourceIdentifier([]velero.ResourceIdentifier(nil)), errors.Errorf("delegate error")},
+		},
+		restartableDelegateTest{
+			function:                "SnapshotItem",
+			inputs:                  []interface{}{ctx, sii},
+			expectedErrorOutputs:    []interface{}{nil, errors.Errorf("reset error")},
+			expectedDelegateOutputs: []interface{}{sio, errors.Errorf("delegate error")},
+		},
+		restartableDelegateTest{
+			function:                "CreateItemFromSnapshot",
+			inputs:                  []interface{}{ctx, cii},
+			expectedErrorOutputs:    []interface{}{nil, errors.Errorf("reset error")},
+			expectedDelegateOutputs: []interface{}{cio, errors.Errorf("delegate error")},
+		},
+		restartableDelegateTest{
+			function:                "Progress",
+			inputs:                  []interface{}{pi},
+			expectedErrorOutputs:    []interface{}{nil, errors.Errorf("reset error")},
+			expectedDelegateOutputs: []interface{}{po, errors.Errorf("delegate error")},
+		},
+		restartableDelegateTest{
+			function:                "DeleteSnapshot",
+			inputs:                  []interface{}{ctx, dsi},
+			expectedErrorOutputs:    []interface{}{errors.Errorf("reset error")},
+			expectedDelegateOutputs: []interface{}{errors.Errorf("delegate error")},
+		},
+	)
+}