added support for caching-node connect

pkg/spdk/controllerserver.go

@@ -66,7 +66,7 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	volumeInfo, err := cs.publishVolume(csiVolume.GetVolumeId())
+	volumeInfo, err := cs.publishVolume(req, csiVolume.GetVolumeId())
 	if err != nil {
 		klog.Errorf("failed to publish volume, volumeID: %s err: %v", volumeID, err)
 		cs.deleteVolume(csiVolume.GetVolumeId()) //nolint:errcheck // we can do little
@@ -333,7 +333,7 @@ func getSPDKVol(csiVolumeID string) (*spdkVolume, error) {
 	return nil, fmt.Errorf("missing poolName in volume: %s", csiVolumeID)
 }
 
-func (cs *controllerServer) publishVolume(volumeID string) (map[string]string, error) {
+func (cs *controllerServer) publishVolume(req *csi.CreateVolumeRequest, volumeID string) (map[string]string, error) {
 	spdkVol, err := getSPDKVol(volumeID)
 	if err != nil {
 		return nil, err
@@ -342,6 +342,20 @@ func (cs *controllerServer) publishVolume(volumeID string) (map[string]string, e
 	if err != nil {
 		return nil, err
 	}
+	if _, ok := req.GetParameters()["type"]; ok {
+		hostId, err := cs.spdkNode.GetVolumeHostID(spdkVol.lvolID)
+		if err != nil {
+			return nil, err
+		}
+		err = cs.spdkNode.CachingNodeConnect(hostId, spdkVol.lvolID)
+		if err != nil {
+			klog.Errorf("error Connecting volume to host: %v", err)
+			return nil, err
+		}
+		return map[string]string{
+			"hostId": hostId,
+		}, nil
+	}
 
 	volumeInfo, err := cs.spdkNode.VolumeInfo(spdkVol.lvolID)
 	if err != nil {

pkg/util/jsonrpc.go

@@ -119,15 +119,10 @@ type connectionInfo struct {
 
 // BDev SPDK block device
 type BDev struct {
-	Name           string `json:"name"`
-	UUID           string `json:"uuid"`
-	BlockSize      int64  `json:"block_size"`
-	NumBlocks      int64  `json:"num_blocks"`
-	DriverSpecific *struct {
-		Lvol struct {
-			LvolStoreUUID string `json:"lvol_store_uuid"`
-		} `json:"lvol"`
-	} `json:"driver_specific,omitempty"`
+	Name     string `json:"lvol_name"`
+	UUID     string `json:"uuid"`
+	LvolSize int64  `json:"size"`
+	HostID   string `json:"host_id"`
 }
 
 type RPCClient struct {
@@ -225,10 +220,13 @@ func (client *RPCClient) getVolume(lvolID string) (*BDev, error) {
 		}
 		return nil, err
 	}
-
-	result, ok := out.([]BDev)
-	if !ok {
-		return nil, fmt.Errorf("failed to convert the response to []BDev type. Interface: %v", out)
+	b, err := json.Marshal(out)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal the response: %w", err)
+	}
+	err = json.Unmarshal(b, &result)
+	if err != nil {
+		return nil, err
 	}
 	return &result[0], err
 }
@@ -300,6 +298,26 @@ func (client *RPCClient) deleteVolume(lvolID string) error {
 	return err
 }
 
+type CachingNodeReq struct {
+	LvolID string `json:"lvol_id"`
+}
+
+func (client *RPCClient) cachingNodeConnect(hostID, lvolID string) (bool, error) {
+	params := CachingNodeReq{
+		LvolID: lvolID,
+	}
+	var result bool
+	out, err := client.CallSBCLI("PUT", "/cachingnode/connect/"+hostID, &params)
+	if err != nil {
+		return false, err
+	}
+	result, ok := out.(bool)
+	if !ok {
+		return false, fmt.Errorf("failed to convert the response to bool type. Interface: %v", out)
+	}
+	return result, nil
+}
+
 func (client *RPCClient) resizeVolume(lvolID string, newSize int64) (bool, error) {
 	params := ResizeVolReq{
 		LvolID:  lvolID,

pkg/util/nvmf.go

@@ -105,6 +105,16 @@ func (node *NodeNVMf) GetVolume(lvolName, poolName string) (string, error) {
 	return lvol.UUID, err
 }
 
+func (node *NodeNVMf) GetVolumeHostID(lvolID string) (string, error) {
+	lvol, err := node.client.getVolume(lvolID)
+	if err != nil {
+		return "", err
+	}
+
+	hostId := lvol.HostID
+	return hostId, err
+}
+
 // ListVolumes returns a list of volumes
 func (node *NodeNVMf) ListVolumes() ([]*BDev, error) {
 	return node.client.listVolumes()
@@ -150,6 +160,15 @@ func (node *NodeNVMf) DeleteSnapshot(snapshotID string) error {
 	return nil
 }
 
+func (node *NodeNVMf) CachingNodeConnect(hostID, lvolID string) error {
+	_, err := node.client.cachingNodeConnect(hostID, lvolID)
+	if err != nil {
+		return err
+	}
+	klog.V(5).Infof("caching node connected: %s", hostID)
+	return nil
+}
+
 // PublishVolume exports a volume through NVMf target
 func (node *NodeNVMf) PublishVolume(lvolID string) error {
 	_, err := node.client.CallSBCLI("GET", "/lvol/"+lvolID, nil)