Validate restore name label length

Velero should handle cases when the label length exceeds 63 characters.

- if the length of the backup/restore name is <= 63 characters, use it as the value of the label
- if it's > 63 characters, take the SHA256 hash of the name. the value of
  the label will be the first 57 characters of the backup/restore name
  plus the first six characters of the SHA256 hash.

Fixes heptio#1021

Signed-off-by: Anshul Chandra <anshulc@vmware.com>

changelogs/unreleased/1392-anshulc

@@ -0,0 +1 @@
+shorten label values when they're longer than 63 characters

pkg/backup/delete_helpers.go

@@ -21,7 +21,8 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	v1 "github.com/heptio/velero/pkg/apis/velero/v1"
+	"github.com/heptio/velero/pkg/apis/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 )
 
 // NewDeleteBackupRequest creates a DeleteBackupRequest for the backup identified by name and uid.
@@ -30,7 +31,7 @@ func NewDeleteBackupRequest(name string, uid string) *v1.DeleteBackupRequest {
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: name + "-",
 			Labels: map[string]string{
-				v1.BackupNameLabel: name,
+				v1.BackupNameLabel: label.GetValidName(name),
 				v1.BackupUIDLabel:  uid,
 			},
 		},
@@ -44,6 +45,6 @@ func NewDeleteBackupRequest(name string, uid string) *v1.DeleteBackupRequest {
 // find DeleteBackupRequests for the backup identified by name and uid.
 func NewDeleteBackupRequestListOptions(name, uid string) metav1.ListOptions {
 	return metav1.ListOptions{
-		LabelSelector: fmt.Sprintf("%s=%s,%s=%s", v1.BackupNameLabel, name, v1.BackupUIDLabel, uid),
+		LabelSelector: fmt.Sprintf("%s=%s,%s=%s", v1.BackupNameLabel, label.GetValidName(name), v1.BackupUIDLabel, uid),
 	}
 }

pkg/controller/backup_controller.go

@@ -42,6 +42,7 @@ import (
 	velerov1client "github.com/heptio/velero/pkg/generated/clientset/versioned/typed/velero/v1"
 	informers "github.com/heptio/velero/pkg/generated/informers/externalversions/velero/v1"
 	listers "github.com/heptio/velero/pkg/generated/listers/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/metrics"
 	"github.com/heptio/velero/pkg/persistence"
 	"github.com/heptio/velero/pkg/plugin/clientmgmt"
@@ -296,7 +297,7 @@ func (c *backupController) prepareBackupRequest(backup *velerov1api.Backup) *pkg
 	if request.Labels == nil {
 		request.Labels = make(map[string]string)
 	}
-	request.Labels[velerov1api.StorageLocationLabel] = request.Spec.StorageLocation
+	request.Labels[velerov1api.StorageLocationLabel] = label.GetValidName(request.Spec.StorageLocation)
 
 	// validate the included/excluded resources
 	for _, err := range collections.ValidateIncludesExcludes(request.Spec.IncludedResources, request.Spec.ExcludedResources) {

pkg/controller/backup_controller_test.go

@@ -34,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/clock"
 
 	v1 "github.com/heptio/velero/pkg/apis/velero/v1"
+	velerov1api "github.com/heptio/velero/pkg/apis/velero/v1"
 	pkgbackup "github.com/heptio/velero/pkg/backup"
 	"github.com/heptio/velero/pkg/generated/clientset/versioned/fake"
 	informers "github.com/heptio/velero/pkg/generated/informers/externalversions"
@@ -191,6 +192,53 @@ func TestProcessBackupValidationFailures(t *testing.T) {
 	}
 }
 
+func TestBackupLocationLabel(t *testing.T) {
+	tests := []struct {
+		name                   string
+		backup                 *v1.Backup
+		backupLocation         *v1.BackupStorageLocation
+		expectedBackupLocation string
+	}{
+		{
+			name:                   "valid backup location name should be used as a label",
+			backup:                 velerotest.NewTestBackup().WithName("backup-1").Backup,
+			backupLocation:         velerotest.NewTestBackupStorageLocation().WithName("loc-1").BackupStorageLocation,
+			expectedBackupLocation: "loc-1",
+		},
+		{
+			name:   "invalid storage location name should be handled while creating label",
+			backup: velerotest.NewTestBackup().WithName("backup-1").Backup,
+			backupLocation: velerotest.NewTestBackupStorageLocation().
+				WithName("defaultdefaultdefaultdefaultdefaultdefaultdefaultdefaultdefaultdefault").BackupStorageLocation,
+			expectedBackupLocation: "defaultdefaultdefaultdefaultdefaultdefaultdefaultdefaultd58343f",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var (
+				clientset       = fake.NewSimpleClientset(test.backup)
+				sharedInformers = informers.NewSharedInformerFactory(clientset, 0)
+				logger          = logging.DefaultLogger(logrus.DebugLevel)
+			)
+
+			c := &backupController{
+				genericController:      newGenericController("backup-test", logger),
+				client:                 clientset.VeleroV1(),
+				lister:                 sharedInformers.Velero().V1().Backups().Lister(),
+				backupLocationLister:   sharedInformers.Velero().V1().BackupStorageLocations().Lister(),
+				snapshotLocationLister: sharedInformers.Velero().V1().VolumeSnapshotLocations().Lister(),
+				defaultBackupLocation:  test.backupLocation.Name,
+				clock:                  &clock.RealClock{},
+			}
+
+			res := c.prepareBackupRequest(test.backup)
+			assert.NotNil(t, res)
+			assert.Equal(t, test.expectedBackupLocation, res.Labels[velerov1api.StorageLocationLabel])
+		})
+	}
+}
+
 func TestDefaultBackupTTL(t *testing.T) {
 
 	var (

pkg/controller/backup_deletion_controller.go

@@ -21,7 +21,7 @@ import (
 	"encoding/json"
 	"time"
 
-	jsonpatch "github.com/evanphx/json-patch"
+	"github.com/evanphx/json-patch"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -32,11 +32,12 @@ import (
 	kubeerrs "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/tools/cache"
 
-	v1 "github.com/heptio/velero/pkg/apis/velero/v1"
+	"github.com/heptio/velero/pkg/apis/velero/v1"
 	pkgbackup "github.com/heptio/velero/pkg/backup"
 	velerov1client "github.com/heptio/velero/pkg/generated/clientset/versioned/typed/velero/v1"
 	informers "github.com/heptio/velero/pkg/generated/informers/externalversions/velero/v1"
 	listers "github.com/heptio/velero/pkg/generated/listers/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/metrics"
 	"github.com/heptio/velero/pkg/persistence"
 	"github.com/heptio/velero/pkg/plugin/clientmgmt"
@@ -196,7 +197,7 @@ func (c *backupDeletionController) processRequest(req *v1.DeleteBackupRequest) e
 		r.Status.Phase = v1.DeleteBackupRequestPhaseInProgress
 
 		if req.Labels[v1.BackupNameLabel] == "" {
-			req.Labels[v1.BackupNameLabel] = req.Spec.BackupName
+			req.Labels[v1.BackupNameLabel] = label.GetValidName(req.Spec.BackupName)
 		}
 	})
 	if err != nil {
@@ -390,7 +391,7 @@ func (c *backupDeletionController) backupStoreForBackup(backup *v1.Backup, plugi
 func (c *backupDeletionController) deleteExistingDeletionRequests(req *v1.DeleteBackupRequest, log logrus.FieldLogger) []error {
 	log.Info("Removing existing deletion requests for backup")
 	selector := labels.SelectorFromSet(labels.Set(map[string]string{
-		v1.BackupNameLabel: req.Spec.BackupName,
+		v1.BackupNameLabel: label.GetValidName(req.Spec.BackupName),
 	}))
 	dbrs, err := c.deleteBackupRequestLister.DeleteBackupRequests(req.Namespace).List(selector)
 	if err != nil {

pkg/controller/backup_deletion_controller_test.go

@@ -465,6 +465,151 @@ func TestBackupDeletionControllerProcessRequest(t *testing.T) {
 		// Make sure snapshot was deleted
 		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
 	})
+
+	t.Run("full delete, no errors, with backup name greater than 63 chars", func(t *testing.T) {
+		backup := velerotest.NewTestBackup().WithName("the-really-long-backup-name-that-is-much-more-than-63-characters").Backup
+		backup.UID = "uid"
+		backup.Spec.StorageLocation = "primary"
+
+		restore1 := velerotest.NewTestRestore("velero", "restore-1", v1.RestorePhaseCompleted).
+			WithBackup("the-really-long-backup-name-that-is-much-more-than-63-characters").Restore
+		restore2 := velerotest.NewTestRestore("velero", "restore-2", v1.RestorePhaseCompleted).
+			WithBackup("the-really-long-backup-name-that-is-much-more-than-63-characters").Restore
+		restore3 := velerotest.NewTestRestore("velero", "restore-3", v1.RestorePhaseCompleted).
+			WithBackup("some-other-backup").Restore
+
+		td := setupBackupDeletionControllerTest(backup, restore1, restore2, restore3)
+		td.req = pkgbackup.NewDeleteBackupRequest(backup.Name, string(backup.UID))
+		td.req.Namespace = "velero"
+		td.req.Name = "foo-abcde"
+		td.sharedInformers.Velero().V1().Restores().Informer().GetStore().Add(restore1)
+		td.sharedInformers.Velero().V1().Restores().Informer().GetStore().Add(restore2)
+		td.sharedInformers.Velero().V1().Restores().Informer().GetStore().Add(restore3)
+
+		location := &v1.BackupStorageLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      backup.Spec.StorageLocation,
+			},
+			Spec: v1.BackupStorageLocationSpec{
+				Provider: "objStoreProvider",
+				StorageType: v1.StorageType{
+					ObjectStorage: &v1.ObjectStorageLocation{
+						Bucket: "bucket",
+					},
+				},
+			},
+		}
+		require.NoError(t, td.sharedInformers.Velero().V1().BackupStorageLocations().Informer().GetStore().Add(location))
+
+		snapshotLocation := &v1.VolumeSnapshotLocation{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: backup.Namespace,
+				Name:      "vsl-1",
+			},
+			Spec: v1.VolumeSnapshotLocationSpec{
+				Provider: "provider-1",
+			},
+		}
+		require.NoError(t, td.sharedInformers.Velero().V1().VolumeSnapshotLocations().Informer().GetStore().Add(snapshotLocation))
+
+		// Clear out req labels to make sure the controller adds them
+		td.req.Labels = make(map[string]string)
+
+		td.client.PrependReactor("get", "backups", func(action core.Action) (bool, runtime.Object, error) {
+			return true, backup, nil
+		})
+		td.volumeSnapshotter.SnapshotsTaken.Insert("snap-1")
+
+		td.client.PrependReactor("patch", "deletebackuprequests", func(action core.Action) (bool, runtime.Object, error) {
+			return true, td.req, nil
+		})
+
+		td.client.PrependReactor("patch", "backups", func(action core.Action) (bool, runtime.Object, error) {
+			return true, backup, nil
+		})
+
+		snapshots := []*volume.Snapshot{
+			{
+				Spec: volume.SnapshotSpec{
+					Location: "vsl-1",
+				},
+				Status: volume.SnapshotStatus{
+					ProviderSnapshotID: "snap-1",
+				},
+			},
+		}
+
+		pluginManager := &pluginmocks.Manager{}
+		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
+		pluginManager.On("CleanupClients")
+		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }
+
+		td.backupStore.On("GetBackupVolumeSnapshots", td.req.Spec.BackupName).Return(snapshots, nil)
+		td.backupStore.On("DeleteBackup", td.req.Spec.BackupName).Return(nil)
+		td.backupStore.On("DeleteRestore", "restore-1").Return(nil)
+		td.backupStore.On("DeleteRestore", "restore-2").Return(nil)
+
+		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(`{"metadata":{"labels":{"velero.io/backup-name":"the-really-long-backup-name-that-is-much-more-than-63-cha6ca4bc"}},"status":{"phase":"InProgress"}}`),
+			),
+			core.NewGetAction(
+				v1.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+			),
+			core.NewPatchAction(
+				v1.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				[]byte(`{"metadata":{"labels":{"velero.io/backup-uid":"uid"}}}`),
+			),
+			core.NewPatchAction(
+				v1.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+				[]byte(`{"status":{"phase":"Deleting"}}`),
+			),
+			core.NewDeleteAction(
+				v1.SchemeGroupVersion.WithResource("restores"),
+				td.req.Namespace,
+				"restore-1",
+			),
+			core.NewDeleteAction(
+				v1.SchemeGroupVersion.WithResource("restores"),
+				td.req.Namespace,
+				"restore-2",
+			),
+			core.NewDeleteAction(
+				v1.SchemeGroupVersion.WithResource("backups"),
+				td.req.Namespace,
+				td.req.Spec.BackupName,
+			),
+			core.NewPatchAction(
+				v1.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				td.req.Name,
+				[]byte(`{"status":{"phase":"Processed"}}`),
+			),
+			core.NewDeleteCollectionAction(
+				v1.SchemeGroupVersion.WithResource("deletebackuprequests"),
+				td.req.Namespace,
+				pkgbackup.NewDeleteBackupRequestListOptions(td.req.Spec.BackupName, "uid"),
+			),
+		}
+
+		velerotest.CompareActions(t, expectedActions, td.client.Actions())
+
+		// Make sure snapshot was deleted
+		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
+	})
 }
 
 func TestBackupDeletionControllerDeleteExpiredRequests(t *testing.T) {

pkg/controller/backup_sync_controller.go

@@ -33,6 +33,7 @@ import (
 	velerov1client "github.com/heptio/velero/pkg/generated/clientset/versioned/typed/velero/v1"
 	informers "github.com/heptio/velero/pkg/generated/informers/externalversions/velero/v1"
 	listers "github.com/heptio/velero/pkg/generated/listers/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/persistence"
 	"github.com/heptio/velero/pkg/plugin/clientmgmt"
 )
@@ -218,7 +219,7 @@ func (c *backupSyncController) run() {
 			if backup.Labels == nil {
 				backup.Labels = make(map[string]string)
 			}
-			backup.Labels[velerov1api.StorageLocationLabel] = backup.Spec.StorageLocation
+			backup.Labels[velerov1api.StorageLocationLabel] = label.GetValidName(backup.Spec.StorageLocation)
 
 			_, err = c.backupClient.Backups(backup.Namespace).Create(backup)
 			switch {
@@ -283,7 +284,7 @@ func patchStorageLocation(backup *velerov1api.Backup, client velerov1client.Back
 // and a phase of Completed, but no corresponding backup in object storage.
 func (c *backupSyncController) deleteOrphanedBackups(locationName string, cloudBackupNames sets.String, log logrus.FieldLogger) {
 	locationSelector := labels.Set(map[string]string{
-		velerov1api.StorageLocationLabel: locationName,
+		velerov1api.StorageLocationLabel: label.GetValidName(locationName),
 	}).AsSelector()
 
 	backups, err := c.backupLister.Backups(c.namespace).List(locationSelector)

pkg/controller/backup_sync_controller_test.go

@@ -28,11 +28,13 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/clock"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/validation"
 	core "k8s.io/client-go/testing"
 
 	velerov1api "github.com/heptio/velero/pkg/apis/velero/v1"
 	"github.com/heptio/velero/pkg/generated/clientset/versioned/fake"
 	informers "github.com/heptio/velero/pkg/generated/informers/externalversions"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/persistence"
 	persistencemocks "github.com/heptio/velero/pkg/persistence/mocks"
 	"github.com/heptio/velero/pkg/plugin/clientmgmt"
@@ -73,13 +75,47 @@ func defaultLocationsList(namespace string) []*velerov1api.BackupStorageLocation
 	}
 }
 
+func defaultLocationsListWithLongerLocationName(namespace string) []*velerov1api.BackupStorageLocation {
+	return []*velerov1api.BackupStorageLocation{
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: namespace,
+				Name:      "the-really-long-location-name-that-is-much-more-than-63-characters-1",
+			},
+			Spec: velerov1api.BackupStorageLocationSpec{
+				Provider: "objStoreProvider",
+				StorageType: velerov1api.StorageType{
+					ObjectStorage: &velerov1api.ObjectStorageLocation{
+						Bucket: "bucket-1",
+					},
+				},
+			},
+		},
+		{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: namespace,
+				Name:      "the-really-long-location-name-that-is-much-more-than-63-characters-2",
+			},
+			Spec: velerov1api.BackupStorageLocationSpec{
+				Provider: "objStoreProvider",
+				StorageType: velerov1api.StorageType{
+					ObjectStorage: &velerov1api.ObjectStorageLocation{
+						Bucket: "bucket-2",
+					},
+				},
+			},
+		},
+	}
+}
+
 func TestBackupSyncControllerRun(t *testing.T) {
 	tests := []struct {
-		name            string
-		namespace       string
-		locations       []*velerov1api.BackupStorageLocation
-		cloudBackups    map[string][]*velerov1api.Backup
-		existingBackups []*velerov1api.Backup
+		name                    string
+		namespace               string
+		locations               []*velerov1api.BackupStorageLocation
+		cloudBackups            map[string][]*velerov1api.Backup
+		existingBackups         []*velerov1api.Backup
+		longLocationNameEnabled bool
 	}{
 		{
 			name: "no cloud backups",
@@ -163,6 +199,21 @@ func TestBackupSyncControllerRun(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:                    "backup storage location names and labels get updated with location name greater than 63 chars",
+			namespace:               "ns-1",
+			locations:               defaultLocationsListWithLongerLocationName("ns-1"),
+			longLocationNameEnabled: true,
+			cloudBackups: map[string][]*velerov1api.Backup{
+				"bucket-1": {
+					velerotest.NewTestBackup().WithNamespace("ns-1").WithName("backup-1").WithStorageLocation("foo").WithLabel(velerov1api.StorageLocationLabel, "foo").Backup,
+					velerotest.NewTestBackup().WithNamespace("ns-1").WithName("backup-2").Backup,
+				},
+				"bucket-2": {
+					velerotest.NewTestBackup().WithNamespace("ns-1").WithName("backup-3").WithStorageLocation("bar").WithLabel(velerov1api.StorageLocationLabel, "bar").Backup,
+				},
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -259,7 +310,13 @@ func TestBackupSyncControllerRun(t *testing.T) {
 					} else {
 						// verify that the storage location field and label are set properly
 						assert.Equal(t, location.Name, obj.Spec.StorageLocation)
-						assert.Equal(t, location.Name, obj.Labels[velerov1api.StorageLocationLabel])
+
+						locationName := location.Name
+						if test.longLocationNameEnabled {
+							locationName = label.GetValidName(locationName)
+						}
+						assert.Equal(t, locationName, obj.Labels[velerov1api.StorageLocationLabel])
+						assert.Equal(t, true, len(obj.Labels[velerov1api.StorageLocationLabel]) <= validation.DNS1035LabelMaxLength)
 					}
 				}
 			}
@@ -402,6 +459,84 @@ func TestDeleteOrphanedBackups(t *testing.T) {
 	}
 }
 
+func TestStorageLabelsInDeleteOrphanedBackups(t *testing.T) {
+	longLabelName := "the-really-long-location-name-that-is-much-more-than-63-characters"
+	tests := []struct {
+		name            string
+		cloudBackups    sets.String
+		k8sBackups      []*velerotest.TestBackup
+		namespace       string
+		expectedDeletes sets.String
+	}{
+		{
+			name:         "some overlapping backups",
+			namespace:    "ns-1",
+			cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
+			k8sBackups: []*velerotest.TestBackup{
+				velerotest.NewTestBackup().WithNamespace("ns-1").WithName("backup-1").
+					WithLabel(velerov1api.StorageLocationLabel, "the-really-long-location-name-that-is-much-more-than-63-c69e779").WithPhase(velerov1api.BackupPhaseCompleted),
+				velerotest.NewTestBackup().WithNamespace("ns-1").WithName("backup-2").
+					WithLabel(velerov1api.StorageLocationLabel, "the-really-long-location-name-that-is-much-more-than-63-c69e779").WithPhase(velerov1api.BackupPhaseCompleted),
+				velerotest.NewTestBackup().WithNamespace("ns-1").WithName("backup-C").
+					WithLabel(velerov1api.StorageLocationLabel, "the-really-long-location-name-that-is-much-more-than-63-c69e779").WithPhase(velerov1api.BackupPhaseCompleted),
+			},
+			expectedDeletes: sets.NewString("backup-C"),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var (
+				client          = fake.NewSimpleClientset()
+				sharedInformers = informers.NewSharedInformerFactory(client, 0)
+			)
+
+			c := NewBackupSyncController(
+				client.VeleroV1(),
+				client.VeleroV1(),
+				sharedInformers.Velero().V1().Backups(),
+				sharedInformers.Velero().V1().BackupStorageLocations(),
+				time.Duration(0),
+				test.namespace,
+				"",
+				nil, // new plugin manager func
+				velerotest.NewLogger(),
+			).(*backupSyncController)
+
+			expectedDeleteActions := make([]core.Action, 0)
+
+			for _, backup := range test.k8sBackups {
+				// add test backup to informer
+				require.NoError(t, sharedInformers.Velero().V1().Backups().Informer().GetStore().Add(backup.Backup), "Error adding backup to informer")
+
+				// add test backup to client
+				_, err := client.VeleroV1().Backups(test.namespace).Create(backup.Backup)
+				require.NoError(t, err, "Error adding backup to clientset")
+
+				// if we expect this backup to be deleted, set up the expected DeleteAction
+				if test.expectedDeletes.Has(backup.Name) {
+					actionDelete := core.NewDeleteAction(
+						velerov1api.SchemeGroupVersion.WithResource("backups"),
+						test.namespace,
+						backup.Name,
+					)
+					expectedDeleteActions = append(expectedDeleteActions, actionDelete)
+				}
+			}
+
+			c.deleteOrphanedBackups(longLabelName, test.cloudBackups, velerotest.NewLogger())
+
+			numBackups, err := numBackups(t, client, c.namespace)
+			assert.NoError(t, err)
+
+			expected := len(test.k8sBackups) - len(test.expectedDeletes)
+			assert.Equal(t, expected, numBackups)
+
+			velerotest.CompareActions(t, expectedDeleteActions, getDeleteActions(client.Actions()))
+		})
+	}
+}
+
 func TestShouldSync(t *testing.T) {
 	c := clock.NewFakeClock(time.Now())
 

pkg/controller/gc_controller.go

@@ -31,6 +31,7 @@ import (
 	velerov1client "github.com/heptio/velero/pkg/generated/clientset/versioned/typed/velero/v1"
 	informers "github.com/heptio/velero/pkg/generated/informers/externalversions/velero/v1"
 	listers "github.com/heptio/velero/pkg/generated/listers/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 )
 
 const (
@@ -133,7 +134,7 @@ func (c *gcController) processQueueItem(key string) error {
 	log.Info("Backup has expired")
 
 	selector := labels.SelectorFromSet(labels.Set(map[string]string{
-		velerov1api.BackupNameLabel: backup.Name,
+		velerov1api.BackupNameLabel: label.GetValidName(backup.Name),
 		velerov1api.BackupUIDLabel:  string(backup.UID),
 	}))
 

pkg/label/label.go

@@ -0,0 +1,45 @@
+/*
+Copyright 2019 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 label
+
+import (
+	"crypto/sha256"
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/util/validation"
+)
+
+// GetValidName converts an input string to valid kubernetes label string in accordance to rfc1035 DNS Label spec
+// (https://github.com/kubernetes/community/blob/master/contributors/design-proposals/architecture/identifiers.md)
+// Length of the label is adjusted basis the DNS1035LabelMaxLength (defined at k8s.io/apimachinery/pkg/util/validation)
+// If length exceeds, we trim the label name to contain only max allowed characters
+// Additionally, the last 6 characters of the label name are replaced by the first 6 characters of the sha256 of original label
+func GetValidName(label string) string {
+	if len(label) <= validation.DNS1035LabelMaxLength {
+		return label
+	}
+
+	sha := sha256.Sum256([]byte(label))
+	strSha := fmt.Sprintf("%x", sha)
+	charsFromLabel := validation.DNS1035LabelMaxLength - 6
+	if charsFromLabel < 0 {
+		// Derive the label name from sha hash in case the DNS1035LabelMaxLength is less than 6
+		return string(strSha[validation.DNS1035LabelMaxLength])
+	}
+
+	return label[:charsFromLabel] + strSha[:6]
+}

pkg/label/label_test.go

@@ -0,0 +1,49 @@
+/*
+Copyright 2019 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 label
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetValidLabelName(t *testing.T) {
+	tests := []struct {
+		name          string
+		label         string
+		expectedLabel string
+	}{
+		{
+			name:          "valid label name should not be modified",
+			label:         "short label value",
+			expectedLabel: "short label value",
+		},
+		{
+			name:          "label with more than 63 characters should be modified",
+			label:         "this_is_a_very_long_label_value_that_will_be_rejected_by_Kubernetes",
+			expectedLabel: "this_is_a_very_long_label_value_that_will_be_rejected_by_8d0722",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			labelVal := GetValidName(test.label)
+			assert.Equal(t, test.expectedLabel, labelVal)
+		})
+	}
+}

pkg/restic/backupper.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	velerov1api "github.com/heptio/velero/pkg/apis/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/util/boolptr"
 )
 
@@ -233,7 +234,7 @@ func newPodVolumeBackup(backup *velerov1api.Backup, pod *corev1api.Pod, volumeNa
 				},
 			},
 			Labels: map[string]string{
-				velerov1api.BackupNameLabel: backup.Name,
+				velerov1api.BackupNameLabel: label.GetValidName(backup.Name),
 				velerov1api.BackupUIDLabel:  string(backup.UID),
 			},
 		},

pkg/restic/common.go

@@ -30,6 +30,7 @@ import (
 	velerov1api "github.com/heptio/velero/pkg/apis/velero/v1"
 	"github.com/heptio/velero/pkg/cloudprovider/azure"
 	velerov1listers "github.com/heptio/velero/pkg/generated/listers/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/util/filesystem"
 )
 
@@ -125,7 +126,7 @@ type SnapshotIdentifier struct {
 // a given Velero backup.
 func GetSnapshotsInBackup(backup *velerov1api.Backup, podVolumeBackupLister velerov1listers.PodVolumeBackupLister) ([]SnapshotIdentifier, error) {
 	selector := labels.Set(map[string]string{
-		velerov1api.BackupNameLabel: backup.Name,
+		velerov1api.BackupNameLabel: label.GetValidName(backup.Name),
 	}).AsSelector()
 
 	podVolumeBackups, err := podVolumeBackupLister.List(selector)
@@ -189,7 +190,7 @@ func TempCredentialsFile(secretLister corev1listers.SecretLister, veleroNamespac
 // find PodVolumeBackups for the backup identified by name.
 func NewPodVolumeBackupListOptions(name string) metav1.ListOptions {
 	return metav1.ListOptions{
-		LabelSelector: fmt.Sprintf("%s=%s", velerov1api.BackupNameLabel, name),
+		LabelSelector: fmt.Sprintf("%s=%s", velerov1api.BackupNameLabel, label.GetValidName(name)),
 	}
 }
 
@@ -197,7 +198,7 @@ func NewPodVolumeBackupListOptions(name string) metav1.ListOptions {
 // find PodVolumeRestores for the restore identified by name.
 func NewPodVolumeRestoreListOptions(name string) metav1.ListOptions {
 	return metav1.ListOptions{
-		LabelSelector: fmt.Sprintf("%s=%s", velerov1api.RestoreNameLabel, name),
+		LabelSelector: fmt.Sprintf("%s=%s", velerov1api.RestoreNameLabel, label.GetValidName(name)),
 	}
 }
 

pkg/restic/common_test.go

@@ -215,9 +215,10 @@ func TestGetVolumesToBackup(t *testing.T) {
 
 func TestGetSnapshotsInBackup(t *testing.T) {
 	tests := []struct {
-		name             string
-		podVolumeBackups []velerov1api.PodVolumeBackup
-		expected         []SnapshotIdentifier
+		name                  string
+		podVolumeBackups      []velerov1api.PodVolumeBackup
+		expected              []SnapshotIdentifier
+		longBackupNameEnabled bool
 	}{
 		{
 			name:             "no pod volume backups",
@@ -294,6 +295,53 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:                  "some pod volume backups with matching label and backup name greater than 63 chars",
+			longBackupNameEnabled: true,
+			podVolumeBackups: []velerov1api.PodVolumeBackup{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "foo", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-1"}},
+					Spec: velerov1api.PodVolumeBackupSpec{
+						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
+					},
+					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-1"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "bar", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-2"}},
+					Spec: velerov1api.PodVolumeBackupSpec{
+						Pod: corev1api.ObjectReference{Name: "pod-2", Namespace: "ns-2"},
+					},
+					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-2"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "completed-pvb", Labels: map[string]string{velerov1api.BackupNameLabel: "the-really-long-backup-name-that-is-much-more-than-63-cha6ca4bc"}},
+					Spec: velerov1api.PodVolumeBackupSpec{
+						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
+					},
+					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-3"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "completed-pvb-2", Labels: map[string]string{velerov1api.BackupNameLabel: "backup-1"}},
+					Spec: velerov1api.PodVolumeBackupSpec{
+						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
+					},
+					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-4"},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "incomplete-or-failed-pvb", Labels: map[string]string{velerov1api.BackupNameLabel: "backup-1"}},
+					Spec: velerov1api.PodVolumeBackupSpec{
+						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-2"},
+					},
+					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: ""},
+				},
+			},
+			expected: []SnapshotIdentifier{
+				{
+					VolumeNamespace: "ns-1",
+					SnapshotID:      "snap-3",
+				},
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -307,6 +355,10 @@ func TestGetSnapshotsInBackup(t *testing.T) {
 
 			veleroBackup.Name = "backup-1"
 
+			if test.longBackupNameEnabled {
+				veleroBackup.Name = "the-really-long-backup-name-that-is-much-more-than-63-characters"
+			}
+
 			for _, pvb := range test.podVolumeBackups {
 				require.NoError(t, pvbInformer.Informer().GetStore().Add(pvb.DeepCopy()))
 			}

pkg/restic/repository_ensurer.go

@@ -32,6 +32,7 @@ import (
 	velerov1client "github.com/heptio/velero/pkg/generated/clientset/versioned/typed/velero/v1"
 	velerov1informers "github.com/heptio/velero/pkg/generated/informers/externalversions/velero/v1"
 	velerov1listers "github.com/heptio/velero/pkg/generated/listers/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 )
 
 // repositoryEnsurer ensures that Velero restic repositories are created and ready.
@@ -99,8 +100,8 @@ func newRepositoryEnsurer(repoInformer velerov1informers.ResticRepositoryInforme
 
 func repoLabels(volumeNamespace, backupLocation string) labels.Set {
 	return map[string]string{
-		velerov1api.ResticVolumeNamespaceLabel: volumeNamespace,
-		velerov1api.StorageLocationLabel:       backupLocation,
+		velerov1api.ResticVolumeNamespaceLabel: label.GetValidName(volumeNamespace),
+		velerov1api.StorageLocationLabel:       label.GetValidName(backupLocation),
 	}
 }
 

pkg/restic/restorer.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	velerov1api "github.com/heptio/velero/pkg/apis/velero/v1"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/util/boolptr"
 )
 
@@ -156,7 +157,7 @@ func newPodVolumeRestore(restore *velerov1api.Restore, pod *corev1api.Pod, volum
 				},
 			},
 			Labels: map[string]string{
-				velerov1api.RestoreNameLabel: restore.Name,
+				velerov1api.RestoreNameLabel: label.GetValidName(restore.Name),
 				velerov1api.RestoreUIDLabel:  string(restore.UID),
 				velerov1api.PodUIDLabel:      string(pod.UID),
 			},

pkg/restore/restore.go

@@ -47,6 +47,7 @@ import (
 	"github.com/heptio/velero/pkg/discovery"
 	listers "github.com/heptio/velero/pkg/generated/listers/velero/v1"
 	"github.com/heptio/velero/pkg/kuberesource"
+	"github.com/heptio/velero/pkg/label"
 	"github.com/heptio/velero/pkg/plugin/velero"
 	"github.com/heptio/velero/pkg/restic"
 	"github.com/heptio/velero/pkg/util/collections"
@@ -1106,8 +1107,8 @@ func addRestoreLabels(obj metav1.Object, restoreName, backupName string) {
 		labels = make(map[string]string)
 	}
 
-	labels[api.BackupNameLabel] = backupName
-	labels[api.RestoreNameLabel] = restoreName
+	labels[api.BackupNameLabel] = label.GetValidName(backupName)
+	labels[api.RestoreNameLabel] = label.GetValidName(restoreName)
 
 	obj.SetLabels(labels)
 }

pkg/restore/restore_test.go

@@ -675,6 +675,130 @@ func TestRestoreResourceForNamespace(t *testing.T) {
 	}
 }
 
+func TestRestoreLabels(t *testing.T) {
+	tests := []struct {
+		name                    string
+		namespace               string
+		resourcePath            string
+		backupName              string
+		restoreName             string
+		labelSelector           labels.Selector
+		includeClusterResources *bool
+		fileSystem              *velerotest.FakeFileSystem
+		actions                 []resolvedAction
+		expectedErrors          Result
+		expectedObjs            []unstructured.Unstructured
+	}{
+		{
+			name:          "backup name and restore name less than 63 characters",
+			namespace:     "ns-1",
+			resourcePath:  "configmaps",
+			backupName:    "less-than-63-characters",
+			restoreName:   "less-than-63-characters-12345",
+			labelSelector: labels.NewSelector(),
+			fileSystem: velerotest.NewFakeFileSystem().
+				WithFile("configmaps/cm-1.json", newNamedTestConfigMap("cm-1").ToJSON()),
+			expectedObjs: toUnstructured(
+				newNamedTestConfigMap("cm-1").WithLabels(map[string]string{
+					api.BackupNameLabel:  "less-than-63-characters",
+					api.RestoreNameLabel: "less-than-63-characters-12345",
+				}).ConfigMap,
+			),
+		},
+		{
+			name:          "backup name equal to 63 characters",
+			namespace:     "ns-1",
+			resourcePath:  "configmaps",
+			backupName:    "the-really-long-kube-service-name-that-is-exactly-63-characters",
+			restoreName:   "the-really-long-kube-service-name-that-is-exactly-63-characters-12345",
+			labelSelector: labels.NewSelector(),
+			fileSystem: velerotest.NewFakeFileSystem().
+				WithFile("configmaps/cm-1.json", newNamedTestConfigMap("cm-1").ToJSON()),
+			expectedObjs: toUnstructured(
+				newNamedTestConfigMap("cm-1").WithLabels(map[string]string{
+					api.BackupNameLabel:  "the-really-long-kube-service-name-that-is-exactly-63-characters",
+					api.RestoreNameLabel: "the-really-long-kube-service-name-that-is-exactly-63-char0871f3",
+				}).ConfigMap,
+			),
+		},
+		{
+			name:          "backup name greter than 63 characters",
+			namespace:     "ns-1",
+			resourcePath:  "configmaps",
+			backupName:    "the-really-long-kube-service-name-that-is-much-greater-than-63-characters",
+			restoreName:   "the-really-long-kube-service-name-that-is-much-greater-than-63-characters-12345",
+			labelSelector: labels.NewSelector(),
+			fileSystem: velerotest.NewFakeFileSystem().
+				WithFile("configmaps/cm-1.json", newNamedTestConfigMap("cm-1").ToJSON()),
+			expectedObjs: toUnstructured(
+				newNamedTestConfigMap("cm-1").WithLabels(map[string]string{
+					api.BackupNameLabel:  "the-really-long-kube-service-name-that-is-much-greater-th8a11b3",
+					api.RestoreNameLabel: "the-really-long-kube-service-name-that-is-much-greater-th1bf26f",
+				}).ConfigMap,
+			),
+		},
+	}
+
+	var (
+		client                 = fake.NewSimpleClientset()
+		sharedInformers        = informers.NewSharedInformerFactory(client, 0)
+		snapshotLocationLister = sharedInformers.Velero().V1().VolumeSnapshotLocations().Lister()
+	)
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resourceClient := &velerotest.FakeDynamicClient{}
+			for i := range test.expectedObjs {
+				resourceClient.On("Create", &test.expectedObjs[i]).Return(&test.expectedObjs[i], nil)
+			}
+
+			dynamicFactory := &velerotest.FakeDynamicFactory{}
+			gv := schema.GroupVersion{Group: "", Version: "v1"}
+
+			configMapResource := metav1.APIResource{Name: "configmaps", Namespaced: true}
+			dynamicFactory.On("ClientForGroupVersionResource", gv, configMapResource, test.namespace).Return(resourceClient, nil)
+
+			ctx := &context{
+				dynamicFactory: dynamicFactory,
+				actions:        test.actions,
+				fileSystem:     test.fileSystem,
+				selector:       test.labelSelector,
+				restore: &api.Restore{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: api.DefaultNamespace,
+						Name:      test.restoreName,
+					},
+					Spec: api.RestoreSpec{
+						IncludeClusterResources: test.includeClusterResources,
+						BackupName:              test.backupName,
+					},
+				},
+				backup: &api.Backup{},
+				log:    velerotest.NewLogger(),
+				pvRestorer: &pvRestorer{
+					logger: logging.DefaultLogger(logrus.DebugLevel),
+					volumeSnapshotterGetter: &fakeVolumeSnapshotterGetter{
+						volumeMap: map[velerotest.VolumeBackupInfo]string{{SnapshotID: "snap-1"}: "volume-1"},
+						volumeID:  "volume-1",
+					},
+					snapshotLocationLister: snapshotLocationLister,
+					backup:                 &api.Backup{},
+				},
+				applicableActions: make(map[schema.GroupResource][]resolvedAction),
+				resourceClients:   make(map[resourceClientKey]pkgclient.Dynamic),
+				restoredItems:     make(map[velero.ResourceIdentifier]struct{}),
+			}
+
+			warnings, errors := ctx.restoreResource(test.resourcePath, test.namespace, test.resourcePath)
+
+			assert.Empty(t, warnings.Velero)
+			assert.Empty(t, warnings.Cluster)
+			assert.Empty(t, warnings.Namespaces)
+			assert.Equal(t, test.expectedErrors, errors)
+		})
+	}
+}
+
 func TestRestoringExistingServiceAccount(t *testing.T) {
 	fromCluster := newTestServiceAccount()
 	fromClusterUnstructured, err := runtime.DefaultUnstructuredConverter.ToUnstructured(fromCluster.ServiceAccount)