make PVProvider optional in server config; disallow snap/restore PVs when not provided

Signed-off-by: Steve Kriss <steve@heptio.com>

pkg/controller/backup_controller.go

@@ -49,9 +49,10 @@ import (
 const backupVersion = 1
 
 type backupController struct {
-	backupper     backup.Backupper
-	backupService cloudprovider.BackupService
-	bucket        string
+	backupper      backup.Backupper
+	backupService  cloudprovider.BackupService
+	bucket         string
+	allowSnapshots bool
 
 	lister       listers.BackupLister
 	listerSynced cache.InformerSynced
@@ -68,11 +69,13 @@ func NewBackupController(
 	backupper backup.Backupper,
 	backupService cloudprovider.BackupService,
 	bucket string,
+	allowSnapshots bool,
 ) Interface {
 	c := &backupController{
-		backupper:     backupper,
-		backupService: backupService,
-		bucket:        bucket,
+		backupper:      backupper,
+		backupService:  backupService,
+		bucket:         bucket,
+		allowSnapshots: allowSnapshots,
 
 		lister:       backupInformer.Lister(),
 		listerSynced: backupInformer.Informer().HasSynced,
@@ -297,6 +300,10 @@ func (controller *backupController) getValidationErrors(itm *api.Backup) []strin
 		validationErrors = append(validationErrors, fmt.Sprintf("Invalid included/excluded namespace lists: %v", err))
 	}
 
+	if !controller.allowSnapshots && itm.Spec.SnapshotVolumes {
+		validationErrors = append(validationErrors, "Server is not configured for PV snapshots")
+	}
+
 	return validationErrors
 }
 

pkg/controller/backup_controller_test.go

@@ -54,6 +54,7 @@ func TestProcessBackup(t *testing.T) {
 		expectedExcludes []string
 		backup           *TestBackup
 		expectBackup     bool
+		allowSnapshots   bool
 	}{
 		{
 			name:        "bad key",
@@ -129,6 +130,20 @@ func TestProcessBackup(t *testing.T) {
 			expectedIncludes: []string{"*"},
 			expectBackup:     true,
 		},
+		{
+			name:         "backup with SnapshotVolumes when allowSnapshots=false fails validation",
+			key:          "heptio-ark/backup1",
+			backup:       NewTestBackup().WithName("backup1").WithPhase(v1.BackupPhaseNew).WithSnapshotVolumes(true),
+			expectBackup: false,
+		},
+		{
+			name:             "backup with SnapshotVolumes when allowSnapshots=true gets executed",
+			key:              "heptio-ark/backup1",
+			backup:           NewTestBackup().WithName("backup1").WithPhase(v1.BackupPhaseNew).WithSnapshotVolumes(true),
+			allowSnapshots:   true,
+			expectedIncludes: []string{"*"},
+			expectBackup:     true,
+		},
 	}
 
 	// flag.Set("logtostderr", "true")
@@ -150,6 +165,7 @@ func TestProcessBackup(t *testing.T) {
 				backupper,
 				cloudBackups,
 				"bucket",
+				test.allowSnapshots,
 			).(*backupController)
 			c.clock = clock.NewFakeClock(time.Now())
 
@@ -180,6 +196,7 @@ func TestProcessBackup(t *testing.T) {
 				}
 
 				backup.Spec.IncludedNamespaces = expectedNSes
+				backup.Spec.SnapshotVolumes = test.backup.Spec.SnapshotVolumes
 				backup.Status.Phase = v1.BackupPhaseInProgress
 				backup.Status.Expiration.Time = expiration
 				backup.Status.Version = 1
@@ -226,6 +243,7 @@ func TestProcessBackup(t *testing.T) {
 						WithExcludedResources(test.expectedExcludes...).
 						WithIncludedNamespaces(expectedNSes...).
 						WithTTL(test.backup.Spec.TTL.Duration).
+						WithSnapshotVolumes(test.backup.Spec.SnapshotVolumes).
 						WithExpiration(expiration).
 						WithVersion(1).
 						Backup,
@@ -241,6 +259,7 @@ func TestProcessBackup(t *testing.T) {
 						WithExcludedResources(test.expectedExcludes...).
 						WithIncludedNamespaces(expectedNSes...).
 						WithTTL(test.backup.Spec.TTL.Duration).
+						WithSnapshotVolumes(test.backup.Spec.SnapshotVolumes).
 						WithExpiration(expiration).
 						WithVersion(1).
 						Backup,

pkg/controller/gc_controller.go

@@ -108,26 +108,36 @@ func (c *gcController) cleanBackups() {
 	// storage should happen first because otherwise there's a possibility the backup sync
 	// controller would re-create the API object after deletion.
 	for _, backup := range backups {
-		if backup.Status.Expiration.Time.Before(now) {
-			glog.Infof("Removing backup %s/%s", backup.Namespace, backup.Name)
-			if err := c.backupService.DeleteBackup(c.bucket, backup.Name); err != nil {
-				glog.Errorf("error deleting backup %s/%s: %v", backup.Namespace, backup.Name, err)
-			}
-
-			for _, volumeBackup := range backup.Status.VolumeBackups {
-				glog.Infof("Removing snapshot %s associated with backup %s/%s", volumeBackup.SnapshotID, backup.Namespace, backup.Name)
-				if err := c.snapshotService.DeleteSnapshot(volumeBackup.SnapshotID); err != nil {
-					glog.Errorf("error deleting snapshot %v: %v", volumeBackup.SnapshotID, err)
-				}
-			}
-
-			glog.Infof("Removing backup API object %s/%s", backup.Namespace, backup.Name)
-			if err := c.client.Backups(backup.Namespace).Delete(backup.Name, &metav1.DeleteOptions{}); err != nil {
-				glog.Errorf("error deleting backup API object %s/%s: %v", backup.Namespace, backup.Name, err)
-			}
-		} else {
+		if !backup.Status.Expiration.Time.Before(now) {
 			glog.Infof("Backup %s/%s has not expired yet, skipping", backup.Namespace, backup.Name)
+			continue
 		}
+
+		// if the backup includes snapshots but we don't currently have a PVProvider, we don't
+		// want to orphan the snapshots so skip garbage-collection entirely.
+		if c.snapshotService == nil && len(backup.Status.VolumeBackups) > 0 {
+			glog.Warningf("Cannot garbage-collect backup %s/%s because backup includes snapshots and server is not configured with PersistentVolumeProvider",
+				backup.Namespace, backup.Name)
+			continue
+		}
+
+		glog.Infof("Removing backup %s/%s", backup.Namespace, backup.Name)
+		if err := c.backupService.DeleteBackup(c.bucket, backup.Name); err != nil {
+			glog.Errorf("error deleting backup %s/%s: %v", backup.Namespace, backup.Name, err)
+		}
+
+		for _, volumeBackup := range backup.Status.VolumeBackups {
+			glog.Infof("Removing snapshot %s associated with backup %s/%s", volumeBackup.SnapshotID, backup.Namespace, backup.Name)
+			if err := c.snapshotService.DeleteSnapshot(volumeBackup.SnapshotID); err != nil {
+				glog.Errorf("error deleting snapshot %v: %v", volumeBackup.SnapshotID, err)
+			}
+		}
+
+		glog.Infof("Removing backup API object %s/%s", backup.Namespace, backup.Name)
+		if err := c.client.Backups(backup.Namespace).Delete(backup.Name, &metav1.DeleteOptions{}); err != nil {
+			glog.Errorf("error deleting backup API object %s/%s: %v", backup.Namespace, backup.Name, err)
+		}
+
 	}
 
 	// also GC any Backup API objects without files in object storage

pkg/controller/gc_controller_test.go

@@ -31,16 +31,18 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	api "github.com/heptio/ark/pkg/apis/ark/v1"
+	"github.com/heptio/ark/pkg/cloudprovider"
 	"github.com/heptio/ark/pkg/generated/clientset/fake"
 	informers "github.com/heptio/ark/pkg/generated/informers/externalversions"
 	. "github.com/heptio/ark/pkg/util/test"
 )
 
 type gcTest struct {
-	name      string
-	bucket    string
-	backups   map[string][]*api.Backup
-	snapshots sets.String
+	name               string
+	bucket             string
+	backups            map[string][]*api.Backup
+	snapshots          sets.String
+	nilSnapshotService bool
 
 	expectedBackupsRemaining   map[string]sets.String
 	expectedSnapshotsRemaining sets.String
@@ -149,11 +151,38 @@ func TestGarbageCollect(t *testing.T) {
 			expectedBackupsRemaining:   make(map[string]sets.String),
 			expectedSnapshotsRemaining: sets.NewString("snapshot-3", "snapshot-4"),
 		},
+		gcTest{
+			name:   "no snapshot service only GC's backups without snapshots",
+			bucket: "bucket-1",
+			backups: map[string][]*api.Backup{
+				"bucket-1": []*api.Backup{
+					NewTestBackup().WithName("backup-1").
+						WithExpiration(fakeClock.Now().Add(-1*time.Second)).
+						WithSnapshot("pv-1", "snapshot-1").
+						WithSnapshot("pv-2", "snapshot-2").
+						Backup,
+					NewTestBackup().WithName("backup-2").
+						WithExpiration(fakeClock.Now().Add(-1 * time.Second)).
+						Backup,
+				},
+			},
+			snapshots:          sets.NewString("snapshot-1", "snapshot-2"),
+			nilSnapshotService: true,
+			expectedBackupsRemaining: map[string]sets.String{
+				"bucket-1": sets.NewString("backup-1"),
+			},
+		},
 	}
 
 	for _, test := range tests {
-		backupService := &fakeBackupService{}
-		snapshotService := &FakeSnapshotService{}
+		var (
+			backupService   = &fakeBackupService{}
+			snapshotService *FakeSnapshotService
+		)
+
+		if !test.nilSnapshotService {
+			snapshotService = &FakeSnapshotService{SnapshotsTaken: test.snapshots}
+		}
 
 		t.Run(test.name, func(t *testing.T) {
 			backupService.backupsByBucket = make(map[string][]*api.Backup)
@@ -167,16 +196,19 @@ func TestGarbageCollect(t *testing.T) {
 				backupService.backupsByBucket[bucket] = data
 			}
 
-			snapshotService.SnapshotsTaken = test.snapshots
-
 			var (
 				client          = fake.NewSimpleClientset()
 				sharedInformers = informers.NewSharedInformerFactory(client, 0)
+				snapSvc         cloudprovider.SnapshotService
 			)
 
+			if snapshotService != nil {
+				snapSvc = snapshotService
+			}
+
 			controller := NewGCController(
 				backupService,
-				snapshotService,
+				snapSvc,
 				test.bucket,
 				1*time.Millisecond,
 				sharedInformers.Ark().V1().Backups(),
@@ -202,7 +234,9 @@ func TestGarbageCollect(t *testing.T) {
 				assert.Equal(t, test.expectedBackupsRemaining[bucket], backupNames)
 			}
 
-			assert.Equal(t, test.expectedSnapshotsRemaining, snapshotService.SnapshotsTaken)
+			if !test.nilSnapshotService {
+				assert.Equal(t, test.expectedSnapshotsRemaining, snapshotService.SnapshotsTaken)
+			}
 		})
 	}
 }

pkg/controller/restore_controller.go

@@ -42,11 +42,12 @@ import (
 )
 
 type restoreController struct {
-	restoreClient arkv1client.RestoresGetter
-	backupClient  arkv1client.BackupsGetter
-	restorer      restore.Restorer
-	backupService cloudprovider.BackupService
-	bucket        string
+	restoreClient         arkv1client.RestoresGetter
+	backupClient          arkv1client.BackupsGetter
+	restorer              restore.Restorer
+	backupService         cloudprovider.BackupService
+	bucket                string
+	allowSnapshotRestores bool
 
 	backupLister        listers.BackupLister
 	backupListerSynced  cache.InformerSynced
@@ -64,18 +65,20 @@ func NewRestoreController(
 	backupService cloudprovider.BackupService,
 	bucket string,
 	backupInformer informers.BackupInformer,
+	allowSnapshotRestores bool,
 ) Interface {
 	c := &restoreController{
-		restoreClient:       restoreClient,
-		backupClient:        backupClient,
-		restorer:            restorer,
-		backupService:       backupService,
-		bucket:              bucket,
-		backupLister:        backupInformer.Lister(),
-		backupListerSynced:  backupInformer.Informer().HasSynced,
-		restoreLister:       restoreInformer.Lister(),
-		restoreListerSynced: restoreInformer.Informer().HasSynced,
-		queue:               workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "restore"),
+		restoreClient:         restoreClient,
+		backupClient:          backupClient,
+		restorer:              restorer,
+		backupService:         backupService,
+		bucket:                bucket,
+		allowSnapshotRestores: allowSnapshotRestores,
+		backupLister:          backupInformer.Lister(),
+		backupListerSynced:    backupInformer.Informer().HasSynced,
+		restoreLister:         restoreInformer.Lister(),
+		restoreListerSynced:   restoreInformer.Informer().HasSynced,
+		queue:                 workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "restore"),
 	}
 
 	c.syncHandler = c.processRestore
@@ -275,6 +278,10 @@ func (controller *restoreController) getValidationErrors(itm *api.Restore) []str
 		validationErrors = append(validationErrors, "BackupName must be non-empty and correspond to the name of a backup in object storage.")
 	}
 
+	if !controller.allowSnapshotRestores && itm.Spec.RestorePVs {
+		validationErrors = append(validationErrors, "Server is not configured for PV snapshot restores")
+	}
+
 	return validationErrors
 }
 

pkg/controller/restore_controller_test.go

@@ -42,6 +42,7 @@ func TestProcessRestore(t *testing.T) {
 		restore                *api.Restore
 		backup                 *api.Backup
 		restorerError          error
+		allowRestoreSnapshots  bool
 		expectedErr            bool
 		expectedRestoreUpdates []*api.Restore
 		expectedRestorerCall   *api.Restore
@@ -137,6 +138,28 @@ func TestProcessRestore(t *testing.T) {
 			},
 			expectedRestorerCall: NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithRestorableNamespace("*").Restore,
 		},
+		{
+			name:                  "valid restore with RestorePVs=true gets executed when allowRestoreSnapshots=true",
+			restore:               NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithRestorableNamespace("ns-1").WithRestorePVs(true).Restore,
+			backup:                NewTestBackup().WithName("backup-1").Backup,
+			allowRestoreSnapshots: true,
+			expectedErr:           false,
+			expectedRestoreUpdates: []*api.Restore{
+				NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithRestorableNamespace("ns-1").WithRestorePVs(true).Restore,
+				NewTestRestore("foo", "bar", api.RestorePhaseCompleted).WithBackup("backup-1").WithRestorableNamespace("ns-1").WithRestorePVs(true).Restore,
+			},
+			expectedRestorerCall: NewTestRestore("foo", "bar", api.RestorePhaseInProgress).WithBackup("backup-1").WithRestorableNamespace("ns-1").WithRestorePVs(true).Restore,
+		},
+		{
+			name:        "restore with RestorePVs=true fails validation when allowRestoreSnapshots=false",
+			restore:     NewTestRestore("foo", "bar", api.RestorePhaseNew).WithBackup("backup-1").WithRestorableNamespace("ns-1").WithRestorePVs(true).Restore,
+			backup:      NewTestBackup().WithName("backup-1").Backup,
+			expectedErr: false,
+			expectedRestoreUpdates: []*api.Restore{
+				NewTestRestore("foo", "bar", api.RestorePhaseFailedValidation).WithBackup("backup-1").WithRestorableNamespace("ns-1").WithRestorePVs(true).
+					WithValidationError("Server is not configured for PV snapshot restores").Restore,
+			},
+		},
 	}
 
 	// flag.Set("logtostderr", "true")
@@ -160,6 +183,7 @@ func TestProcessRestore(t *testing.T) {
 				backupSvc,
 				"bucket",
 				sharedInformers.Ark().V1().Backups(),
+				test.allowRestoreSnapshots,
 			).(*restoreController)
 
 			if test.restore != nil {