Cherry-pick: Verify PV/PVC binding and driver

Manually resolved conflicts in the following files:
	pkg/controller/snapshot_controller.go
	pkg/controller/snapshot_create_test.go
	pkg/controller/snapshot_ready_test.go

Author: xing-yang

Makefile

@@ -16,7 +16,7 @@
 
 REGISTRY_NAME=quay.io/k8scsi
 IMAGE_NAME=csi-snapshotter
-IMAGE_VERSION=v0.4.1
+IMAGE_VERSION=v0.4-canary
 IMAGE_TAG=$(REGISTRY_NAME)/$(IMAGE_NAME):$(IMAGE_VERSION)
 
 REV=$(shell git describe --long --tags --match='v*' --dirty)

pkg/controller/framework_test.go

@@ -875,7 +875,15 @@ func newClaimArray(name, claimUID, capacity, boundToVolume string, phase v1.Pers
 }
 
 // newVolume returns a new volume with given attributes
-func newVolume(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundToClaimName string, phase v1.PersistentVolumePhase, reclaimPolicy v1.PersistentVolumeReclaimPolicy, class string, annotations ...string) *v1.PersistentVolume {
+func newVolume(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundToClaimName string, phase v1.PersistentVolumePhase, reclaimPolicy v1.PersistentVolumeReclaimPolicy, class string, driver string, namespace string, annotations ...string) *v1.PersistentVolume {
+	inDriverName := mockDriverName
+	if driver != "" {
+		inDriverName = driver
+	}
+	inNamespace := testNamespace
+	if namespace != "" {
+		inNamespace = namespace
+	}
 	volume := v1.PersistentVolume{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            name,
@@ -889,7 +897,7 @@ func newVolume(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundTo
 			},
 			PersistentVolumeSource: v1.PersistentVolumeSource{
 				CSI: &v1.CSIPersistentVolumeSource{
-					Driver:       mockDriverName,
+					Driver:       inDriverName,
 					VolumeHandle: volumeHandle,
 				},
 			},
@@ -907,7 +915,7 @@ func newVolume(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundTo
 			Kind:       "PersistentVolumeClaim",
 			APIVersion: "v1",
 			UID:        types.UID(boundToClaimUID),
-			Namespace:  testNamespace,
+			Namespace:  inNamespace,
 			Name:       boundToClaimName,
 		}
 	}
@@ -917,9 +925,9 @@ func newVolume(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundTo
 
 // newVolumeArray returns array with a single volume that would be returned by
 // newVolume() with the same parameters.
-func newVolumeArray(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundToClaimName string, phase v1.PersistentVolumePhase, reclaimPolicy v1.PersistentVolumeReclaimPolicy, class string) []*v1.PersistentVolume {
+func newVolumeArray(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundToClaimName string, phase v1.PersistentVolumePhase, reclaimPolicy v1.PersistentVolumeReclaimPolicy, class string, driver string, namespace string) []*v1.PersistentVolume {
 	return []*v1.PersistentVolume{
-		newVolume(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundToClaimName, phase, reclaimPolicy, class),
+		newVolume(name, volumeUID, volumeHandle, capacity, boundToClaimUID, boundToClaimName, phase, reclaimPolicy, class, driver, namespace),
 	}
 }
 

pkg/controller/snapshot_controller.go

@@ -735,11 +735,40 @@ func (ctrl *csiSnapshotController) getVolumeFromVolumeSnapshot(snapshot *crdv1.V
 		return nil, fmt.Errorf("failed to retrieve PV %s from the API server: %q", pvName, err)
 	}
 
+	// Verify binding between PV/PVC is still valid
+	bound := ctrl.IsVolumeBoundToClaim(pv, pvc)
+	if bound == false {
+		glog.Warningf("binding between PV %s and PVC %s is broken", pvName, pvc.Name)
+		return nil, fmt.Errorf("claim in dataSource not bound or invalid")
+	}
+
+	// Verify driver for PVC is the same as driver for VolumeSnapshot
+	if pv.Spec.PersistentVolumeSource.CSI == nil || pv.Spec.PersistentVolumeSource.CSI.Driver != ctrl.snapshotterName {
+		glog.Warningf("driver for PV %s is different from driver %s for snapshot %s", pvName, ctrl.snapshotterName, snapshot.Name)
+		return nil, fmt.Errorf("claim in dataSource not bound or invalid")
+	}
+
 	glog.V(5).Infof("getVolumeFromVolumeSnapshot: snapshot [%s] PV name [%s]", snapshot.Name, pvName)
 
 	return pv, nil
 }
 
+// IsVolumeBoundToClaim returns true, if given volume is pre-bound or bound
+// to specific claim. Both claim.Name and claim.Namespace must be equal.
+// If claim.UID is present in volume.Spec.ClaimRef, it must be equal too.
+func (ctrl *csiSnapshotController) IsVolumeBoundToClaim(volume *v1.PersistentVolume, claim *v1.PersistentVolumeClaim) bool {
+	if volume.Spec.ClaimRef == nil {
+		return false
+	}
+	if claim.Name != volume.Spec.ClaimRef.Name || claim.Namespace != volume.Spec.ClaimRef.Namespace {
+		return false
+	}
+	if volume.Spec.ClaimRef.UID != "" && claim.UID != volume.Spec.ClaimRef.UID {
+		return false
+	}
+	return true
+}
+
 func (ctrl *csiSnapshotController) getStorageClassFromVolumeSnapshot(snapshot *crdv1.VolumeSnapshot) (*storagev1.StorageClass, error) {
 	// Get storage class from PVC or PV
 	pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot)