Fix/sas initiator (#50)

feat: SAS attached volume support

Add support for mapping volumes through SAS HBA.
Automatic SAS address discovery or manual specification through
optional config file.

Makefile

@@ -9,7 +9,7 @@ endif
 ifdef VERSION
 VERSION := $(VERSION)
 else
-VERSION := v1.5.0
+VERSION := v1.5.1
 endif
 
 VERSION_FLAG = -X github.com/Seagate/seagate-exos-x-csi/pkg/common.Version=$(VERSION)

go.mod

@@ -1,29 +1,45 @@
 module github.com/Seagate/seagate-exos-x-csi
 
-go 1.16
+go 1.19
 
 require (
 	github.com/Seagate/csi-lib-iscsi v1.0.3
 	github.com/Seagate/csi-lib-sas v1.0.1
-	github.com/Seagate/seagate-exos-x-api-go v1.0.8-0.20220531203625-3d1a38b18ac6
-	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/Seagate/seagate-exos-x-api-go v1.0.9
 	github.com/container-storage-interface/spec v1.4.0
 	github.com/golang/protobuf v1.5.2
-	github.com/google/uuid v1.1.2
+	github.com/google/uuid v1.3.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.2
 	github.com/kubernetes-csi/csi-test v0.0.0-20191016154743-6931aedb3df0
 	github.com/onsi/gomega v1.10.5
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.11.0
-	github.com/prometheus/common v0.32.1 // indirect
-	github.com/prometheus/procfs v0.7.3 // indirect
-	golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
-	golang.org/x/sys v0.0.0-20211124211545-fe61309f8881 // indirect
-	google.golang.org/genproto v0.0.0-20211118181313-81c1377c94b1 // indirect
-	google.golang.org/grpc v1.42.0
+	github.com/prometheus/client_golang v1.13.0
+	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f
+	google.golang.org/grpc v1.50.0
 	k8s.io/klog v1.0.0
 )
 
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/fsnotify/fsnotify v1.4.7 // indirect
+	github.com/go-logr/logr v1.2.0 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.2 // indirect
+	github.com/nxadm/tail v1.4.4 // indirect
+	github.com/onsi/ginkgo v1.12.1 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.37.0 // indirect
+	github.com/prometheus/procfs v0.8.0 // indirect
+	golang.org/x/net v0.0.0-20220909164309-bea034e7d591 // indirect
+	golang.org/x/sys v0.0.0-20221010170243-090e33056c14 // indirect
+	golang.org/x/text v0.3.7 // indirect
+	google.golang.org/genproto v0.0.0-20221010155953-15ba04fc1c0e // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
+	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+	k8s.io/klog/v2 v2.70.0 // indirect
+)
+
 // replace github.com/Seagate/seagate-exos-x-api-go => ../seagate-exos-x-api-go
 // replace github.com/Seagate/csi-lib-iscsi => ../csi-lib-iscsi
 // replace github.com/Seagate/csi-lib-sas => ../csi-lib-sas

helm/csi-charts/values.yaml

@@ -17,14 +17,15 @@ image:
   pullPolicy: Always
 
 # -- Controller sidecar for provisioning
+# AKA external-provisioner
 csiProvisioner:
   image:
     repository: k8s.gcr.io/sig-storage/csi-provisioner
     tag: v3.0.0
   # -- Timeout for gRPC calls from the csi-provisioner to the controller
   timeout: 30s
   # -- Extra arguments for csi-provisioner controller sidecar
-  extraArgs: []
+  extraArgs: [--feature-gates=Topology=true]
 
 # -- Controller sidecar for attachment handling
 csiAttacher:

pkg/common/driver.go

@@ -38,6 +38,8 @@ const (
 	StorageProtocolISCSI      = "iscsi"
 	StorageProtocolFC         = "fc"
 	StorageProtocolSAS        = "sas"
+	TopologyInitiatorPrefix   = "com.seagate-exos-x-csi"
+	TopologySASInitiatorLabel = "sas-address"
 
 	MaximumLUN            = 255
 	VolumeNameMaxLength   = 32

pkg/common/identity.go

@@ -32,6 +32,13 @@ func (driver *Driver) GetPluginCapabilities(ctx context.Context, req *csi.GetPlu
 					},
 				},
 			},
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
+					},
+				},
+			},
 		},
 	}, nil
 }

