sync controller: fill in missing .spec.storageLocation

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

pkg/controller/backup_sync_controller.go

@@ -163,16 +163,27 @@ func (c *backupSyncController) run() {
 
 			// use the controller's namespace when getting the backup because that's where we
 			// are syncing backups to, regardless of the namespace of the cloud backup.
-			_, err := c.backupClient.Backups(c.namespace).Get(backupName, metav1.GetOptions{})
+			backup, err := c.backupClient.Backups(c.namespace).Get(backupName, metav1.GetOptions{})
 			if err == nil {
 				log.Debug("Backup already exists in cluster")
+
+				if backup.Spec.StorageLocation != "" {
+					continue
+				}
+
+				// pre-v0.10 backups won't initially have a .spec.storageLocation so fill it in
+				log.Debug("Patching backup's .spec.storageLocation because it's missing")
+				if err := patchStorageLocation(backup, c.backupClient.Backups(c.namespace), location.Name); err != nil {
+					log.WithError(err).Error("Error patching backup's .spec.storageLocation")
+				}
+
 				continue
 			}
 			if !kuberrs.IsNotFound(err) {
 				log.WithError(errors.WithStack(err)).Error("Error getting backup from client, proceeding with sync into cluster")
 			}
 
-			backup, err := backupStore.GetBackupMetadata(backupName)
+			backup, err = backupStore.GetBackupMetadata(backupName)
 			if err != nil {
 				log.WithError(errors.WithStack(err)).Error("Error getting backup metadata from backup store")
 				continue
@@ -233,6 +244,25 @@ func (c *backupSyncController) run() {
 	}
 }
 
+func patchStorageLocation(backup *arkv1api.Backup, client arkv1client.BackupInterface, location string) error {
+	patch := map[string]interface{}{
+		"spec": map[string]interface{}{
+			"storageLocation": location,
+		},
+	}
+
+	patchBytes, err := json.Marshal(patch)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	if _, err := client.Patch(backup.Name, types.MergePatchType, patchBytes); err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}
+
 // deleteOrphanedBackups deletes backup objects from Kubernetes that have the specified location
 // and a phase of Completed, but no corresponding backup in object storage.
 func (c *backupSyncController) deleteOrphanedBackups(locationName string, cloudBackupNames sets.String, log logrus.FieldLogger) {

pkg/controller/backup_sync_controller_test.go

@@ -138,11 +138,26 @@ func TestBackupSyncControllerRun(t *testing.T) {
 					arktest.NewTestBackup().WithNamespace("ns-1").WithName("backup-4").Backup,
 				},
 			},
+			existingBackups: []*arkv1api.Backup{
+				// add a label to each existing backup so we can differentiate it from the cloud
+				// backup during verification
+				arktest.NewTestBackup().WithNamespace("ns-1").WithName("backup-1").WithLabel("i-exist", "true").WithStorageLocation("location-1").Backup,
+				arktest.NewTestBackup().WithNamespace("ns-1").WithName("backup-3").WithLabel("i-exist", "true").WithStorageLocation("location-2").Backup,
+			},
+		},
+		{
+			name:      "existing backups without a StorageLocation get it filled in",
+			namespace: "ns-1",
+			locations: defaultLocationsList("ns-1"),
+			cloudBackups: map[string][]*arkv1api.Backup{
+				"bucket-1": {
+					arktest.NewTestBackup().WithNamespace("ns-1").WithName("backup-1").Backup,
+				},
+			},
 			existingBackups: []*arkv1api.Backup{
 				// add a label to each existing backup so we can differentiate it from the cloud
 				// backup during verification
 				arktest.NewTestBackup().WithNamespace("ns-1").WithName("backup-1").WithLabel("i-exist", "true").Backup,
-				arktest.NewTestBackup().WithNamespace("ns-1").WithName("backup-3").WithLabel("i-exist", "true").Backup,
 			},
 		},
 		{
@@ -219,6 +234,17 @@ func TestBackupSyncControllerRun(t *testing.T) {
 			c.run()
 
 			for bucket, backups := range test.cloudBackups {
+				// figure out which location this bucket is for; we need this for verification
+				// purposes later
+				var location *arkv1api.BackupStorageLocation
+				for _, loc := range test.locations {
+					if loc.Spec.ObjectStorage.Bucket == bucket {
+						location = loc
+						break
+					}
+				}
+				require.NotNil(t, location)
+
 				for _, cloudBackup := range backups {
 					obj, err := client.ArkV1().Backups(test.namespace).Get(cloudBackup.Name, metav1.GetOptions{})
 					require.NoError(t, err)
@@ -235,19 +261,19 @@ func TestBackupSyncControllerRun(t *testing.T) {
 					if existing != nil {
 						// if this cloud backup already exists in the cluster, make sure that what we get from the
 						// client is the existing backup, not the cloud one.
-						assert.Equal(t, existing, obj)
+
+						// verify that the in-cluster backup has its storage location populated, if it's not already.
+						expected := existing.DeepCopy()
+						expected.Spec.StorageLocation = location.Name
+
+						assert.Equal(t, expected, obj)
 					} else {
 						// verify that the GC finalizer is removed
 						assert.Equal(t, stringslice.Except(cloudBackup.Finalizers, gcFinalizer), obj.Finalizers)
 
 						// verify that the storage location field and label are set properly
-						for _, location := range test.locations {
-							if location.Spec.ObjectStorage.Bucket == bucket {
-								assert.Equal(t, location.Name, obj.Spec.StorageLocation)
-								assert.Equal(t, location.Name, obj.Labels[arkv1api.StorageLocationLabel])
-								break
-							}
-						}
+						assert.Equal(t, location.Name, obj.Spec.StorageLocation)
+						assert.Equal(t, location.Name, obj.Labels[arkv1api.StorageLocationLabel])
 					}
 				}
 			}