Add support for running csi sanity tests.

.travis.yml

@@ -24,6 +24,7 @@ script:
   - go test -v github.com/opensds/nbp/csi/... -cover
   - go test -v github.com/opensds/nbp/flexvolume/... -cover
   - go test -v github.com/opensds/nbp/cindercompatibleapi/... -cover
+  - go test -v github.com/opensds/nbp/csi/server/sanity/... -cover
 
 after_success:
   # Clean OpenSDS northbound plugin built data

Gopkg.toml

@@ -77,6 +77,10 @@
   name = "k8s.io/client-go"
   version = "kubernetes-1.13.0-beta.1"
 
+[[constraint]]
+  name = "github.com/kubernetes-csi/csi-test"
+  version = "1.1.0"
+
 [prune]
   non-go = true
   go-tests = true

csi/server/plugin/opensds/controller.go

@@ -154,6 +154,16 @@ func (p *Plugin) ControllerPublishVolume(ctx context.Context,
 		return nil, status.Error(codes.InvalidArgument, msg)
 	}
 
+	if req.GetVolumeCapability() == nil {
+		msg := "volume capability must be provided"
+		glog.Error(msg)
+		return nil, status.Error(codes.InvalidArgument, msg)
+	}
+
+	if req.GetReadonly() {
+		return nil, status.Error(codes.AlreadyExists, "read only volumes are not supported")
+	}
+
 	glog.V(5).Infof("plugin information %#v", p)
 	glog.V(5).Infof("current storage type: %s", p.PluginStorageType)
 
@@ -193,11 +203,29 @@ func (p *Plugin) ControllerUnpublishVolume(
 }
 
 // ValidateVolumeCapabilities implementation
-func (p *Plugin) ValidateVolumeCapabilities(
-	ctx context.Context,
+func (p *Plugin) ValidateVolumeCapabilities(ctx context.Context,
 	req *csi.ValidateVolumeCapabilitiesRequest) (
 	*csi.ValidateVolumeCapabilitiesResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+
+	volId := req.GetVolumeId()
+	if volId == "" {
+		return nil, status.Error(codes.InvalidArgument, "")
+	}
+
+	if req.GetVolumeCapabilities() == nil {
+		return nil, status.Error(codes.InvalidArgument, "")
+	}
+
+	vol, err := p.Client.GetVolume(volId)
+	if vol == nil || err != nil {
+		return nil, status.Error(codes.NotFound, err.Error())
+	}
+
+	return &csi.ValidateVolumeCapabilitiesResponse{
+		Confirmed: &csi.ValidateVolumeCapabilitiesResponse_Confirmed{
+			VolumeCapabilities: req.VolumeCapabilities,
+		},
+	}, nil
 }
 
 // ListVolumes implementation
@@ -296,6 +324,13 @@ func (p *Plugin) ControllerGetCapabilities(
 					},
 				},
 			},
+			&csi.ControllerServiceCapability{
+				Type: &csi.ControllerServiceCapability_Rpc{
+					Rpc: &csi.ControllerServiceCapability_RPC{
+						Type: csi.ControllerServiceCapability_RPC_PUBLISH_READONLY,
+					},
+				},
+			},
 		},
 	}, nil
 }
