Append "/mount" to directory for CSI volumes (#1615)

CSI volumes are mounted one level deeper than "native" kubernetes
volumes, and this needs to be appended for proper restic support.

Fixes #1313.

Signed-off-by: Nolan Brubaker <brubakern@vmware.com>

changelogs/unreleased/1615-nrb

@@ -0,0 +1 @@
+Add restic support for CSI volumes

pkg/cmd/cli/restic/server.go

@@ -165,6 +165,7 @@ func (s *resticServer) run() {
 		s.podInformer,
 		s.secretInformer,
 		s.kubeInformerFactory.Core().V1().PersistentVolumeClaims(),
+		s.kubeInformerFactory.Core().V1().PersistentVolumes(),
 		s.veleroInformerFactory.Velero().V1().BackupStorageLocations(),
 		os.Getenv("NODE_NAME"),
 	)
@@ -181,6 +182,7 @@ func (s *resticServer) run() {
 		s.podInformer,
 		s.secretInformer,
 		s.kubeInformerFactory.Core().V1().PersistentVolumeClaims(),
+		s.kubeInformerFactory.Core().V1().PersistentVolumes(),
 		s.veleroInformerFactory.Velero().V1().BackupStorageLocations(),
 		os.Getenv("NODE_NAME"),
 	)

pkg/controller/pod_volume_backup_controller.go

@@ -51,6 +51,7 @@ type podVolumeBackupController struct {
 	secretLister          corev1listers.SecretLister
 	podLister             corev1listers.PodLister
 	pvcLister             corev1listers.PersistentVolumeClaimLister
+	pvLister              corev1listers.PersistentVolumeLister
 	backupLocationLister  listers.BackupStorageLocationLister
 	nodeName              string
 
@@ -67,6 +68,7 @@ func NewPodVolumeBackupController(
 	podInformer cache.SharedIndexInformer,
 	secretInformer cache.SharedIndexInformer,
 	pvcInformer corev1informers.PersistentVolumeClaimInformer,
+	pvInformer corev1informers.PersistentVolumeInformer,
 	backupLocationInformer informers.BackupStorageLocationInformer,
 	nodeName string,
 ) Interface {
@@ -77,6 +79,7 @@ func NewPodVolumeBackupController(
 		podLister:             corev1listers.NewPodLister(podInformer.GetIndexer()),
 		secretLister:          corev1listers.NewSecretLister(secretInformer.GetIndexer()),
 		pvcLister:             pvcInformer.Lister(),
+		pvLister:              pvInformer.Lister(),
 		backupLocationLister:  backupLocationInformer.Lister(),
 		nodeName:              nodeName,
 
@@ -191,7 +194,7 @@ func (c *podVolumeBackupController) processBackup(req *velerov1api.PodVolumeBack
 		return c.fail(req, errors.Wrap(err, "error getting pod").Error(), log)
 	}
 
-	volumeDir, err := kube.GetVolumeDirectory(pod, req.Spec.Volume, c.pvcLister)
+	volumeDir, err := kube.GetVolumeDirectory(pod, req.Spec.Volume, c.pvcLister, c.pvLister)
 	if err != nil {
 		log.WithError(err).Error("Error getting volume directory name")
 		return c.fail(req, errors.Wrap(err, "error getting volume directory name").Error(), log)

pkg/controller/pod_volume_restore_controller.go

@@ -55,6 +55,7 @@ type podVolumeRestoreController struct {
 	podLister              corev1listers.PodLister
 	secretLister           corev1listers.SecretLister
 	pvcLister              corev1listers.PersistentVolumeClaimLister
+	pvLister               corev1listers.PersistentVolumeLister
 	backupLocationLister   listers.BackupStorageLocationLister
 	nodeName               string
 
@@ -71,6 +72,7 @@ func NewPodVolumeRestoreController(
 	podInformer cache.SharedIndexInformer,
 	secretInformer cache.SharedIndexInformer,
 	pvcInformer corev1informers.PersistentVolumeClaimInformer,
+	pvInformer corev1informers.PersistentVolumeInformer,
 	backupLocationInformer informers.BackupStorageLocationInformer,
 	nodeName string,
 ) Interface {
@@ -81,6 +83,7 @@ func NewPodVolumeRestoreController(
 		podLister:              corev1listers.NewPodLister(podInformer.GetIndexer()),
 		secretLister:           corev1listers.NewSecretLister(secretInformer.GetIndexer()),
 		pvcLister:              pvcInformer.Lister(),
+		pvLister:               pvInformer.Lister(),
 		backupLocationLister:   backupLocationInformer.Lister(),
 		nodeName:               nodeName,
 
@@ -276,7 +279,7 @@ func (c *podVolumeRestoreController) processRestore(req *velerov1api.PodVolumeRe
 		return c.failRestore(req, errors.Wrap(err, "error getting pod").Error(), log)
 	}
 
-	volumeDir, err := kube.GetVolumeDirectory(pod, req.Spec.Volume, c.pvcLister)
+	volumeDir, err := kube.GetVolumeDirectory(pod, req.Spec.Volume, c.pvcLister, c.pvLister)
 	if err != nil {
 		log.WithError(err).Error("Error getting volume directory name")
 		return c.failRestore(req, errors.Wrap(err, "error getting volume directory name").Error(), log)

pkg/restore/restore_test.go

@@ -1752,7 +1752,7 @@ func TestRestorePersistentVolumes(t *testing.T) {
 				).
 				addItems("persistentvolumeclaims",
 					test.NewPVC("ns-1", "pvc-1",
-						test.WithVolumeName("pv-1"),
+						test.WithPVName("pv-1"),
 						test.WithAnnotations("pv.kubernetes.io/bind-completed", "true", "pv.kubernetes.io/bound-by-controller", "true", "foo", "bar"),
 					),
 				).

pkg/test/resources.go

@@ -249,6 +249,19 @@ func NewNamespace(name string, opts ...ObjectOpts) *corev1.Namespace {
 	return obj
 }
 
+// VolumeOpts exists because corev1.Volume does not implement metav1.Object
+type VolumeOpts func(*corev1.Volume)
+
+func NewVolume(name string, opts ...VolumeOpts) *corev1.Volume {
+	obj := &corev1.Volume{Name: name}
+
+	for _, opt := range opts {
+		opt(obj)
+	}
+
+	return obj
+}
+
 func objectMeta(ns, name string) metav1.ObjectMeta {
 	return metav1.ObjectMeta{
 		Namespace: ns,
@@ -376,15 +389,59 @@ func WithAWSEBSVolumeID(volumeID string) func(obj metav1.Object) {
 	}
 }
 
-// WithVolumeName is a functional option for persistent volume claims that sets the specified
-// volume name. It panics if the object is not a persistent volume claim.
-func WithVolumeName(name string) func(obj metav1.Object) {
+// WithCSI is a functional option for persistent volumes that sets the specified CSI driver name
+// and volume handle. It panics if the object is not a persistent volume.
+func WithCSI(driver, volumeHandle string) func(object metav1.Object) {
+	return func(obj metav1.Object) {
+		pv, ok := obj.(*corev1.PersistentVolume)
+		if !ok {
+			panic("WithCSI is only valid for persistent volumes")
+		}
+		if pv.Spec.CSI == nil {
+			pv.Spec.CSI = new(corev1.CSIPersistentVolumeSource)
+		}
+		pv.Spec.CSI.Driver = driver
+		pv.Spec.CSI.VolumeHandle = volumeHandle
+	}
+}
+
+// WithPVName is a functional option for persistent volume claims that sets the specified
+// persistent volume name. It panics if the object is not a persistent volume claim.
+func WithPVName(name string) func(obj metav1.Object) {
 	return func(obj metav1.Object) {
 		pvc, ok := obj.(*corev1.PersistentVolumeClaim)
 		if !ok {
-			panic("WithVolumeName is only valid for persistent volume claims")
+			panic("WithPVName is only valid for persistent volume claims")
 		}
 
 		pvc.Spec.VolumeName = name
 	}
 }
+
+// WithVolume is a functional option for pods that sets the specified
+// volume on the pod's Spec. It panics if the object is not a pod.
+func WithVolume(volume *corev1.Volume) func(obj metav1.Object) {
+	return func(obj metav1.Object) {
+		pod, ok := obj.(*corev1.Pod)
+		if !ok {
+			panic("WithVolume is only valid for pods")
+		}
+		pod.Spec.Volumes = append(pod.Spec.Volumes, *volume)
+	}
+}
+
+// WithPVCSource is a functional option for volumes that creates a
+// PersistentVolumeClaimVolumeSource with the specified name.
+func WithPVCSource(claimName string) func(vol *corev1.Volume) {
+	return func(vol *corev1.Volume) {
+		vol.VolumeSource.PersistentVolumeClaim = &corev1.PersistentVolumeClaimVolumeSource{ClaimName: claimName}
+	}
+}
+
+// WithCSISource is a functional option for volumes that creates a
+// CSIVolumeSource with the specified driver name.
+func WithCSISource(driverName string) func(vol *corev1.Volume) {
+	return func(vol *corev1.Volume) {
+		vol.VolumeSource.CSI = &corev1.CSIVolumeSource{Driver: driverName}
+	}
+}

pkg/util/kube/utils.go

@@ -93,7 +93,8 @@ func EnsureNamespaceExistsAndIsReady(namespace *corev1api.Namespace, client core
 
 // GetVolumeDirectory gets the name of the directory on the host, under /var/lib/kubelet/pods/<podUID>/volumes/,
 // where the specified volume lives.
-func GetVolumeDirectory(pod *corev1api.Pod, volumeName string, pvcLister corev1listers.PersistentVolumeClaimLister) (string, error) {
+// For volumes with a CSIVolumeSource, append "/mount" to the directory name.
+func GetVolumeDirectory(pod *corev1api.Pod, volumeName string, pvcLister corev1listers.PersistentVolumeClaimLister, pvLister corev1listers.PersistentVolumeLister) (string, error) {
 	var volume *corev1api.Volume
 
 	for _, item := range pod.Spec.Volumes {
@@ -107,14 +108,30 @@ func GetVolumeDirectory(pod *corev1api.Pod, volumeName string, pvcLister corev1l
 		return "", errors.New("volume not found in pod")
 	}
 
+	// This case implies the administrator created the PV and attached it directly, without PVC.
+	// Note that only one VolumeSource can be populated per Volume on a pod
 	if volume.VolumeSource.PersistentVolumeClaim == nil {
+		if volume.VolumeSource.CSI != nil {
+			return volume.Name + "/mount", nil
+		}
 		return volume.Name, nil
 	}
 
+	// Most common case is that we have a PVC VolumeSource, and we need to check the PV it points to for a CSI source.
 	pvc, err := pvcLister.PersistentVolumeClaims(pod.Namespace).Get(volume.VolumeSource.PersistentVolumeClaim.ClaimName)
 	if err != nil {
 		return "", errors.WithStack(err)
 	}
 
+	pv, err := pvLister.Get(pvc.Spec.VolumeName)
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+
+	// PV's been created with a CSI source.
+	if pv.Spec.CSI != nil {
+		return pvc.Spec.VolumeName + "/mount", nil
+	}
+
 	return pvc.Spec.VolumeName, nil
 }

pkg/util/kube/utils_test.go

@@ -20,12 +20,16 @@ import (
 	"testing"
 	"time"
 
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
-	"k8s.io/api/core/v1"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	kubeinformers "k8s.io/client-go/informers"
 
+	"github.com/heptio/velero/pkg/test"
 	velerotest "github.com/heptio/velero/pkg/util/test"
 )
 
@@ -37,7 +41,7 @@ func TestEnsureNamespaceExistsAndIsReady(t *testing.T) {
 	tests := []struct {
 		name           string
 		expectNSFound  bool
-		nsPhase        v1.NamespacePhase
+		nsPhase        corev1.NamespacePhase
 		nsDeleting     bool
 		expectCreate   bool
 		alreadyExists  bool
@@ -51,7 +55,7 @@ func TestEnsureNamespaceExistsAndIsReady(t *testing.T) {
 		{
 			name:           "namespace found, terminating phase",
 			expectNSFound:  true,
-			nsPhase:        v1.NamespaceTerminating,
+			nsPhase:        corev1.NamespaceTerminating,
 			expectedResult: false,
 		},
 		{
@@ -73,14 +77,14 @@ func TestEnsureNamespaceExistsAndIsReady(t *testing.T) {
 		{
 			name:           "namespace not found initially, create returns already exists error, returned namespace is terminating",
 			alreadyExists:  true,
-			nsPhase:        v1.NamespaceTerminating,
+			nsPhase:        corev1.NamespaceTerminating,
 			expectedResult: false,
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			namespace := &v1.Namespace{
+			namespace := &corev1.Namespace{
 				ObjectMeta: metav1.ObjectMeta{
 					Name: "test",
 				},
@@ -102,7 +106,7 @@ func TestEnsureNamespaceExistsAndIsReady(t *testing.T) {
 			if test.expectNSFound {
 				nsClient.On("Get", "test", metav1.GetOptions{}).Return(namespace, nil)
 			} else {
-				nsClient.On("Get", "test", metav1.GetOptions{}).Return(&v1.Namespace{}, k8serrors.NewNotFound(schema.GroupResource{Resource: "namespaces"}, "test"))
+				nsClient.On("Get", "test", metav1.GetOptions{}).Return(&corev1.Namespace{}, k8serrors.NewNotFound(schema.GroupResource{Resource: "namespaces"}, "test"))
 			}
 
 			if test.alreadyExists {
@@ -120,3 +124,79 @@ func TestEnsureNamespaceExistsAndIsReady(t *testing.T) {
 	}
 
 }
+
+type harness struct {
+	*test.APIServer
+
+	log logrus.FieldLogger
+}
+
+func newHarness(t *testing.T) *harness {
+	t.Helper()
+
+	return &harness{
+		APIServer: test.NewAPIServer(t),
+		log:       logrus.StandardLogger(),
+	}
+}
+
+// TestGetVolumeDirectorySuccess tests that the GetVolumeDirectory function
+// returns a volume's name or a volume's name plus '/mount' when a PVC is present.
+func TestGetVolumeDirectorySuccess(t *testing.T) {
+	tests := []struct {
+		name string
+		pod  *corev1.Pod
+		pvc  *corev1.PersistentVolumeClaim
+		pv   *corev1.PersistentVolume
+		want string
+	}{
+		{
+			name: "Non-CSI volume with a PVC/PV returns the volume's name",
+			pod: test.NewPod("ns-1", "my-pod",
+				test.WithVolume(test.NewVolume("my-vol", test.WithPVCSource("my-pvc")))),
+			pvc:  test.NewPVC("ns-1", "my-pvc", test.WithPVName("a-pv")),
+			pv:   test.NewPV("a-pv"),
+			want: "a-pv",
+		},
+		{
+			name: "CSI volume with a PVC/PV appends '/mount' to the volume name",
+			pod: test.NewPod("ns-1", "my-pod",
+				test.WithVolume(test.NewVolume("my-vol", test.WithPVCSource("my-pvc")))),
+			pvc:  test.NewPVC("ns-1", "my-pvc", test.WithPVName("a-pv")),
+			pv:   test.NewPV("a-pv", test.WithCSI("csi.test.com", "provider-volume-id")),
+			want: "a-pv/mount",
+		},
+		{
+			name: "CSI volume mounted without a PVC appends '/mount' to the volume name",
+			pod: test.NewPod("ns-1", "my-pod",
+				test.WithVolume(test.NewVolume("my-vol", test.WithCSISource("csi.test.com")))),
+			want: "my-vol/mount",
+		},
+		{
+			name: "Non-CSI volume without a PVC returns the volume name",
+			pod: test.NewPod("ns-1", "my-pod",
+				test.WithVolume(test.NewVolume("my-vol"))),
+			want: "my-vol",
+		},
+	}
+
+	for _, tc := range tests {
+		h := newHarness(t)
+
+		pvcInformer := kubeinformers.NewSharedInformerFactoryWithOptions(h.KubeClient, 0, kubeinformers.WithNamespace("ns-1")).Core().V1().PersistentVolumeClaims()
+		pvInformer := kubeinformers.NewSharedInformerFactory(h.KubeClient, 0).Core().V1().PersistentVolumes()
+
+		if tc.pvc != nil {
+			require.NoError(t, pvcInformer.Informer().GetStore().Add(tc.pvc))
+		}
+		if tc.pv != nil {
+			require.NoError(t, pvInformer.Informer().GetStore().Add(tc.pv))
+		}
+
+		// Function under test
+		dir, err := GetVolumeDirectory(tc.pod, tc.pod.Spec.Volumes[0].Name, pvcInformer.Lister(), pvInformer.Lister())
+
+		require.NoError(t, err)
+		assert.Equal(t, tc.want, dir)
+	}
+}