[Driver] Enhancement for concurrency

contrib/drivers/drivers.go

@@ -85,9 +85,14 @@ type VolumeDriver interface {
 	ListPools() ([]*model.StoragePoolSpec, error)
 }
 
+var (
+	OceanStorDriver = oceanstor.Driver{}
+)
+
 // Init
-func Init(resourceType string) VolumeDriver {
+func Init(resourceType string) (VolumeDriver, error) {
 	var d VolumeDriver
+
 	switch resourceType {
 	case config.CinderDriverType:
 		d = &cinder.Driver{}
@@ -102,7 +107,7 @@ func Init(resourceType string) VolumeDriver {
 		d = &spectrumscale.Driver{}
 		break
 	case config.HuaweiOceanStorBlockDriverType:
-		d = &oceanstor.Driver{}
+		d = &OceanStorDriver
 		break
 	case config.HuaweiFusionStorageDriverType:
 		d = &fusionstorage.Driver{}
@@ -119,12 +124,17 @@ func Init(resourceType string) VolumeDriver {
 		d = &sample.Driver{}
 		break
 	}
-	d.Setup()
-	return d
+
+	err := d.Setup()
+	if err != nil {
+		return nil, err
+	}
+
+	return d, nil
 }
 
 // Clean
-func Clean(d VolumeDriver) VolumeDriver {
+func Clean(d VolumeDriver) {
 	// Execute different clean operations according to the VolumeDriver type.
 	switch d.(type) {
 	case *cinder.Driver:
@@ -136,7 +146,7 @@ func Clean(d VolumeDriver) VolumeDriver {
 	case *spectrumscale.Driver:
 		break
 	case *oceanstor.Driver:
-		break
+		return // No need to clean anything for oceanstor
 	case *fusionstorage.Driver:
 		break
 	case *nimble.Driver:
@@ -148,10 +158,9 @@ func Clean(d VolumeDriver) VolumeDriver {
 	default:
 		break
 	}
+
 	d.Unset()
 	d = nil
-
-	return d
 }
 
 func CleanMetricDriver(d MetricDriver) MetricDriver {

contrib/drivers/drivers_test.go

@@ -29,7 +29,7 @@ func TestInit(t *testing.T) {
 	var expectedVd = []VolumeDriver{&sample.Driver{}}
 
 	for i, rs := range rsList {
-		if vp := Init(rs); !reflect.DeepEqual(vp, expectedVd[i]) {
+		if vp, _ := Init(rs); !reflect.DeepEqual(vp, expectedVd[i]) {
 			t.Errorf("Expected %v, got %v\n", expectedVd, vp)
 		}
 	}
@@ -44,8 +44,6 @@ func TestClean(t *testing.T) {
 	}
 
 	for _, driver := range driverList {
-		if d := Clean(driver); !reflect.DeepEqual(d, nil) {
-			t.Errorf("Expected %v, got %v\n", nil, d)
-		}
+		Clean(driver)
 	}
 }

contrib/drivers/huawei/oceanstor/client.go

@@ -82,15 +82,13 @@ func NewClient(opt *AuthOptions) (*OceanStorClient, error) {
 		pwdCiphertext = password
 	}
 
-	c := &OceanStorClient{
+	return &OceanStorClient{
 		user:       opt.Username,
 		passwd:     pwdCiphertext,
 		vstoreName: opt.VstoreName,
 		endpoints:  endpoints,
 		insecure:   opt.Insecure,
-	}
-	err := c.login()
-	return c, err
+	}, nil
 }
 
 func (c *OceanStorClient) Destroy() error {
@@ -137,13 +135,12 @@ func (c *OceanStorClient) request(method, url string, in, out interface{}) error
 	var b []byte
 	var err error
 	for i := 0; i < 2; i++ {
-		// For debugging
-		log.V(5).Infof("URL:%s %s\n BODY:%v", method, c.urlPrefix+url, in)
+		log.Infof("URL:%s %s\n BODY:%v", method, c.urlPrefix+url, in)
 		b, _, err = c.doRequest(method, c.urlPrefix+url, in)
 		if err == nil {
 			break
 		}
-		log.Errorf("URL:%s %s\n BODY:%v", method, c.urlPrefix+url, in)
+
 		if inErr, ok := err.(*ArrayInnerError); ok {
 			errCode := inErr.Err.Code
 			if errCode == ErrorConnectToServer || errCode == ErrorUnauthorizedToServer {
@@ -155,6 +152,7 @@ func (c *OceanStorClient) request(method, url string, in, out interface{}) error
 			}
 			err = inErr
 		}
+
 		log.Errorf("Request %d times error:%v", i+1, err)
 		return err
 	}
@@ -164,6 +162,7 @@ func (c *OceanStorClient) request(method, url string, in, out interface{}) error
 		log.V(8).Infof("Response Body: %s", string(b))
 		json.Unmarshal(b, out)
 	}
+
 	return nil
 }
 
@@ -980,6 +979,10 @@ func (c *OceanStorClient) GetHostLunId(hostId, lunId string) (int, error) {
 func (c *OceanStorClient) RemoveLunFromLunGroup(lunGrpId, lunId string) error {
 	url := fmt.Sprintf("/lungroup/associate?ID=%s&ASSOCIATEOBJTYPE=11&ASSOCIATEOBJID=%s", lunGrpId, lunId)
 	if err := c.request("DELETE", url, nil, nil); err != nil {
+		if c.checkErrorCode(err, ErrorObjectUnavailable) || c.checkErrorCode(err, ErrorLunNotExist) {
+			return nil
+		}
+
 		log.Errorf("Remove lun %s from lun group %s failed, %v", lunId, lunGrpId, err)
 		return err
 	}
@@ -1226,6 +1229,21 @@ func (c *OceanStorClient) GetHostFCInitiators(hostId string) ([]string, error) {
 	return initiators, nil
 }
 
+func (c *OceanStorClient) GetFCInitiatorByWWN(wwn string) (*FCInitiator, error) {
+	resp := &FCInitiatorsResp{}
+	url := fmt.Sprintf("/fc_initiator?filter=ID::%s", wwn)
+	if err := c.request("GET", url, nil, resp); err != nil {
+		log.Errorf("Get fc initiator of WWN %s failed.", wwn)
+		return nil, err
+	}
+
+	if len(resp.Data) > 0 {
+		return &resp.Data[0], nil
+	}
+
+	return nil, nil
+}
+
 func (c *OceanStorClient) GetHostIscsiInitiators(hostId string) ([]string, error) {
 	resp := &InitiatorsResp{}
 	url := fmt.Sprintf("/iscsi_initiator?PARENTTYPE=21&PARENTID=%s", hostId)

contrib/drivers/huawei/oceanstor/oceanstor.go

@@ -35,7 +35,12 @@ type Driver struct {
 	client *OceanStorClient
 }
 
-func (d *Driver) Setup() (err error) {
+func (d *Driver) Setup() error {
+	if d.client != nil {
+		// Login already, return
+		return nil
+	}
+
 	// Read huawei oceanstor config file
 	conf := &OceanStorConfig{}
 	d.conf = conf
@@ -44,17 +49,29 @@ func (d *Driver) Setup() (err error) {
 	if "" == path {
 		path = defaultConfPath
 	}
+
 	Parse(conf, path)
-	d.client, err = NewClient(&d.conf.AuthOptions)
+
+	client, err := NewClient(&d.conf.AuthOptions)
 	if err != nil {
 		log.Errorf("Get new client failed, %v", err)
 		return err
 	}
+
+	err = client.login()
+	if err != nil {
+		log.Errorf("Client login failed, %v", err)
+		return err
+	}
+
+	d.client = client
 	return nil
 }
 
 func (d *Driver) Unset() error {
-	d.client.logout()
+	if d.client != nil {
+		d.client.logout()
+	}
 	return nil
 }
 
@@ -576,69 +593,44 @@ func (d *Driver) InitializeConnectionFC(opt *pb.CreateVolumeAttachmentOpts) (*mo
 }
 
 func (d *Driver) connectFCUseNoSwitch(opt *pb.CreateVolumeAttachmentOpts, initiators []string, hostId string) ([]string, map[string][]string, error) {
-	wwns := initiators
-
-	onlineWWNsInHost, err := d.client.GetHostOnlineFCInitiators(hostId)
-	if err != nil {
-		return nil, nil, err
-	}
-	onlineFreeWWNs, err := d.client.GetOnlineFreeWWNs()
-	if err != nil {
-		return nil, nil, err
-	}
-	onlineFCInitiators, err := d.client.GetOnlineFCInitiatorOnArray()
-	if err != nil {
-		return nil, nil, err
-	}
+	var addWWNs []string
+	var hostWWNs []string
 
-	var wwnsNew []string
-	for _, w := range wwns {
-		if d.isInStringArray(w, onlineFCInitiators) {
-			wwnsNew = append(wwnsNew, w)
+	for _, wwn := range initiators {
+		initiator, err := d.client.GetFCInitiatorByWWN(wwn)
+		if err != nil {
+			log.Errorf("Get FC initiator %s error: %v", wwn, err)
+			return nil, nil, err
+		}
+		if initiator == nil {
+			log.Warningf("FC initiator %s does not exist", wwn)
+			continue
 		}
-	}
-	log.Infof("initialize connection, online initiators on the array:%s", wwnsNew)
-
-	if wwnsNew == nil {
-		return nil, nil, errors.New("no available host initiator")
-	}
-
-	for _, wwn := range wwnsNew {
-		if !d.isInStringArray(wwn, onlineWWNsInHost) && !d.isInStringArray(wwn, onlineFreeWWNs) {
-			wwnsInHost, err := d.client.GetHostFCInitiators(hostId)
-			if err != nil {
-				return nil, nil, err
-			}
-			iqnsInHost, err := d.client.GetHostIscsiInitiators(hostId)
-			if err != nil {
-				return nil, nil, err
-			}
-			flag, err := d.client.IsHostAssociatedToHostgroup(hostId)
-			if err != nil {
-				return nil, nil, err
-			}
 
-			if wwnsInHost == nil && iqnsInHost == nil && flag == false {
-				if err = d.client.RemoveHost(hostId); err != nil {
-					return nil, nil, err
-				}
-			}
+		if initiator.RunningStatus != "27" {
+			log.Warningf("FC initiator %s is not online", wwn)
+			continue
+		}
 
-			msg := fmt.Sprintf("host initiator occupied: Can not add FC initiator %s to host %s, please check if this initiator has been added to other host.", wwn, hostId)
-			log.Errorf(msg)
+		if len(initiator.ParentId) == 0 {
+			addWWNs = append(addWWNs, wwn)
+		} else if initiator.ParentId != hostId {
+			msg := fmt.Sprintf("FC initiator %s is already associated to another host %s", wwn, initiator.ParentId)
+			log.Error(msg)
 			return nil, nil, errors.New(msg)
 		}
+
+		hostWWNs = append(hostWWNs, wwn)
 	}
 
-	for _, wwn := range wwnsNew {
-		if d.isInStringArray(wwn, onlineFreeWWNs) {
-			if err = d.client.AddFCPortTohost(hostId, wwn); err != nil {
-				return nil, nil, err
-			}
+	for _, wwn := range addWWNs {
+		if err := d.client.AddFCPortTohost(hostId, wwn); err != nil {
+			log.Errorf("Add initiator %s to host %s error: %v", wwn, hostId, err)
+			return nil, nil, err
 		}
 	}
 
-	tgtPortWWNs, initTargMap, err := d.client.GetIniTargMap(wwnsNew)
+	tgtPortWWNs, initTargMap, err := d.client.GetIniTargMap(hostWWNs)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -677,6 +669,9 @@ func (d *Driver) detachVolumeFC(opt *pb.DeleteVolumeAttachmentOpts) (string, err
 	if err != nil {
 		return "", err
 	}
+	if len(hostId) == 0 {
+		return "", nil
+	}
 
 	if lunId != "" && lunGrpId != "" {
 		if err := d.client.RemoveLunFromLunGroup(lunGrpId, lunId); err != nil {

pkg/dock/discovery/discovery.go

@@ -170,7 +170,12 @@ func (pdd *provisionDockDiscoverer) Discover() error {
 				pol.Status = availableStatus
 			}
 		} else {
-			d := drivers.Init(dck.DriverName)
+			d, err := drivers.Init(dck.DriverName)
+			if err != nil {
+				log.Errorf("Init driver %s failed: %v", dck.DriverName, err)
+				continue
+			}
+
 			defer drivers.Clean(d)
 			pols, err = d.ListPools()