Merge pull request kubernetes-csi#78 from davidz627/feature/csiMigration

Add CSI Migration logic for attach/detach

Gopkg.toml

@@ -26,7 +26,7 @@
 
 [[constraint]]
   name = "k8s.io/api"
-  version = "kubernetes-1.12.0"
+  version = "kubernetes-1.12.1"
 
 [[constraint]]
   name = "k8s.io/apimachinery"

pkg/controller/csi_handler.go

@@ -34,6 +34,8 @@ import (
 	csilisters "k8s.io/csi-api/pkg/client/listers/csi/v1alpha1"
 
 	"github.com/kubernetes-csi/external-attacher/pkg/connection"
+
+	csiMigration "github.com/kubernetes-csi/kubernetes-csi-migration-library"
 )
 
 // csiHandler is a handler that calls CSI to attach/detach volume.
@@ -213,6 +215,22 @@ func (h *csiHandler) hasVAFinalizer(va *storage.VolumeAttachment) bool {
 	return false
 }
 
+func getCSISource(pv *v1.PersistentVolume) (*v1.CSIPersistentVolumeSource, error) {
+	if pv == nil {
+		return nil, fmt.Errorf("could not get CSI source, pv was nil")
+	}
+	if pv.Spec.CSI != nil {
+		return pv.Spec.CSI, nil
+	} else if csiMigration.IsPVMigrated(pv) {
+		csiPV, err := csiMigration.TranslateInTreePVToCSI(pv)
+		if err != nil {
+			return nil, fmt.Errorf("failed to translate in tree pv to CSI: %v", err)
+		}
+		return csiPV.Spec.CSI, nil
+	}
+	return nil, fmt.Errorf("pv contained non-csi source that was not migrated")
+}
+
 func (h *csiHandler) csiAttach(va *storage.VolumeAttachment) (*storage.VolumeAttachment, map[string]string, error) {
 	glog.V(4).Infof("Starting attach operation for %q", va.Name)
 	// Check as much as possible before adding VA finalizer - it would block
@@ -235,20 +253,25 @@ func (h *csiHandler) csiAttach(va *storage.VolumeAttachment) (*storage.VolumeAtt
 		return va, nil, fmt.Errorf("could not add PersistentVolume finalizer: %s", err)
 	}
 
-	attributes, err := GetVolumeAttributes(pv)
+	csiSource, err := getCSISource(pv)
 	if err != nil {
 		return va, nil, err
 	}
 
-	volumeHandle, readOnly, err := GetVolumeHandle(pv)
+	attributes, err := GetVolumeAttributes(csiSource)
 	if err != nil {
 		return va, nil, err
 	}
-	volumeCapabilities, err := GetVolumeCapabilities(pv)
+
+	volumeHandle, readOnly, err := GetVolumeHandle(csiSource)
 	if err != nil {
 		return va, nil, err
 	}
-	secrets, err := h.getCredentialsFromPV(pv)
+	volumeCapabilities, err := GetVolumeCapabilities(pv, csiSource)
+	if err != nil {
+		return va, nil, err
+	}
+	secrets, err := h.getCredentialsFromPV(csiSource)
 	if err != nil {
 		return va, nil, err
 	}
@@ -284,11 +307,17 @@ func (h *csiHandler) csiDetach(va *storage.VolumeAttachment) (*storage.VolumeAtt
 	if err != nil {
 		return va, err
 	}
-	volumeHandle, _, err := GetVolumeHandle(pv)
+
+	csiSource, err := getCSISource(pv)
+	if err != nil {
+		return va, err
+	}
+
+	volumeHandle, _, err := GetVolumeHandle(csiSource)
 	if err != nil {
 		return va, err
 	}
-	secrets, err := h.getCredentialsFromPV(pv)
+	secrets, err := h.getCredentialsFromPV(csiSource)
 	if err != nil {
 		return va, err
 	}
@@ -419,11 +448,11 @@ func (h *csiHandler) SyncNewOrUpdatedPersistentVolume(pv *v1.PersistentVolume) {
 	return
 }
 
-func (h *csiHandler) getCredentialsFromPV(pv *v1.PersistentVolume) (map[string]string, error) {
-	if pv.Spec.PersistentVolumeSource.CSI == nil {
-		return nil, fmt.Errorf("persistent volume does not contain CSI volume source")
+func (h *csiHandler) getCredentialsFromPV(csiSource *v1.CSIPersistentVolumeSource) (map[string]string, error) {
+	if csiSource == nil {
+		return nil, fmt.Errorf("CSI volume source was nil")
 	}
-	secretRef := pv.Spec.PersistentVolumeSource.CSI.ControllerPublishSecretRef
+	secretRef := csiSource.ControllerPublishSecretRef
 	if secretRef == nil {
 		return nil, nil
 	}

pkg/controller/csi_handler_test.go

@@ -77,6 +77,36 @@ func pv() *v1.PersistentVolume {
 	}
 }
 
+func gcePDPV() *v1.PersistentVolume {
+	return &v1.PersistentVolume{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: testPVName,
+			Labels: map[string]string{
+				"failure-domain.beta.kubernetes.io/zone": "testZone",
+			},
+		},
+		Spec: v1.PersistentVolumeSpec{
+			PersistentVolumeSource: v1.PersistentVolumeSource{
+				GCEPersistentDisk: &v1.GCEPersistentDiskVolumeSource{
+					PDName:    "testpd",
+					FSType:    "ext4",
+					Partition: 0,
+					ReadOnly:  false,
+				},
+			},
+			AccessModes: []v1.PersistentVolumeAccessMode{
+				v1.ReadWriteOnce,
+			},
+		},
+	}
+}
+
+func gcePDPVWithFinalizer() *v1.PersistentVolume {
+	pv := gcePDPV()
+	pv.Finalizers = []string{fin}
+	return pv
+}
+
 func pvWithFinalizer() *v1.PersistentVolume {
 	pv := pv()
 	pv.Finalizers = []string{fin}
@@ -537,6 +567,20 @@ func TestCSIHandler(t *testing.T) {
 				{"attach", testVolumeHandle, testNodeID, noAttrs, noSecrets, success, notDetached, noMetadata, 0},
 			},
 		},
+		{
+			name:           "VolumeAttachment with GCEPersistentDiskVolumeSource -> successful attachment",
+			initialObjects: []runtime.Object{gcePDPVWithFinalizer(), node()},
+			addedVA:        va(false /*attached*/, "" /*finalizer*/),
+			expectedActions: []core.Action{
+				// Finalizer is saved first
+				core.NewUpdateAction(vaGroupResourceVersion, metav1.NamespaceNone, va(false /*attached*/, fin)),
+				core.NewUpdateAction(vaGroupResourceVersion, metav1.NamespaceNone, va(true /*attached*/, fin)),
+			},
+			expectedCSICalls: []csiCall{
+				{"attach", "projects/UNSPECIFIED/zones/testZone/disks/testpd", testNodeID,
+					map[string]string{"partition": "0"}, noSecrets, success, notDetached, noMetadata, 0},
+			},
+		},
 		//
 		// DETACH
 		//
@@ -713,7 +757,19 @@ func TestCSIHandler(t *testing.T) {
 				{"detach", testVolumeHandle, testNodeID, noAttrs, noSecrets, success, detached, noMetadata, 0},
 			},
 		},
-
+		{
+			name:           "VolumeAttachment with GCEPersistentDiskVolumeSource marked for deletion -> successful detach",
+			initialObjects: []runtime.Object{gcePDPVWithFinalizer(), node()},
+			addedVA:        deleted(va(true /*attached*/, fin /*finalizer*/)),
+			expectedActions: []core.Action{
+				// Finalizer is saved first
+				core.NewUpdateAction(vaGroupResourceVersion, metav1.NamespaceNone, deleted(va(false /*attached*/, ""))),
+			},
+			expectedCSICalls: []csiCall{
+				{"detach", "projects/UNSPECIFIED/zones/testZone/disks/testpd", testNodeID,
+					map[string]string{"partition": "0"}, noSecrets, success, detached, noMetadata, 0},
+			},
+		},
 		//
 		// PV finalizers
 		//

pkg/controller/util.go

@@ -131,17 +131,17 @@ func GetNodeIDFromNodeInfo(driver string, nodeInfo *csiapi.CSINodeInfo) (string,
 	return "", false
 }
 
-func GetVolumeCapabilities(pv *v1.PersistentVolume) (*csi.VolumeCapability, error) {
+func GetVolumeCapabilities(pv *v1.PersistentVolume, csiSource *v1.CSIPersistentVolumeSource) (*csi.VolumeCapability, error) {
 	m := map[v1.PersistentVolumeAccessMode]bool{}
 	for _, mode := range pv.Spec.AccessModes {
 		m[mode] = true
 	}
 
-	if pv.Spec.PersistentVolumeSource.CSI == nil {
-		return nil, fmt.Errorf("persistent volume does not contain CSI volume source")
+	if csiSource == nil {
+		return nil, fmt.Errorf("CSI volume source was nil")
 	}
 
-	fsType := pv.Spec.CSI.FSType
+	fsType := csiSource.FSType
 	if len(fsType) == 0 {
 		fsType = defaultFSType
 	}
@@ -180,16 +180,16 @@ func GetVolumeCapabilities(pv *v1.PersistentVolume) (*csi.VolumeCapability, erro
 	return cap, nil
 }
 
-func GetVolumeHandle(pv *v1.PersistentVolume) (string, bool, error) {
-	if pv.Spec.PersistentVolumeSource.CSI == nil {
-		return "", false, fmt.Errorf("persistent volume does not contain CSI volume source")
+func GetVolumeHandle(csiSource *v1.CSIPersistentVolumeSource) (string, bool, error) {
+	if csiSource == nil {
+		return "", false, fmt.Errorf("csi source was nil")
 	}
-	return pv.Spec.PersistentVolumeSource.CSI.VolumeHandle, pv.Spec.PersistentVolumeSource.CSI.ReadOnly, nil
+	return csiSource.VolumeHandle, csiSource.ReadOnly, nil
 }
 
-func GetVolumeAttributes(pv *v1.PersistentVolume) (map[string]string, error) {
-	if pv.Spec.PersistentVolumeSource.CSI == nil {
-		return nil, fmt.Errorf("persistent volume does not contain CSI volume source")
+func GetVolumeAttributes(csiSource *v1.CSIPersistentVolumeSource) (map[string]string, error) {
+	if csiSource == nil {
+		return nil, fmt.Errorf("csi source was nil")
 	}
-	return pv.Spec.PersistentVolumeSource.CSI.VolumeAttributes, nil
+	return csiSource.VolumeAttributes, nil
 }

pkg/controller/util_test.go

@@ -164,7 +164,7 @@ func TestGetVolumeCapabilities(t *testing.T) {
 				},
 			},
 		}
-		cap, err := GetVolumeCapabilities(pv)
+		cap, err := GetVolumeCapabilities(pv, pv.Spec.CSI)
 
 		if err == nil && test.expectError {
 			t.Errorf("test %s: expected error, got none", test.name)
@@ -272,7 +272,7 @@ func TestGetVolumeHandle(t *testing.T) {
 	}
 
 	for i, test := range tests {
-		output, readOnly, err := GetVolumeHandle(test.pv)
+		output, readOnly, err := GetVolumeHandle(test.pv.Spec.CSI)
 		if output != test.output {
 			t.Errorf("test %d: expected volume ID %q, got %q", i, test.output, output)
 		}