Add check for volumecontent source in CreateVolume

If the volume is already found,we also need to
check the volumecountent source id valid or not.

pkg/hostpath/controllerserver.go

@@ -129,6 +129,24 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		// request
 		if exVol.VolSize >= int64(req.GetCapacityRange().GetRequiredBytes()) {
 			// exisiting volume is compatible with new request and should be reused.
+			if req.GetVolumeContentSource() != nil {
+				volumeSource := req.VolumeContentSource
+				switch volumeSource.Type.(type) {
+				case *csi.VolumeContentSource_Snapshot:
+					if volumeSource.GetSnapshot() != nil && exVol.ParentSnapID != volumeSource.GetSnapshot().GetSnapshotId() {
+						err = fmt.Errorf("existing volume source snapshot id not matching")
+					}
+				case *csi.VolumeContentSource_Volume:
+					if volumeSource.GetVolume() != nil && exVol.ParentVolID != volumeSource.GetVolume().GetVolumeId() {
+						err = fmt.Errorf("existing volume source volume id not matching")
+					}
+				default:
+					err = fmt.Errorf("%v not a proper volume source", volumeSource)
+				}
+				if err != nil {
+					return nil, status.Error(codes.InvalidArgument, err.Error())
+				}
+			}
 			// TODO (sbezverk) Do I need to make sure that volume still exists?
 			return &csi.CreateVolumeResponse{
 				Volume: &csi.Volume{
@@ -143,7 +161,6 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	}
 
 	volumeID := uuid.NewUUID().String()
-	path := getVolumePath(volumeID)
 
 	vol, err := createHostpathVolume(volumeID, req.GetName(), capacity, requestedAccessType, false /* ephemeral */)
 	if err != nil {
@@ -152,12 +169,21 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	glog.V(4).Infof("created volume %s at path %s", vol.VolID, vol.VolPath)
 
 	if req.GetVolumeContentSource() != nil {
-		contentSource := req.GetVolumeContentSource()
-		if snapshot := contentSource.GetSnapshot(); snapshot != nil {
-			err = loadFromSnapshot(snapshot.GetSnapshotId(), path)
-		}
-		if srcVolume := contentSource.GetVolume(); srcVolume != nil {
-			err = loadFromVolume(srcVolume.GetVolumeId(), path)
+		path := getVolumePath(volumeID)
+		volumeSource := req.VolumeContentSource
+		switch volumeSource.Type.(type) {
+		case *csi.VolumeContentSource_Snapshot:
+			if snapshot := volumeSource.GetSnapshot(); snapshot != nil {
+				err = loadFromSnapshot(snapshot.GetSnapshotId(), path)
+				vol.ParentSnapID = snapshot.GetSnapshotId()
+			}
+		case *csi.VolumeContentSource_Volume:
+			if srcVolume := volumeSource.GetVolume(); srcVolume != nil {
+				err = loadFromVolume(srcVolume.GetVolumeId(), path)
+				vol.ParentVolID = srcVolume.GetVolumeId()
+			}
+		default:
+			err = status.Errorf(codes.InvalidArgument, "%v not a proper volume source", volumeSource)
 		}
 		if err != nil {
 			if delErr := deleteHostpathVolume(volumeID); delErr != nil {

pkg/hostpath/hostpath.go

@@ -60,6 +60,8 @@ type hostPathVolume struct {
 	VolSize       int64      `json:"volSize"`
 	VolPath       string     `json:"volPath"`
 	VolAccessType accessType `json:"volAccessType"`
+	ParentVolID   string     `json:"parentVolID,omitempty"`
+	ParentSnapID  string     `json:"parentSnapID,omitempty"`
 	Ephemeral     bool       `json:"ephemeral"`
 }