pkg/controller/controller.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"sync"
 
 	storageapi "github.com/Seagate/seagate-exos-x-api-go"
@@ -59,7 +60,8 @@ var nonAuthenticatedMethods = []string{
 type Controller struct {
 	*common.Driver
 
-	client *storageapi.Client
+	client  *storageapi.Client
+	runPath string
 }
 
 // DriverCtx contains data common to most calls
@@ -73,8 +75,13 @@ type DriverCtx struct {
 func New() *Controller {
 	client := storageapi.NewClient()
 	controller := &Controller{
-		Driver: common.NewDriver(client.Collector),
-		client: client,
+		Driver:  common.NewDriver(client.Collector),
+		client:  client,
+		runPath: fmt.Sprintf("/var/run/%s", common.PluginName),
+	}
+
+	if err := os.MkdirAll(controller.runPath, 0755); err != nil {
+		panic(err)
 	}
 
 	controller.InitServer(

pkg/controller/provisioner.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	"github.com/Seagate/seagate-exos-x-csi/pkg/common"
 	"github.com/Seagate/seagate-exos-x-csi/pkg/storage"
@@ -23,10 +24,28 @@ var (
 	}
 )
 
+// Extract available SAS addresses for the Node from topology segments
+func parseTopology(topology []*csi.Topology, parameters *map[string]string) error {
+	klog.Infof("Topology: %v", topology)
+
+	sasAddressSearchString := fmt.Sprintf("%s/%s", common.TopologyInitiatorPrefix, common.TopologySASInitiatorLabel)
+	for _, topo := range topology {
+		for key, val := range topo.GetSegments() {
+			if strings.Contains(key, sasAddressSearchString) {
+				(*parameters)[key] = val
+			}
+		}
+	}
+	return nil
+}
+
 // CreateVolume creates a new volume from the given request. The function is idempotent.
 func (controller *Controller) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
 
 	parameters := req.GetParameters()
+	//insert topology keys into the parameters map, so they will be available in ControllerPublishVolume
+	parseTopology(req.GetAccessibilityRequirements().GetRequisite(), &parameters)
+
 	volumeName, err := common.TranslateName(req.GetName(), parameters[common.VolumePrefixKey])
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, "translate volume name contains invalid characters")
@@ -35,7 +54,7 @@ func (controller *Controller) CreateVolume(ctx context.Context, req *csi.CreateV
 	// Extract the storage interface protocol to be used for this volume (iscsi, fc, sas, etc)
 	storageProtocol := storage.ValidateStorageProtocol(parameters[common.StorageProtocolKey])
 
-	if common.ValidateName(volumeName) == false {
+	if !common.ValidateName(volumeName) {
 		return nil, status.Error(codes.InvalidArgument, "volume name contains invalid characters")
 	}
 

pkg/controller/publisher.go

@@ -2,6 +2,11 @@ package controller
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"strings"
 
 	"github.com/Seagate/seagate-exos-x-csi/pkg/common"
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -10,6 +15,51 @@ import (
 	"k8s.io/klog"
 )
 
+// getConnectorInfoPath
+func (driver *Controller) getConnectorInfoPath(volumeID string) string {
+	return fmt.Sprintf("%s/%s.json", driver.runPath, volumeID)
+}
+
+// Read connector json file and return initiator address info for the given volume
+func (driver *Controller) readInitiatorMapFromFile(filePath string, volumeID string) ([]string, error) {
+	klog.Infof("Reading initiator value for volume %v from file %v", volumeID, filePath)
+	f, err := ioutil.ReadFile(filePath)
+	if err != nil {
+		return nil, err
+	}
+	initiatorMap := make(map[string][]string)
+	err = json.Unmarshal(f, &initiatorMap)
+	if err != nil {
+		return nil, fmt.Errorf("error unmarshaling initiator info file for specified volume ID %v", volumeID)
+	}
+	initiators, found := initiatorMap[volumeID]
+	if found {
+		return initiators, nil
+	} else {
+		return nil, fmt.Errorf("initiator value for volume ID %v not found", volumeID)
+	}
+}
+
+// PersistConnector persists the provided Connector to the specified file
+func persistInitiatorMap(volumeID string, initiators []string, filePath string) error {
+	initiatorMap := map[string][]string{
+		volumeID: initiators,
+	}
+	f, err := os.Create(filePath)
+	if err != nil {
+		klog.Error("error encoding initiator info: %v", err)
+		return fmt.Errorf("error creating initiator map file %s: %s", filePath, err)
+	}
+	defer f.Close()
+	encoder := json.NewEncoder(f)
+	if err = encoder.Encode(initiatorMap); err != nil {
+		klog.Error("error encoding initiator info: %v", err)
+		return fmt.Errorf("error encoding initiator info: %v", err)
+	}
+	klog.Infof("wrote initiator persistence file at %s", filePath)
+	return nil
+}
+
 // ControllerPublishVolume attaches the given volume to the node
 func (driver *Controller) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
 	if len(req.GetVolumeId()) == 0 {
@@ -21,12 +71,28 @@ func (driver *Controller) ControllerPublishVolume(ctx context.Context, req *csi.
 	if req.GetVolumeCapability() == nil {
 		return nil, status.Error(codes.InvalidArgument, "cannot publish volume without capabilities")
 	}
+	parameters := req.GetVolumeContext()
 
 	volumeName, _ := common.VolumeIdGetName(req.GetVolumeId())
-	initiatorName := req.GetNodeId()
-	klog.Infof("attach request for initiator %s, volume id: %s", initiatorName, volumeName)
 
-	lun, err := driver.client.PublishVolume(volumeName, initiatorName)
+	var initiatorNames []string
+	// Available SAS initiators for the node are provided here through NodeGetInfo
+	if parameters[common.StorageProtocolKey] == common.StorageProtocolSAS {
+		for key, val := range parameters {
+			if strings.Contains(key, common.TopologyInitiatorPrefix) {
+				initiatorNames = append(initiatorNames, val)
+			}
+		}
+	} else {
+		initiatorNames = []string{req.GetNodeId()}
+	}
+
+	persistentInfoFilepath := driver.getConnectorInfoPath(req.GetVolumeId())
+	persistInitiatorMap(volumeName, initiatorNames, persistentInfoFilepath)
+
+	klog.Infof("attach request for initiator(s) %v, volume id: %s", initiatorNames, volumeName)
+
+	lun, err := driver.client.PublishVolume(volumeName, initiatorNames)
 
 	if err != nil {
 		return nil, err
@@ -44,17 +110,33 @@ func (driver *Controller) ControllerUnpublishVolume(ctx context.Context, req *cs
 	}
 
 	volumeName, _ := common.VolumeIdGetName(req.GetVolumeId())
-	klog.Infof("unmapping volume %s from initiator %s", volumeName, req.GetNodeId())
-	_, status, err := driver.client.UnmapVolume(volumeName, req.GetNodeId())
-	if err != nil {
-		if status != nil && status.ReturnCode == unmapFailedErrorCode {
-			klog.Info("unmap failed, assuming volume is already unmapped")
-			return &csi.ControllerUnpublishVolumeResponse{}, nil
+
+	var initiators []string
+	var err error
+	if protocol, _ := common.VolumeIdGetStorageProtocol(req.GetVolumeId()); protocol == common.StorageProtocolSAS {
+		initiators, err = driver.readInitiatorMapFromFile(driver.getConnectorInfoPath(req.GetVolumeId()), volumeName)
+		if err != nil {
+			return nil, fmt.Errorf("error retrieving initiator! cannot unpublish volume %v", volumeName)
 		}
+	} else {
+		initiators = []string{req.GetNodeId()}
+	}
 
-		return nil, err
+	klog.Infof("unmapping volume %s from initiator %s", volumeName, initiators)
+	for _, initiator := range initiators {
+		_, status, err := driver.client.UnmapVolume(volumeName, initiator)
+		if err != nil {
+			if status != nil && status.ReturnCode == unmapFailedErrorCode {
+				klog.Info("unmap failed, assuming volume is already unmapped")
+			} else {
+				klog.Errorf("unknown error while unmapping initiator %s: %v", initiator, err)
+			}
+		}
 	}
 
+	persistentInfoFilepath := driver.getConnectorInfoPath(req.GetVolumeId())
+	os.Remove(persistentInfoFilepath)
+
 	klog.Infof("successfully unmapped volume %s from all initiators", volumeName)
 	return &csi.ControllerUnpublishVolumeResponse{}, nil
 }