Don't allow deletion of in-progress backups

Signed-off-by: Andy Goldstein <andy.goldstein@gmail.com>

pkg/cmd/server/server.go

@@ -476,6 +476,8 @@ func (s *server) runControllers(config *api.Config) error {
 	if config.RestoreOnlyMode {
 		s.logger.Info("Restore only mode - not starting the backup, schedule, delete-backup, or GC controllers")
 	} else {
+		backupTracker := controller.NewBackupTracker()
+
 		backupper, err := newBackupper(discoveryHelper, s.clientPool, s.backupService, s.snapshotService, s.kubeClientConfig, s.kubeClient.CoreV1())
 		cmd.CheckError(err)
 		backupController := controller.NewBackupController(
@@ -487,6 +489,7 @@ func (s *server) runControllers(config *api.Config) error {
 			s.snapshotService != nil,
 			s.logger,
 			s.pluginManager,
+			backupTracker,
 		)
 		wg.Add(1)
 		go func() {
@@ -530,6 +533,7 @@ func (s *server) runControllers(config *api.Config) error {
 			config.BackupStorageProvider.Bucket,
 			s.sharedInformerFactory.Ark().V1().Restores(),
 			s.arkClient.ArkV1(), // restoreClient
+			backupTracker,
 		)
 		wg.Add(1)
 		go func() {

pkg/controller/backup_controller.go

@@ -66,6 +66,7 @@ type backupController struct {
 	clock            clock.Clock
 	logger           logrus.FieldLogger
 	pluginManager    plugin.Manager
+	backupTracker    BackupTracker
 }
 
 func NewBackupController(
@@ -77,6 +78,7 @@ func NewBackupController(
 	pvProviderExists bool,
 	logger logrus.FieldLogger,
 	pluginManager plugin.Manager,
+	backupTracker BackupTracker,
 ) Interface {
 	c := &backupController{
 		backupper:        backupper,
@@ -90,6 +92,7 @@ func NewBackupController(
 		clock:            &clock.RealClock{},
 		logger:           logger,
 		pluginManager:    pluginManager,
+		backupTracker:    backupTracker,
 	}
 
 	c.syncHandler = c.processBackup
@@ -261,6 +264,9 @@ func (controller *backupController) processBackup(key string) error {
 		return nil
 	}
 
+	controller.backupTracker.Add(backup.Namespace, backup.Name)
+	defer controller.backupTracker.Delete(backup.Namespace, backup.Name)
+
 	logContext.Debug("Running backup")
 	// execution & upload of backup
 	if err := controller.runBackup(backup, controller.bucket); err != nil {

pkg/controller/backup_controller_test.go

@@ -167,6 +167,7 @@ func TestProcessBackup(t *testing.T) {
 				test.allowSnapshots,
 				logger,
 				pluginManager,
+				NewBackupTracker(),
 			).(*backupController)
 			c.clock = clock.NewFakeClock(time.Now())
 

pkg/controller/backup_deletion_controller.go

@@ -49,6 +49,7 @@ type backupDeletionController struct {
 	bucket                    string
 	restoreLister             listers.RestoreLister
 	restoreClient             arkv1client.RestoresGetter
+	backupTracker             BackupTracker
 
 	processRequestFunc func(*v1.DeleteBackupRequest) error
 	clock              clock.Clock
@@ -65,6 +66,7 @@ func NewBackupDeletionController(
 	bucket string,
 	restoreInformer informers.RestoreInformer,
 	restoreClient arkv1client.RestoresGetter,
+	backupTracker BackupTracker,
 ) Interface {
 	c := &backupDeletionController{
 		genericController:         newGenericController("backup-deletion", logger),
@@ -76,6 +78,7 @@ func NewBackupDeletionController(
 		bucket:                    bucket,
 		restoreLister:             restoreInformer.Lister(),
 		restoreClient:             restoreClient,
+		backupTracker:             backupTracker,
 		clock:                     &clock.RealClock{},
 	}
 
@@ -133,16 +136,26 @@ func (c *backupDeletionController) processRequest(req *v1.DeleteBackupRequest) e
 		"backup":    req.Spec.BackupName,
 	})
 
+	var err error
+
 	// Make sure we have the backup name
 	if req.Spec.BackupName == "" {
-		_, err := c.patchDeleteBackupRequest(req, func(r *v1.DeleteBackupRequest) {
+		_, err = c.patchDeleteBackupRequest(req, func(r *v1.DeleteBackupRequest) {
 			r.Status.Phase = v1.DeleteBackupRequestPhaseProcessed
 			r.Status.Errors = []string{"spec.backupName is required"}
 		})
 		return err
 	}
 
-	var err error
+	// Don't allow deleting an in-progress backup
+	if c.backupTracker.Contains(req.Namespace, req.Spec.BackupName) {
+		_, err = c.patchDeleteBackupRequest(req, func(r *v1.DeleteBackupRequest) {
+			r.Status.Phase = v1.DeleteBackupRequestPhaseProcessed
+			r.Status.Errors = []string{"backup is still in progress"}
+		})
+
+		return err
+	}
 
 	// Update status to InProgress and set backup-name label if needed
 	req, err = c.patchDeleteBackupRequest(req, func(r *v1.DeleteBackupRequest) {

pkg/controller/backup_deletion_controller_test.go

@@ -64,6 +64,7 @@ func TestBackupDeletionControllerControllerHasUpdateFunc(t *testing.T) {
 		"bucket",
 		sharedInformers.Ark().V1().Restores(),
 		client.ArkV1(), // restoreClient
+		NewBackupTracker(),
 	).(*backupDeletionController)
 
 	// disable resync handler since we don't want to test it here
@@ -116,6 +117,7 @@ func TestBackupDeletionControllerProcessQueueItem(t *testing.T) {
 		"bucket",
 		sharedInformers.Ark().V1().Restores(),
 		client.ArkV1(), // restoreClient
+		NewBackupTracker(),
 	).(*backupDeletionController)
 
 	// Error splitting key
@@ -196,6 +198,7 @@ func setupBackupDeletionControllerTest(objects ...runtime.Object) *backupDeletio
 			"bucket",
 			sharedInformers.Ark().V1().Restores(),
 			client.ArkV1(), // restoreClient
+			NewBackupTracker(),
 		).(*backupDeletionController),
 
 		req: req,
@@ -228,6 +231,27 @@ func TestBackupDeletionControllerProcessRequest(t *testing.T) {
 		assert.Equal(t, expectedActions, td.client.Actions())
 	})
 
+	t.Run("deleting an in progress backup isn't allowed", func(t *testing.T) {
+		td := setupBackupDeletionControllerTest()
+		defer td.backupService.AssertExpectations(t)
+
+		td.controller.backupTracker.Add(td.req.Namespace, td.req.Spec.BackupName)
+
+		err := td.controller.processRequest(td.req)
+		require.NoError(t, err)
+
+		expectedActions := []core.Action{
+			core.NewPatchAction(
+				v1.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				[]byte(`{"status":{"errors":["backup is still in progress"],"phase":"Processed"}}`),
+			),
+		}
+
+		assert.Equal(t, expectedActions, td.client.Actions())
+	})
+
 	t.Run("patching to InProgress fails", func(t *testing.T) {
 		td := setupBackupDeletionControllerTest()
 		defer td.backupService.AssertExpectations(t)
@@ -562,6 +586,7 @@ func TestBackupDeletionControllerDeleteExpiredRequests(t *testing.T) {
 				"bucket",
 				sharedInformers.Ark().V1().Restores(),
 				client.ArkV1(), // restoreClient
+				NewBackupTracker(),
 			).(*backupDeletionController)
 
 			fakeClock := &clock.FakeClock{}

pkg/controller/backup_tracker.go

@@ -0,0 +1,71 @@
+/*
+Copyright 2018 the Heptio Ark 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 controller
+
+import (
+	"fmt"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+// BackupTracker keeps track of in-progress backups.
+type BackupTracker interface {
+	// Add informs the tracker that a backup is in progress.
+	Add(ns, name string)
+	// Delete informs the tracker that a backup is no longer in progress.
+	Delete(ns, name string)
+	// Contains returns true if the tracker is tracking the backup.
+	Contains(ns, name string) bool
+}
+
+type backupTracker struct {
+	lock    sync.RWMutex
+	backups sets.String
+}
+
+// NewBackupTracker returns a new BackupTracker.
+func NewBackupTracker() BackupTracker {
+	return &backupTracker{
+		backups: sets.NewString(),
+	}
+}
+
+func (bt *backupTracker) Add(ns, name string) {
+	bt.lock.Lock()
+	defer bt.lock.Unlock()
+
+	bt.backups.Insert(backupTrackerKey(ns, name))
+}
+
+func (bt *backupTracker) Delete(ns, name string) {
+	bt.lock.Lock()
+	defer bt.lock.Unlock()
+
+	bt.backups.Delete(backupTrackerKey(ns, name))
+}
+
+func (bt *backupTracker) Contains(ns, name string) bool {
+	bt.lock.RLock()
+	defer bt.lock.RUnlock()
+
+	return bt.backups.Has(backupTrackerKey(ns, name))
+}
+
+func backupTrackerKey(ns, name string) string {
+	return fmt.Sprintf("%s/%s", ns, name)
+}

pkg/controller/backup_tracker_test.go

@@ -0,0 +1,43 @@
+/*
+Copyright 2018 the Heptio Ark 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 controller
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBackupTracker(t *testing.T) {
+	bt := NewBackupTracker()
+
+	assert.False(t, bt.Contains("ns", "name"))
+
+	bt.Add("ns", "name")
+	assert.True(t, bt.Contains("ns", "name"))
+
+	bt.Add("ns2", "name2")
+	assert.True(t, bt.Contains("ns", "name"))
+	assert.True(t, bt.Contains("ns2", "name2"))
+
+	bt.Delete("ns", "name")
+	assert.False(t, bt.Contains("ns", "name"))
+	assert.True(t, bt.Contains("ns2", "name2"))
+
+	bt.Delete("ns2", "name2")
+	assert.False(t, bt.Contains("ns2", "name2"))
+}