@@ -426,12 +461,17 @@ func (p *Plugin) DeleteSnapshot(
 	glog.V(5).Infof("start to delete snapshot, snapshot id: %v, delete snapshot secrets: %v!",
 		req.SnapshotId, req.Secrets)
 
-	if 0 == len(req.SnapshotId) {
+	snpId := req.GetSnapshotId()
+	if snpId == "" {
 		return nil, status.Error(codes.InvalidArgument, "snapshot id cannot be empty")
 	}
 
-	err := p.Client.DeleteVolumeSnapshot(req.SnapshotId, nil)
+	snp, _ := p.Client.GetVolumeSnapshot(snpId)
+	if snp == nil {
+		return &csi.DeleteSnapshotResponse{}, nil
+	}
 
+	err := p.Client.DeleteVolumeSnapshot(req.SnapshotId, nil)
 	if nil != err {
 		msg := fmt.Sprintf("delete snapshot failed: %v", err)
 		glog.Error(msg)
@@ -488,14 +528,14 @@ func (p *Plugin) ListSnapshots(
 		break
 	case (0 == snapshotIDLen) && (0 != sourceVolumeIdLen):
 		if len(snapshotsFilterByVolumeId) <= 0 {
-			return nil, status.Error(codes.NotFound, fmt.Sprintf("no snapshot with source volume id %s", sourceVolumeId))
+			return &csi.ListSnapshotsResponse{Entries: []*csi.ListSnapshotsResponse_Entry{}}, nil
 		}
 
 		filterResult = snapshotsFilterByVolumeId
 		break
 	case (0 != snapshotIDLen) && (0 == sourceVolumeIdLen):
 		if len(snapshotsFilterById) <= 0 {
-			return nil, status.Error(codes.NotFound, fmt.Sprintf("no snapshot with id %s", snapshotId))
+			return &csi.ListSnapshotsResponse{Entries: []*csi.ListSnapshotsResponse_Entry{}}, nil
 		}
 
 		filterResult = snapshotsFilterById
@@ -687,28 +727,24 @@ func (p *Plugin) UnpublishRoutine() {
 
 					if err := p.Client.DeleteVolumeAttachment(act.Id, act); err != nil {
 						glog.Errorf("%s failed to unpublish: %v", act.Id, err)
-					} else {
-						waitAttachmentDeleted(act.Id, func(id string) (interface{}, error) {
-							return p.Client.GetVolumeAttachment(id)
-						}, e)
 					}
 
+					waitAttachmentDeleted(act.Id, func(id string) (interface{}, error) {
+						return p.Client.GetVolumeAttachment(id)
+					}, e)
+
 				// delete fileshare access control list if storage type is file
 				case *model.FileShareAclSpec:
 					act := e.Value.(*model.FileShareAclSpec)
 
 					if err := p.Client.DeleteFileShareAcl(act.Id); err != nil {
-						if strings.Contains(err.Error(), "Not Found") {
-							glog.Infof("delete attachment %s successfully", act.Id)
-							UnpublishAttachmentList.Delete(e)
-						} else {
-							glog.Errorf("%s failed to unpublish: %v", act.Id, err)
-						}
-					} else {
-						waitAttachmentDeleted(act.Id, func(id string) (interface{}, error) {
-							return p.Client.GetFileShareAcl(id)
-						}, e)
+						glog.Errorf("%s failed to unpublish: %v", act.Id, err)
 					}
+
+					waitAttachmentDeleted(act.Id, func(id string) (interface{}, error) {
+						return p.Client.GetFileShareAcl(id)
+					}, e)
+
 				}
 
 				time.Sleep(10 * time.Second)
@@ -726,16 +762,25 @@ func waitAttachmentDeleted(id string, f func(string) (interface{}, error), e *li
 	for {
 		select {
 		case <-ticker.C:
-			_, err := f(id)
-
-			if err != nil && strings.Contains(err.Error(), "Not Found") {
-				glog.Infof("delete attachment %s successfully", id)
-				UnpublishAttachmentList.Delete(e)
-				return
-			} else {
-				glog.Errorf("delete attachment failed: %v", err)
+			v, _ := f(id)
+
+			switch v.(type) {
+			case *model.VolumeAttachmentSpec:
+				if v.(*model.VolumeAttachmentSpec) == nil {
+					glog.Infof("delete attachment %s successfully", id)
+					UnpublishAttachmentList.Delete(e)
+					return
+				}
+			case *model.FileShareAclSpec:
+				if v.(*model.FileShareAclSpec) == nil {
+					glog.Infof("delete attachment %s successfully", id)
+					UnpublishAttachmentList.Delete(e)
+					return
+				}
 			}
 
+			glog.Errorf("delete attachment %#v failed", v)
+
 		case <-timeout:
 			glog.Errorf("waiting to delete %s timeout", id)
 			return
@@ -755,14 +800,14 @@ func extractISCSIInitiatorFromNodeInfo(nodeInfo string) (string, error) {
 }
 
 func extractNvmeofInitiatorFromNodeInfo(nodeInfo string) (string, error) {
-        for _, v := range strings.Split(nodeInfo, ",") {
-                if strings.Contains(v, "nqn") {
-                        glog.V(5).Info("Nvmeof initiator is ", v)
-                        return v, nil
-                }
-        }
-
-         return "", errors.New("no Nvmeof initiators found")
+	for _, v := range strings.Split(nodeInfo, ",") {
+		if strings.Contains(v, "nqn") {
+			glog.V(5).Info("Nvmeof initiator is ", v)
+			return v, nil
+		}
+	}
+
+	return "", errors.New("no Nvmeof initiators found")
 }
 
 func extractFCInitiatorFromNodeInfo(nodeInfo string) ([]string, error) {
@@ -782,6 +827,17 @@ func extractFCInitiatorFromNodeInfo(nodeInfo string) ([]string, error) {
 	return wwpns, nil
 }
 
+func extractIpFromNodeInfo(nodeInfo string) (string, error) {
+	for _, v := range strings.Split(nodeInfo, ",") {
+		ip := net.ParseIP(v)
+		if ip != nil {
+			return v, nil
+		}
+	}
+
+	return "", errors.New("cannot find valid ip address")
+}
+
 func getZone(requirement *csi.TopologyRequirement) string {
 	if requirement == nil {
 		return ""

csi/server/plugin/opensds/fileshare.go

@@ -34,11 +34,15 @@ const (
 )
 
 type FileShare struct {
-	Client *client.Client
+	Client  *client.Client
+	Mounter connector.Mounter
 }
 
-func NewFileshare(c *client.Client) *FileShare {
-	return &FileShare{Client: c}
+func NewFileshare(c *client.Client, Mounter connector.Mounter) *FileShare {
+	return &FileShare{
+		Client:  c,
+		Mounter: Mounter,
+	}
 }
 
 func (f *FileShare) CreateFileShare(req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {

csi/server/plugin/opensds/node.go

@@ -162,7 +162,7 @@ func (p *Plugin) NodeGetInfo(
 	glog.Info("start to get node info")
 	defer glog.Info("end to get node info")
 
-	hostName, err := connector.GetHostName()
+	hostName, err := p.Mounter.GetHostName()
 	if err != nil {
 		msg := fmt.Sprintf("failed to get node name: %v", err)
 		glog.Error(msg)
@@ -171,7 +171,7 @@ func (p *Plugin) NodeGetInfo(
 
 	var initiators []string
 
-	volDriverTypes := []string{connector.FcDriver, connector.IscsiDriver, connector.NvmeofDriver}
+	volDriverTypes := []string{connector.FcDriver, connector.IscsiDriver, connector.NvmeofDriver, connector.SampleDriver}
 
 	for _, volDriverType := range volDriverTypes {
 		volDriver := connector.NewConnector(volDriverType)
@@ -195,7 +195,7 @@ func (p *Plugin) NodeGetInfo(
 		return nil, status.Error(codes.FailedPrecondition, msg)
 	}
 
-	nodeId := hostName + "," + strings.Join(initiators, ",") + "," + connector.GetHostIP()
+	nodeId := hostName + "," + strings.Join(initiators, ",") + "," + p.Mounter.GetHostIP()
 
 	glog.Infof("node info is %s", nodeId)
 

csi/server/plugin/opensds/plugin.go

@@ -19,6 +19,7 @@ import (
 	"github.com/opensds/nbp/client/opensds"
 	"github.com/opensds/nbp/csi/server/plugin"
 	"github.com/opensds/opensds/client"
+	"github.com/opensds/opensds/contrib/connector"
 )
 
 // Plugin define
@@ -27,6 +28,7 @@ type Plugin struct {
 	Client            *client.Client
 	VolumeClient      *Volume
 	FileShareClient   *FileShare
+	Mounter           connector.Mounter
 }
 
 func NewServer(endpoint, authStrategy, storageType string) (plugin.Service, error) {
@@ -40,8 +42,9 @@ func NewServer(endpoint, authStrategy, storageType string) (plugin.Service, erro
 	p := &Plugin{
 		PluginStorageType: storageType,
 		Client:            client,
-		VolumeClient:      NewVolume(client),
-		FileShareClient:   NewFileshare(client),
+		VolumeClient:      NewVolume(client, connector.GetCommonMounter()),
+		FileShareClient:   NewFileshare(client, connector.GetCommonMounter()),
+		Mounter:           connector.GetCommonMounter(),
 	}
 
 	// When there are multiple volumes unmount at the same time,