update restore process using snapshot locations

Signed-off-by: Wayne Witzel III <wayne@riotousliving.com>

Author: wwitzel3

pkg/backup/item_backupper.go

@@ -459,7 +459,7 @@ func (ib *defaultItemBackupper) takePVSnapshot(obj runtime.Unstructured, log log
 	}
 
 	log.Info("Snapshotting PersistentVolume")
-	snapshot := volumeSnapshot(ib.backupRequest.Backup, volumeID, volumeType, pvFailureDomainZone, location, iops)
+	snapshot := volumeSnapshot(ib.backupRequest.Backup, metadata.GetName(), volumeID, volumeType, pvFailureDomainZone, location, iops)
 
 	var errs []error
 	snapshotID, err := blockStore.CreateSnapshot(snapshot.Spec.ProviderVolumeID, snapshot.Spec.VolumeAZ, tags)
@@ -477,16 +477,17 @@ func (ib *defaultItemBackupper) takePVSnapshot(obj runtime.Unstructured, log log
 	return kubeerrs.NewAggregate(errs)
 }
 
-func volumeSnapshot(backup *api.Backup, volumeID, volumeType, az, location string, iops *int64) *volume.Snapshot {
+func volumeSnapshot(backup *api.Backup, volumeName, volumeID, volumeType, az, location string, iops *int64) *volume.Snapshot {
 	return &volume.Snapshot{
 		Spec: volume.SnapshotSpec{
-			BackupName:       backup.Name,
-			BackupUID:        string(backup.UID),
-			Location:         location,
-			ProviderVolumeID: volumeID,
-			VolumeType:       volumeType,
-			VolumeAZ:         az,
-			VolumeIOPS:       iops,
+			BackupName:           backup.Name,
+			BackupUID:            string(backup.UID),
+			Location:             location,
+			PersistentVolumeName: volumeName,
+			ProviderVolumeID:     volumeID,
+			VolumeType:           volumeType,
+			VolumeAZ:             az,
+			VolumeIOPS:           iops,
 		},
 		Status: volume.SnapshotStatus{
 			Phase: volume.SnapshotPhaseNew,

pkg/cmd/server/server.go

@@ -717,9 +717,7 @@ func (s *server) runControllers(config *api.Config, defaultVolumeSnapshotLocatio
 	restorer, err := restore.NewKubernetesRestorer(
 		s.discoveryHelper,
 		client.NewDynamicFactory(s.dynamicClient),
-		nil,
 		s.config.restoreResourcePriorities,
-		s.arkClient.ArkV1(),
 		s.kubeClient.CoreV1().Namespaces(),
 		s.resticManager,
 		s.config.podVolumeOperationTimeout,
@@ -735,6 +733,7 @@ func (s *server) runControllers(config *api.Config, defaultVolumeSnapshotLocatio
 		restorer,
 		s.sharedInformerFactory.Ark().V1().Backups(),
 		s.sharedInformerFactory.Ark().V1().BackupStorageLocations(),
+		s.sharedInformerFactory.Ark().V1().VolumeSnapshotLocations(),
 		false,
 		s.logger,
 		s.logLevel,

pkg/controller/restore_controller.go

@@ -68,17 +68,18 @@ var nonRestorableResources = []string{
 type restoreController struct {
 	*genericController
 
-	namespace             string
-	restoreClient         arkv1client.RestoresGetter
-	backupClient          arkv1client.BackupsGetter
-	restorer              restore.Restorer
-	pvProviderExists      bool
-	backupLister          listers.BackupLister
-	restoreLister         listers.RestoreLister
-	backupLocationLister  listers.BackupStorageLocationLister
-	restoreLogLevel       logrus.Level
-	defaultBackupLocation string
-	metrics               *metrics.ServerMetrics
+	namespace              string
+	restoreClient          arkv1client.RestoresGetter
+	backupClient           arkv1client.BackupsGetter
+	restorer               restore.Restorer
+	pvProviderExists       bool
+	backupLister           listers.BackupLister
+	restoreLister          listers.RestoreLister
+	backupLocationLister   listers.BackupStorageLocationLister
+	snapshotLocationLister listers.VolumeSnapshotLocationLister
+	restoreLogLevel        logrus.Level
+	defaultBackupLocation  string
+	metrics                *metrics.ServerMetrics
 
 	newPluginManager func(logger logrus.FieldLogger) plugin.Manager
 	newBackupStore   func(*api.BackupStorageLocation, persistence.ObjectStoreGetter, logrus.FieldLogger) (persistence.BackupStore, error)
@@ -92,6 +93,7 @@ func NewRestoreController(
 	restorer restore.Restorer,
 	backupInformer informers.BackupInformer,
 	backupLocationInformer informers.BackupStorageLocationInformer,
+	snapshotLocationInformer informers.VolumeSnapshotLocationInformer,
 	pvProviderExists bool,
 	logger logrus.FieldLogger,
 	restoreLogLevel logrus.Level,
@@ -100,18 +102,19 @@ func NewRestoreController(
 	metrics *metrics.ServerMetrics,
 ) Interface {
 	c := &restoreController{
-		genericController:     newGenericController("restore", logger),
-		namespace:             namespace,
-		restoreClient:         restoreClient,
-		backupClient:          backupClient,
-		restorer:              restorer,
-		pvProviderExists:      pvProviderExists,
-		backupLister:          backupInformer.Lister(),
-		restoreLister:         restoreInformer.Lister(),
-		backupLocationLister:  backupLocationInformer.Lister(),
-		restoreLogLevel:       restoreLogLevel,
-		defaultBackupLocation: defaultBackupLocation,
-		metrics:               metrics,
+		genericController:      newGenericController("restore", logger),
+		namespace:              namespace,
+		restoreClient:          restoreClient,
+		backupClient:           backupClient,
+		restorer:               restorer,
+		pvProviderExists:       pvProviderExists,
+		backupLister:           backupInformer.Lister(),
+		restoreLister:          restoreInformer.Lister(),
+		backupLocationLister:   backupLocationInformer.Lister(),
+		snapshotLocationLister: snapshotLocationInformer.Lister(),
+		restoreLogLevel:        restoreLogLevel,
+		defaultBackupLocation:  defaultBackupLocation,
+		metrics:                metrics,
 
 		// use variables to refer to these functions so they can be
 		// replaced with fakes for testing.
@@ -124,6 +127,7 @@ func NewRestoreController(
 		backupInformer.Informer().HasSynced,
 		restoreInformer.Informer().HasSynced,
 		backupLocationInformer.Informer().HasSynced,
+		snapshotLocationInformer.Informer().HasSynced,
 	)
 
 	restoreInformer.Informer().AddEventHandler(
@@ -233,6 +237,7 @@ func (c *restoreController) processRestore(key string) error {
 		restore,
 		actions,
 		info,
+		pluginManager,
 	)
 
 	restore.Status.Warnings = len(restoreWarnings.Ark) + len(restoreWarnings.Cluster)
@@ -482,6 +487,7 @@ func (c *restoreController) runRestore(
 	restore *api.Restore,
 	actions []restore.ItemAction,
 	info backupInfo,
+	pluginManager plugin.Manager,
 ) (restoreWarnings, restoreErrors api.RestoreResult, restoreFailure error) {
 	logFile, err := ioutil.TempFile("", "")
 	if err != nil {
@@ -531,10 +537,18 @@ func (c *restoreController) runRestore(
 	}
 	defer closeAndRemoveFile(resultsFile, c.logger)
 
+	volumeSnapshots, err := info.backupStore.GetBackupVolumeSnapshots(restore.Spec.BackupName)
+	if err != nil {
+		log.WithError(errors.WithStack(err)).Error("Error fetching volume snapshots")
+		restoreErrors.Ark = append(restoreErrors.Ark, err.Error())
+		restoreFailure = err
+		return
+	}
+
 	// Any return statement above this line means a total restore failure
 	// Some failures after this line *may* be a total restore failure
 	log.Info("starting restore")
-	restoreWarnings, restoreErrors = c.restorer.Restore(log, restore, info.backup, backupFile, actions)
+	restoreWarnings, restoreErrors = c.restorer.Restore(log, restore, info.backup, volumeSnapshots, backupFile, actions, c.snapshotLocationLister, pluginManager)
 	log.Info("restore completed")
 
 	// Try to upload the log file. This is best-effort. If we fail, we'll add to the ark errors.

pkg/controller/restore_controller_test.go

@@ -24,20 +24,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/pkg/errors"
-	"github.com/sirupsen/logrus"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	core "k8s.io/client-go/testing"
-	"k8s.io/client-go/tools/cache"
-
 	api "github.com/heptio/ark/pkg/apis/ark/v1"
 	"github.com/heptio/ark/pkg/generated/clientset/versioned/fake"
 	informers "github.com/heptio/ark/pkg/generated/informers/externalversions"
+	listers "github.com/heptio/ark/pkg/generated/listers/ark/v1"
 	"github.com/heptio/ark/pkg/metrics"
 	"github.com/heptio/ark/pkg/persistence"
 	persistencemocks "github.com/heptio/ark/pkg/persistence/mocks"
@@ -46,6 +36,16 @@ import (
 	"github.com/heptio/ark/pkg/restore"
 	"github.com/heptio/ark/pkg/util/collections"
 	arktest "github.com/heptio/ark/pkg/util/test"
+	"github.com/heptio/ark/pkg/volume"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	core "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/cache"
 )
 
 func TestFetchBackupInfo(t *testing.T) {
@@ -104,6 +104,7 @@ func TestFetchBackupInfo(t *testing.T) {
 				restorer,
 				sharedInformers.Ark().V1().Backups(),
 				sharedInformers.Ark().V1().BackupStorageLocations(),
+				sharedInformers.Ark().V1().VolumeSnapshotLocations(),
 				false,
 				logger,
 				logrus.InfoLevel,
@@ -197,6 +198,7 @@ func TestProcessRestoreSkips(t *testing.T) {
 				restorer,
 				sharedInformers.Ark().V1().Backups(),
 				sharedInformers.Ark().V1().BackupStorageLocations(),
+				sharedInformers.Ark().V1().VolumeSnapshotLocations(),
 				false, // pvProviderExists
 				logger,
 				logrus.InfoLevel,
@@ -422,6 +424,7 @@ func TestProcessRestore(t *testing.T) {
 				restorer,
 				sharedInformers.Ark().V1().Backups(),
 				sharedInformers.Ark().V1().BackupStorageLocations(),
+				sharedInformers.Ark().V1().VolumeSnapshotLocations(),
 				test.allowRestoreSnapshots,
 				logger,
 				logrus.InfoLevel,
@@ -498,6 +501,16 @@ func TestProcessRestore(t *testing.T) {
 				backupStore.On("PutRestoreLog", test.backup.Name, test.restore.Name, mock.Anything).Return(test.putRestoreLogErr)
 
 				backupStore.On("PutRestoreResults", test.backup.Name, test.restore.Name, mock.Anything).Return(nil)
+
+				volumeSnapshots := []*volume.Snapshot{
+					{
+						Spec: volume.SnapshotSpec{
+							PersistentVolumeName: "test-pv",
+							BackupName:           test.backup.Name,
+						},
+					},
+				}
+				backupStore.On("GetBackupVolumeSnapshots", test.backup.Name).Return(volumeSnapshots, nil)
 			}
 
 			var (
@@ -629,6 +642,7 @@ func TestvalidateAndCompleteWhenScheduleNameSpecified(t *testing.T) {
 		nil,
 		sharedInformers.Ark().V1().Backups(),
 		sharedInformers.Ark().V1().BackupStorageLocations(),
+		sharedInformers.Ark().V1().VolumeSnapshotLocations(),
 		false,
 		logger,
 		logrus.DebugLevel,
@@ -815,8 +829,11 @@ func (r *fakeRestorer) Restore(
 	log logrus.FieldLogger,
 	restore *api.Restore,
 	backup *api.Backup,
+	volumeSnapshots []*volume.Snapshot,
 	backupReader io.Reader,
 	actions []restore.ItemAction,
+	snapshotLocationLister listers.VolumeSnapshotLocationLister,
+	blockStoreGetter restore.BlockStoreGetter,
 ) (api.RestoreResult, api.RestoreResult) {
 	res := r.Called(log, restore, backup, backupReader, actions)