Sync CSI API objects during backup sync (#2496)

* add changelog

Signed-off-by: Ashish Amarnath <ashisham@vmware.com>

* Sync CSI API volumesnapshotcontents during backup sync

Signed-off-by: Ashish Amarnath <ashisham@vmware.com>

changelogs/unreleased/2496-ashish-amarnath

@@ -0,0 +1 @@
+sync backups' CSI API objects into the cluster as part of the backup sync controller

pkg/cmd/server/server.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2017, 2019 the Velero contributors.
+Copyright 2017, 2019, 2020 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.
@@ -615,6 +615,8 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			s.sharedInformerFactory.Velero().V1().BackupStorageLocations().Lister(),
 			s.config.backupSyncPeriod,
 			s.namespace,
+			s.csiSnapshotClient,
+			s.kubeClient,
 			s.config.defaultBackupLocation,
 			newPluginManager,
 			s.logger,

pkg/controller/backup_sync_controller.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2017 the Velero contributors.
+Copyright 2017, 2020 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.
@@ -20,14 +20,17 @@ import (
 	"encoding/json"
 	"time"
 
+	snapshotterClientSet "github.com/kubernetes-csi/external-snapshotter/v2/pkg/client/clientset/versioned"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	kuberrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/client-go/kubernetes"
 
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/features"
 	velerov1client "github.com/vmware-tanzu/velero/pkg/generated/clientset/versioned/typed/velero/v1"
 	velerov1listers "github.com/vmware-tanzu/velero/pkg/generated/listers/velero/v1"
 	"github.com/vmware-tanzu/velero/pkg/label"
@@ -42,6 +45,8 @@ type backupSyncController struct {
 	backupLocationClient        velerov1client.BackupStorageLocationsGetter
 	podVolumeBackupClient       velerov1client.PodVolumeBackupsGetter
 	backupLister                velerov1listers.BackupLister
+	csiSnapshotClient           *snapshotterClientSet.Clientset
+	kubeClient                  kubernetes.Interface
 	backupStorageLocationLister velerov1listers.BackupStorageLocationLister
 	namespace                   string
 	defaultBackupLocation       string
@@ -58,6 +63,8 @@ func NewBackupSyncController(
 	backupStorageLocationLister velerov1listers.BackupStorageLocationLister,
 	syncPeriod time.Duration,
 	namespace string,
+	csiSnapshotClient *snapshotterClientSet.Clientset,
+	kubeClient kubernetes.Interface,
 	defaultBackupLocation string,
 	newPluginManager func(logrus.FieldLogger) clientmgmt.Manager,
 	logger logrus.FieldLogger,
@@ -77,6 +84,8 @@ func NewBackupSyncController(
 		defaultBackupSyncPeriod:     syncPeriod,
 		backupLister:                backupLister,
 		backupStorageLocationLister: backupStorageLocationLister,
+		csiSnapshotClient:           csiSnapshotClient,
+		kubeClient:                  kubeClient,
 
 		// use variables to refer to these functions so they can be
 		// replaced with fakes for testing.
@@ -262,6 +271,34 @@ func (c *backupSyncController) run() {
 					log.Debug("Synced pod volume backup into cluster")
 				}
 			}
+
+			if features.IsEnabled(velerov1api.CSIFeatureFlag) {
+				// we are syncing these objects only to ensure that the storage snapshots are cleaned up
+				// on backup deletion or expiry.
+				log.Info("Syncing CSI volumesnapshotcontents in backup")
+				snapConts, err := backupStore.GetCSIVolumeSnapshotContents(backupName)
+				if err != nil {
+					log.WithError(errors.WithStack(err)).Error("Error getting CSI volumesnapshotcontents for this backup from backup store")
+					continue
+				}
+
+				log.Infof("Syncing %d CSI volumesnapshotcontents in backup", len(snapConts))
+				for _, snapCont := range snapConts {
+					// TODO: Reset ResourceVersion prior to persisting VolumeSnapshotContents
+					snapCont.ResourceVersion = ""
+					created, err := c.csiSnapshotClient.SnapshotV1beta1().VolumeSnapshotContents().Create(snapCont)
+					switch {
+					case err != nil && kuberrs.IsAlreadyExists(err):
+						log.Debugf("volumesnapshotcontent %s already exists in cluster", snapCont.Name)
+						continue
+					case err != nil && !kuberrs.IsAlreadyExists(err):
+						log.WithError(errors.WithStack(err)).Errorf("Error syncing volumesnapshotcontent %s into cluster", snapCont.Name)
+						continue
+					default:
+						log.Infof("Created CSI volumesnapshotcontent %s", created.Name)
+					}
+				}
+			}
 		}
 
 		c.deleteOrphanedBackups(location.Name, backupStoreBackups, log)

pkg/controller/backup_sync_controller_test.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2017 the Velero contributors.
+Copyright 2017, 2020 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.
@@ -345,6 +345,8 @@ func TestBackupSyncControllerRun(t *testing.T) {
 				sharedInformers.Velero().V1().BackupStorageLocations().Lister(),
 				time.Duration(0),
 				test.namespace,
+				nil, // csiSnapshotClient
+				nil, // kubeClient
 				"",
 				func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
 				velerotest.NewLogger(),
@@ -568,6 +570,8 @@ func TestDeleteOrphanedBackups(t *testing.T) {
 				sharedInformers.Velero().V1().BackupStorageLocations().Lister(),
 				time.Duration(0),
 				test.namespace,
+				nil, // csiSnapshotClient
+				nil, // kubeClient
 				"",
 				nil, // new plugin manager func
 				velerotest.NewLogger(),
@@ -659,6 +663,8 @@ func TestStorageLabelsInDeleteOrphanedBackups(t *testing.T) {
 				sharedInformers.Velero().V1().BackupStorageLocations().Lister(),
 				time.Duration(0),
 				test.namespace,
+				nil, // csiSnapshotClient
+				nil, // kubeClient
 				"",
 				nil, // new plugin manager func
 				velerotest.NewLogger(),

pkg/persistence/mocks/backup_store.go

@@ -1,12 +1,32 @@
-// Code generated by mockery v1.0.0. DO NOT EDIT.
+/*
+Copyright 2020 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 mocks
 
-import io "io"
-import mock "github.com/stretchr/testify/mock"
-import persistence "github.com/vmware-tanzu/velero/pkg/persistence"
-import v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
-import volume "github.com/vmware-tanzu/velero/pkg/volume"
+import (
+	io "io"
+
+	mock "github.com/stretchr/testify/mock"
+
+	snapshotv1beta1api "github.com/kubernetes-csi/external-snapshotter/v2/pkg/apis/volumesnapshot/v1beta1"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	persistence "github.com/vmware-tanzu/velero/pkg/persistence"
+	volume "github.com/vmware-tanzu/velero/pkg/volume"
+)
 
 // BackupStore is an autogenerated mock type for the BackupStore type
 type BackupStore struct {
@@ -253,3 +273,13 @@ func (_m *BackupStore) PutRestoreResults(backup string, restore string, results
 
 	return r0
 }
+
+func (_m *BackupStore) GetCSIVolumeSnapshots(backup string) ([]*snapshotv1beta1api.VolumeSnapshot, error) {
+	panic("Not implemented")
+	return nil, nil
+}
+
+func (_m *BackupStore) GetCSIVolumeSnapshotContents(backup string) ([]*snapshotv1beta1api.VolumeSnapshotContent, error) {
+	panic("Not implemented")
+	return nil, nil
+}

pkg/persistence/object_store.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2018 the Velero contributors.
+Copyright 2018, 2020 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.
@@ -24,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	snapshotv1beta1api "github.com/kubernetes-csi/external-snapshotter/v2/pkg/apis/volumesnapshot/v1beta1"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
@@ -58,7 +59,8 @@ type BackupStore interface {
 	GetBackupVolumeSnapshots(name string) ([]*volume.Snapshot, error)
 	GetPodVolumeBackups(name string) ([]*velerov1api.PodVolumeBackup, error)
 	GetBackupContents(name string) (io.ReadCloser, error)
-	// TODO(nrb-csi): Any Get methods relevant to the CSI VolumeSnapshots should be added with the client-side PRs.
+	GetCSIVolumeSnapshots(name string) ([]*snapshotv1beta1api.VolumeSnapshot, error)
+	GetCSIVolumeSnapshotContents(name string) ([]*snapshotv1beta1api.VolumeSnapshotContent, error)
 
 	// BackupExists checks if the backup metadata file exists in object storage.
 	BackupExists(bucket, backupName string) (bool, error)
@@ -324,6 +326,42 @@ func decode(jsongzReader io.Reader, into interface{}) error {
 	return nil
 }
 
+func (s *objectBackupStore) GetCSIVolumeSnapshots(name string) ([]*snapshotv1beta1api.VolumeSnapshot, error) {
+	res, err := tryGet(s.objectStore, s.bucket, s.layout.getCSIVolumeSnapshotKey(name))
+	if err != nil {
+		return nil, err
+	}
+	if res == nil {
+		// this indicates that the no CSI volumesnapshots were prensent in the backup
+		return nil, nil
+	}
+	defer res.Close()
+
+	var csiSnaps []*snapshotv1beta1api.VolumeSnapshot
+	if err := decode(res, &csiSnaps); err != nil {
+		return nil, err
+	}
+	return csiSnaps, nil
+}
+
+func (s *objectBackupStore) GetCSIVolumeSnapshotContents(name string) ([]*snapshotv1beta1api.VolumeSnapshotContent, error) {
+	res, err := tryGet(s.objectStore, s.bucket, s.layout.getCSIVolumeSnapshotContentsKey(name))
+	if err != nil {
+		return nil, err
+	}
+	if res == nil {
+		// this indicates that the no CSI volumesnapshotcontents were prensent in the backup
+		return nil, nil
+	}
+	defer res.Close()
+
+	var snapConts []*snapshotv1beta1api.VolumeSnapshotContent
+	if err := decode(res, &snapConts); err != nil {
+		return nil, err
+	}
+	return snapConts, nil
+}
+
 func (s *objectBackupStore) GetPodVolumeBackups(name string) ([]*velerov1api.PodVolumeBackup, error) {
 	// if the podvolumebackups file doesn't exist, we don't want to return an error, since
 	// a legacy backup or a backup with no pod volume backups would not have this file, so