Issue 6693: partially fail restore if CSI snapshot is involved but CSI feature is not ready

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

pkg/cmd/server/server.go

@@ -267,7 +267,7 @@ type server struct {
 	mgr                   manager.Manager
 	credentialFileStore   credentials.FileStore
 	credentialSecretStore credentials.SecretStore
-	featureVerifier       *features.Verifier
+	featureVerifier       features.Verifier
 }
 
 func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*server, error) {

pkg/features/mocks/PluginFinder.go

@@ -0,0 +1,43 @@
+// Code generated by mockery v2.22.1. DO NOT EDIT.
+
+package mocks
+
+import (
+	common "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// PluginFinder is an autogenerated mock type for the PluginFinder type
+type PluginFinder struct {
+	mock.Mock
+}
+
+// Find provides a mock function with given fields: kind, name
+func (_m *PluginFinder) Find(kind common.PluginKind, name string) bool {
+	ret := _m.Called(kind, name)
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(common.PluginKind, string) bool); ok {
+		r0 = rf(kind, name)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewPluginFinder interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewPluginFinder creates a new instance of PluginFinder. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewPluginFinder(t mockConstructorTestingTNewPluginFinder) *PluginFinder {
+	mock := &PluginFinder{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/features/mocks/Verifier.go

@@ -0,0 +1,49 @@
+// Code generated by mockery v2.22.1. DO NOT EDIT.
+
+package mocks
+
+import mock "github.com/stretchr/testify/mock"
+
+// Verifier is an autogenerated mock type for the Verifier type
+type Verifier struct {
+	mock.Mock
+}
+
+// Verify provides a mock function with given fields: name
+func (_m *Verifier) Verify(name string) (bool, error) {
+	ret := _m.Called(name)
+
+	var r0 bool
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (bool, error)); ok {
+		return rf(name)
+	}
+	if rf, ok := ret.Get(0).(func(string) bool); ok {
+		r0 = rf(name)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(name)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+type mockConstructorTestingTNewVerifier interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewVerifier creates a new instance of Verifier. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewVerifier(t mockConstructorTestingTNewVerifier) *Verifier {
+	mock := &Verifier{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

pkg/features/verify.go

@@ -27,17 +27,21 @@ type PluginFinder interface {
 	Find(kind common.PluginKind, name string) bool
 }
 
-type Verifier struct {
+type Verifier interface {
+	Verify(name string) (bool, error)
+}
+
+type verifier struct {
 	finder PluginFinder
 }
 
-func NewVerifier(finder PluginFinder) *Verifier {
-	return &Verifier{
+func NewVerifier(finder PluginFinder) Verifier {
+	return &verifier{
 		finder: finder,
 	}
 }
 
-func (v *Verifier) Verify(name string) (bool, error) {
+func (v *verifier) Verify(name string) (bool, error) {
 	enabled := IsEnabled(name)
 
 	switch name {

pkg/features/verify_test.go

@@ -0,0 +1,61 @@
+/*
+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 features
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+
+	findermocks "github.com/vmware-tanzu/velero/pkg/features/mocks"
+)
+
+func TestVerify(t *testing.T) {
+	NewFeatureFlagSet()
+	verifier := verifier{}
+
+	finder := new(findermocks.PluginFinder)
+	finder.On("Find", mock.Anything, mock.Anything).Return(false)
+	verifier.finder = finder
+	ready, err := verifier.Verify("EnableCSI")
+	assert.Equal(t, false, ready)
+	assert.Nil(t, err)
+
+	finder = new(findermocks.PluginFinder)
+	finder.On("Find", mock.Anything, mock.Anything).Return(true)
+	verifier.finder = finder
+	ready, err = verifier.Verify("EnableCSI")
+	assert.Equal(t, false, ready)
+	assert.EqualError(t, err, "CSI plugins are registered, but the EnableCSI feature is not enabled")
+
+	Enable("EnableCSI")
+	finder = new(findermocks.PluginFinder)
+	finder.On("Find", mock.Anything, mock.Anything).Return(false)
+	verifier.finder = finder
+	ready, err = verifier.Verify("EnableCSI")
+	assert.Equal(t, false, ready)
+	assert.EqualError(t, err, "CSI feature is enabled, but CSI plugins are not registered")
+
+	Enable("EnableCSI")
+	finder = new(findermocks.PluginFinder)
+	finder.On("Find", mock.Anything, mock.Anything).Return(true)
+	verifier.finder = finder
+	ready, err = verifier.Verify("EnableCSI")
+	assert.Equal(t, true, ready)
+	assert.Nil(t, err)
+}

pkg/restore/restore.go

@@ -115,7 +115,7 @@ type kubernetesRestorer struct {
 	podGetter                  cache.Getter
 	credentialFileStore        credentials.FileStore
 	kbClient                   crclient.Client
-	featureVerifier            *features.Verifier
+	featureVerifier            features.Verifier
 }
 
 // NewKubernetesRestorer creates a new kubernetesRestorer.
@@ -133,7 +133,7 @@ func NewKubernetesRestorer(
 	podGetter cache.Getter,
 	credentialStore credentials.FileStore,
 	kbClient crclient.Client,
-	featureVerifier *features.Verifier,
+	featureVerifier features.Verifier,
 ) (Restorer, error) {
 	return &kubernetesRestorer{
 		discoveryHelper:            discoveryHelper,
@@ -376,7 +376,7 @@ type restoreContext struct {
 	itemOperationsList             *[]*itemoperation.RestoreOperation
 	resourceModifiers              *resourcemodifiers.ResourceModifiers
 	disableInformerCache           bool
-	featureVerifier                *features.Verifier
+	featureVerifier                features.Verifier
 }
 
 type resourceClientKey struct {

pkg/restore/restore_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	corev1api "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -46,6 +47,7 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/builder"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/discovery"
+	verifiermocks "github.com/vmware-tanzu/velero/pkg/features/mocks"
 	"github.com/vmware-tanzu/velero/pkg/itemoperation"
 	"github.com/vmware-tanzu/velero/pkg/kuberesource"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
@@ -2272,6 +2274,7 @@ func TestRestorePersistentVolumes(t *testing.T) {
 		want                    []*test.APIResource
 		wantError               bool
 		wantWarning             bool
+		csiFeatureVerifierErr   string
 	}{
 		{
 			name:    "when a PV with a reclaim policy of delete has no snapshot and does not exist in-cluster, it does not get restored, and its PVC gets reset for dynamic provisioning",
@@ -2966,6 +2969,47 @@ func TestRestorePersistentVolumes(t *testing.T) {
 			},
 			want: []*test.APIResource{},
 		},
+		{
+			name:    "when a PV has a CSI VolumeSnapshot, but CSI modules are not ready, the PV is not restored",
+			restore: defaultRestore().Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("persistentvolumes",
+					builder.ForPersistentVolume("pv-1").
+						ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).
+						ClaimRef("velero", testPVCName).
+						Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.PVs(),
+				test.PVCs(),
+			},
+			csiVolumeSnapshots: []*snapshotv1api.VolumeSnapshot{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "velero",
+						Name:      "test",
+					},
+					Spec: snapshotv1api.VolumeSnapshotSpec{
+						Source: snapshotv1api.VolumeSnapshotSource{
+							PersistentVolumeClaimName: &testPVCName,
+						},
+					},
+				},
+			},
+			volumeSnapshotLocations: []*velerov1api.VolumeSnapshotLocation{
+				builder.ForVolumeSnapshotLocation(velerov1api.DefaultNamespace, "default").Provider("provider-1").Result(),
+			},
+			volumeSnapshotterGetter: map[string]vsv1.VolumeSnapshotter{
+				"provider-1": &volumeSnapshotter{
+					snapshotVolumes: map[string]string{"snapshot-1": "new-volume"},
+				},
+			},
+			want:                  []*test.APIResource{},
+			csiFeatureVerifierErr: "fake-feature-check-error",
+			wantError:             true,
+		},
 		{
 			name:    "when a PV with a reclaim policy of retain has a DataUpload result CM and does not exist in-cluster, the PV is not restored",
 			restore: defaultRestore().ObjectMeta(builder.WithUID("fakeUID")).Result(),
@@ -2993,11 +3037,45 @@ func TestRestorePersistentVolumes(t *testing.T) {
 			},
 			dataUploadResult: builder.ForConfigMap("velero", "test").ObjectMeta(builder.WithLabelsMap(map[string]string{
 				velerov1api.RestoreUIDLabel:       "fakeUID",
-				velerov1api.PVCNamespaceNameLabel: "velero/testPVC",
+				velerov1api.PVCNamespaceNameLabel: "velero.testPVC",
 				velerov1api.ResourceUsageLabel:    string(velerov1api.VeleroResourceUsageDataUploadResult),
 			})).Result(),
 			want: []*test.APIResource{},
 		},
+		{
+			name:    "when a PV has a DataUpload result CM, but CSI modules are not ready, the PV is not restored",
+			restore: defaultRestore().ObjectMeta(builder.WithUID("fakeUID")).Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("persistentvolumes",
+					builder.ForPersistentVolume("pv-1").
+						ReclaimPolicy(corev1api.PersistentVolumeReclaimRetain).
+						ClaimRef("velero", testPVCName).
+						Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.PVs(),
+				test.PVCs(),
+				test.ConfigMaps(),
+			},
+			volumeSnapshotLocations: []*velerov1api.VolumeSnapshotLocation{
+				builder.ForVolumeSnapshotLocation(velerov1api.DefaultNamespace, "default").Provider("provider-1").Result(),
+			},
+			volumeSnapshotterGetter: map[string]vsv1.VolumeSnapshotter{
+				"provider-1": &volumeSnapshotter{
+					snapshotVolumes: map[string]string{"snapshot-1": "new-volume"},
+				},
+			},
+			dataUploadResult: builder.ForConfigMap("velero", "test").ObjectMeta(builder.WithLabelsMap(map[string]string{
+				velerov1api.RestoreUIDLabel:       "fakeUID",
+				velerov1api.PVCNamespaceNameLabel: "velero.testPVC",
+				velerov1api.ResourceUsageLabel:    string(velerov1api.VeleroResourceUsageDataUploadResult),
+			})).Result(),
+			want:                  []*test.APIResource{},
+			csiFeatureVerifierErr: "fake-feature-check-error",
+			wantError:             true,
+		},
 	}
 
 	for _, tc := range tests {
@@ -3009,6 +3087,14 @@ func TestRestorePersistentVolumes(t *testing.T) {
 				return renamed, nil
 			}
 
+			verifierMock := new(verifiermocks.Verifier)
+			if tc.csiFeatureVerifierErr != "" {
+				verifierMock.On("Verify", mock.Anything, mock.Anything).Return(false, errors.New(tc.csiFeatureVerifierErr))
+			} else {
+				verifierMock.On("Verify", mock.Anything, mock.Anything).Return(true, nil)
+			}
+			h.restorer.featureVerifier = verifierMock
+
 			// set up the VolumeSnapshotLocation client and add test data to it
 			for _, vsl := range tc.volumeSnapshotLocations {
 				require.NoError(t, h.restorer.kbClient.Create(context.Background(), vsl))