func (s *service) verifyAndUpdateInitiatorsInADiffHost(ctx context.Context, symID string, nodeInitiators []string, hostID string, pmaxClient pmax.Pmax) (int, error) {
	initList, err := pmaxClient.GetInitiatorList(ctx, symID, "", false, false)
	if err != nil {
		log.Warning("Failed to fetch initiator list for the SYM :" + symID)
		return 0, err
	}
	var nValidInitiators int
	for _, nodeInitiator := range nodeInitiators {
		if strings.HasPrefix(nodeInitiator, "0x") {
			nodeInitiator = strings.Replace(nodeInitiator, "0x", "", 1)
		}
		for _, initiatorID := range initList.InitiatorIDs {
			if initiatorID == nodeInitiator || strings.HasSuffix(initiatorID, nodeInitiator) {
				log.Infof("Checking initiator %s against host %s\n", initiatorID, hostID)
				initiator, err := pmaxClient.GetInitiatorByID(ctx, symID, initiatorID)
				if err != nil {
					log.Warning("Failed to fetch initiator details for initiator: " + initiatorID)
					continue
				}
				if initiator.HostID != "" {
					if initiator.HostID != hostID &&
						s.opts.ModifyHostName {
						log.Infof("UpdateHostName processing: %s to %s", initiator.HostID, hostID)
						_, err := pmaxClient.UpdateHostName(ctx, symID, initiator.HostID, hostID)
						if err != nil {
							errormsg := fmt.Sprintf("Failed to change host name from %s to %s: %s", initiator.HostID, hostID, err)
							log.Error(errormsg)
							return 0, fmt.Errorf(errormsg)
						}
					} else if initiator.HostID != hostID {
						errormsg := fmt.Sprintf("initiator: %s is already a part of a different host: %s on: %s",
							initiatorID, initiator.HostID, symID)
						log.Error(errormsg)
						return 0, fmt.Errorf(errormsg)
					}
				}
				log.Infof("valid initiator: %s\n", initiatorID)
				nValidInitiators++
			}
		}
	}
	return nValidInitiators, nil
}
verifyAndUpdateInitiatorsInADiffHost verifies that a set of node initiators are not in a different host than expected.
If they are, it updates the host name for the initiators if ModifyHostName env variable is set.
These can be either FC initiators (hex numbers) or iSCSI initiators (starting with iqn.)
It returns the number of initiators for the host that were found.
Do not mix both FC and iSCSI initiators in a single call.
package service
import (
	"context"
	"fmt"
	"io/ioutil"
	"math/rand"
	"os"
	"path"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"
	pmax "github.com/dell/gopowermax"
	csi "github.com/container-storage-interface/spec/lib/go/csi"
	"github.com/coreos/go-systemd/dbus"
	"github.com/dell/gobrick"
	csictx "github.com/dell/gocsi/context"
	"github.com/dell/gofsutil"
	"github.com/dell/goiscsi"
	log "github.com/sirupsen/logrus"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	types "github.com/dell/gopowermax/types/v90"
)
var (
	maximumStartupDelay         = 30
	getMappedVolMaxRetry        = 20
	nodePublishSleepTime        = 3 * time.Second
	lipSleepTime                = 5 * time.Second
	multipathSleepTime          = 5 * time.Second
	removeDeviceSleepTime       = 5000 * time.Millisecond
	deviceDeletionTimeout       = 5000 * time.Millisecond
	deviceDeletionPoll          = 50 * time.Millisecond
	maxBlockDevicesPerWWN       = 16
	targetMountRecheckSleepTime = 3 * time.Second
	multipathMutex              sync.Mutex
	deviceDeleteMutex           sync.Mutex
	disconnectVolumeRetryTime   = 1 * time.Second
	nodePendingState            pendingState
	sysBlock                    = "/sys/block"  
)
type maskingViewTargetInfo struct {
	target           goiscsi.ISCSITarget
	IsCHAPConfigured bool
}
var symToAllISCSITargets sync.Map
var symToMaskingViewTargets sync.Map
var isSymConnFC = make(map[string]bool)
func (s *service) InvalidateSymToMaskingViewTargets() {
	deletefunc := func(key interface{}, value interface{}) bool {
		symToMaskingViewTargets.Delete(key)
		return true
	}
	symToMaskingViewTargets.Range(deletefunc)
}
func (s *service) NodeStageVolume(
	ctx context.Context,
	req *csi.NodeStageVolumeRequest) (
	*csi.NodeStageVolumeResponse, error) {
	privTgt := req.GetStagingTargetPath()
	if privTgt == "" {
		return nil, status.Error(codes.InvalidArgument, "Target Path is required")
	}
	var reqID string
	headers, ok := metadata.FromIncomingContext(ctx)
	if ok {
		if req, ok := headers["csi.requestid"]; ok && len(req) > 0 {
			reqID = req[0]
		}
	}
	id := req.GetVolumeId()
	_, symID, _, remoteSymID, remoteVolID, err := s.parseCsiID(id)
	if err != nil {
		log.Errorf("Invalid volumeid: %s", id)
		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume id: %s", id)
	}
	pmaxClient, err := s.GetPowerMaxClient(symID, remoteSymID)
	if err != nil {
		log.Error(err.Error())
		return nil, status.Error(codes.InvalidArgument, err.Error())
	}
	var volID volumeIDType = volumeIDType(id)
	if err := volID.checkAndUpdatePendingState(&nodePendingState); err != nil {
		return nil, err
	}
	defer volID.clearPending(&nodePendingState)
	err = s.nodeProbe(ctx)
	if err != nil {
		log.Error("nodeProbe failed with error :" + err.Error())
		return nil, err
	}
	symID, devID, vol, err := s.GetVolumeByID(ctx, id, pmaxClient)
	if err != nil {
		return nil, err
	}
	volumeWWN := vol.WWN
	err = s.writeWWNFile(id, volumeWWN)
	if err != nil {
		log.Error("Could not write WWN file: " + volumeWWN)
	}
	publishContext := req.GetPublishContext()
	volumeLUNAddress := publishContext[PublishContextLUNAddress]
	var keyCount int
	targetIdentifiers := ""
	if count, ok := publishContext[PortIdentifierKeyCount]; ok {
		keyCount, _ = strconv.Atoi(count)
		for i := 1; i <= keyCount; i++ {
			portIdentifierKey := fmt.Sprintf("%s_%d", PortIdentifiers, i)
			targetIdentifiers += publishContext[portIdentifierKey]
		}
	} else {
		targetIdentifiers = publishContext[PortIdentifiers]
	}
	f := log.Fields{
		"CSIRequestID":      reqID,
		"DeviceID":          devID,
		"ID":                req.VolumeId,
		"LUNAddress":        volumeLUNAddress,
		"PrivTgt":           privTgt,
		"SymmetrixID":       symID,
		"TargetIdentifiers": targetIdentifiers,
		"WWN":               volumeWWN,
	}
	log.WithFields(f).Info("NodeStageVolume")
	ctx = setLogFields(ctx, f)
	localPublishContextData := publishContextData{
		deviceWWN:        "0x" + volumeWWN,
		volumeLUNAddress: volumeLUNAddress,
	}
	iscsiTargets, fcTargets, useFC := s.getArrayTargets(ctx, targetIdentifiers, symID, pmaxClient)
	iscsiChroot, _ := csictx.LookupEnv(context.Background(), EnvISCSIChroot)
	if useFC {
		s.initFCConnector(iscsiChroot)
		localPublishContextData.fcTargets = fcTargets
	} else {
		s.initISCSIConnector(iscsiChroot)
		localPublishContextData.iscsiTargets = iscsiTargets
	}
	devicePath, err := s.connectDevice(ctx, localPublishContextData, useFC)
	if err != nil {
		return nil, err
	}
	if remoteSymID != "" {
		remDeviceWWN := publishContext[RemotePublishContextDeviceWWN]
		remVolumeLUNAddress := publishContext[RemotePublishContextLUNAddress]
		remotePublishContextData := publishContextData{
			deviceWWN:        "0x" + remDeviceWWN,
			volumeLUNAddress: remVolumeLUNAddress,
		}
		if remDeviceWWN == "" {
			log.Error("Remote device WWN required to be in PublishContext")
			return nil, status.Error(codes.InvalidArgument, "Remote device WWN required to be in PublishContext")
		}
		f := log.Fields{
			"CSIRequestID":      reqID,
			"DeviceID":          remoteVolID,
			"ID":                req.VolumeId,
			"LUNAddress":        remVolumeLUNAddress,
			"PrivTgt":           privTgt,
			"SymmetrixID":       symID,
			"RemoteSymID":       remoteSymID,
			"TargetIdentifiers": targetIdentifiers,
			"WWN":               remDeviceWWN,
		}
		log.WithFields(f).Info("NodeStageVolume for Remote Device")
		ctx = setLogFields(ctx, f)
		remoteTargetIdentifiers := ""
		if count, ok := publishContext[RemotePortIdentifierKeyCount]; ok {
			keyCount, _ = strconv.Atoi(count)
			for i := 1; i <= keyCount; i++ {
				portIdentifierKey := fmt.Sprintf("%s_%d", RemotePortIdentifiers, i)
				remoteTargetIdentifiers += publishContext[portIdentifierKey]
			}
		} else {
			remoteTargetIdentifiers = publishContext[RemotePortIdentifiers]
		}
		remIscsiTargets, remFcTargets, remUseFC := s.getArrayTargets(ctx, remoteTargetIdentifiers, remoteSymID, pmaxClient)
		log.Infof("Remote ISCSI Targets %v", remIscsiTargets)
		if remUseFC {
			s.initFCConnector(iscsiChroot)
			remotePublishContextData.fcTargets = remFcTargets
		} else {
			s.initISCSIConnector(iscsiChroot)
			remotePublishContextData.iscsiTargets = remIscsiTargets
		}
		remoteDevicePath, err := s.connectDevice(ctx, remotePublishContextData, remUseFC)
		if err != nil {
			return nil, err
		}
		f["RemoteDevicePath"] = remoteDevicePath
		f["RemoteLUNAddress"] = remVolumeLUNAddress
		f["TargetIdentifiers"] = remoteTargetIdentifiers
		f["RemoteWWN"] = remDeviceWWN
	}
	log.WithFields(f).WithField("devPath", devicePath).Info("NodeStageVolume completed")
	return &csi.NodeStageVolumeResponse{}, nil
}
type publishContextData struct {
	deviceWWN        string
	volumeLUNAddress string
	iscsiTargets     []ISCSITargetInfo
	fcTargets        []FCTargetInfo
}
type ISCSITargetInfo struct {
	Portal string
	Target string
}
type FCTargetInfo struct {
	WWPN string
}
func (s *service) connectDevice(ctx context.Context, data publishContextData, useFC bool) (string, error) {
	logFields := getLogFields(ctx)
	var err error
	lun, err := strconv.ParseInt(data.volumeLUNAddress, 16, 0)
	if err != nil {
		log.WithFields(logFields).Errorf("failed to convert lun number to int: %s", err.Error())
		return "", err
	}
	var device gobrick.Device
	if useFC {
		device, err = s.connectFCDevice(ctx, int(lun), data)
	} else {
		device, err = s.connectISCSIDevice(ctx, int(lun), data)
	}
	if err != nil {
		log.Errorf("Unable to find device after multiple discovery attempts: %s", err.Error())
		return "", status.Errorf(codes.Internal,
			"Unable to find device after multiple discovery attempts: %s", err.Error())
	}
	devicePath := path.Join("/dev/", device.Name)
	return devicePath, nil
}
func (s *service) connectISCSIDevice(ctx context.Context,
	lun int, data publishContextData) (gobrick.Device, error) {
	logFields := getLogFields(ctx)
	var targets []gobrick.ISCSITargetInfo
	for _, t := range data.iscsiTargets {
		targets = append(targets, gobrick.ISCSITargetInfo{Target: t.Target, Portal: t.Portal})
	}
	connectorCtx, cFunc := context.WithTimeout(context.Background(), time.Second*120)
	connectorCtx = setLogFields(connectorCtx, logFields)
	defer cFunc()
	connectorCtx = setLogFields(connectorCtx, logFields)
	return s.iscsiConnector.ConnectVolume(connectorCtx, gobrick.ISCSIVolumeInfo{
		Targets: targets,
		Lun:     lun,
	})
}
func (s *service) connectFCDevice(ctx context.Context,
	lun int, data publishContextData) (gobrick.Device, error) {
	logFields := getLogFields(ctx)
	var targets []gobrick.FCTargetInfo
	for _, t := range data.fcTargets {
		targets = append(targets, gobrick.FCTargetInfo{WWPN: t.WWPN})
	}
	connectorCtx, cFunc := context.WithTimeout(context.Background(), time.Second*120)
	connectorCtx = setLogFields(connectorCtx, logFields)
	defer cFunc()
	connectorCtx = setLogFields(connectorCtx, logFields)
	return s.fcConnector.ConnectVolume(connectorCtx, gobrick.FCVolumeInfo{
		Targets: targets,
		Lun:     lun,
	})
}
func (s *service) NodeUnstageVolume(
	ctx context.Context,
	req *csi.NodeUnstageVolumeRequest) (
	*csi.NodeUnstageVolumeResponse, error) {
	var reqID string
	headers, ok := metadata.FromIncomingContext(ctx)
	if ok {
		if req, ok := headers["csi.requestid"]; ok && len(req) > 0 {
			reqID = req[0]
		}
	}
	id := req.GetVolumeId()
	_, symID, _, _, _, err := s.parseCsiID(id)
	if err != nil {
		log.Errorf("Invalid volumeid: %s", id)
		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume id: %s", id)
	}
	pmaxClient, err := s.GetPowerMaxClient(symID)
	if err != nil {
		log.Error(err.Error())
		return nil, status.Error(codes.InvalidArgument, err.Error())
	}
	var volID volumeIDType = volumeIDType(id)
	if err := volID.checkAndUpdatePendingState(&nodePendingState); err != nil {
		return nil, err
	}
	defer volID.clearPending(&nodePendingState)
	stageTgt := req.GetStagingTargetPath()
	if stageTgt == "" {
		return nil, status.Error(codes.InvalidArgument, "A Staging Target argument is required")
	}
	err = gofsutil.Unmount(context.Background(), stageTgt)
	if err != nil {
		log.Infof("NodeUnstageVolume error unmount stage target %s: %s", stageTgt, err.Error())
	}
	removeWithRetry(stageTgt)  
	var volumeWWN string
	volumeWWN, err = s.readWWNFile(id)
	if err != nil {
		log.Infof("Fallback to retrieve WWN from server: %s", err)
		err = s.nodeProbe(ctx)
		if err != nil {
			log.Error("nodeProbe failed with error :" + err.Error())
			return nil, err
		}
		_, _, vol, err := s.GetVolumeByID(ctx, id, pmaxClient)
		if err != nil {
			if strings.Contains(err.Error(), notFound) || strings.Contains(err.Error(), failedToValidateVolumeNameAndID) {
				return &csi.NodeUnstageVolumeResponse{}, nil
			}
			return nil, err
		}
		volumeWWN = vol.WWN
	}
	_, symID, devID, _, _, err := s.parseCsiID(id)
	if err != nil {
		return nil, status.Error(codes.InvalidArgument, err.Error())
	}
	if err := s.disconnectVolume(reqID, symID, devID, volumeWWN); err != nil {
		return nil, err
	}
	privTgt := getPrivateMountPoint(s.privDir, id)
	err = removeWithRetry(privTgt)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	s.removeWWNFile(id)
	return &csi.NodeUnstageVolumeResponse{}, nil
}
func (s *service) disconnectVolume(reqID, symID, devID, volumeWWN string) error {
	for i := 0; i < 3; i++ {
		var deviceName string
		symlinkPath, devicePath, _ := gofsutil.WWNToDevicePathX(context.Background(), volumeWWN)
		if devicePath == "" {
			if i == 0 {
				log.Infof("NodeUnstage Note- Didn't find device path for volume %s", volumeWWN)
			}
			return nil
		}
		devicePathComponents := strings.Split(devicePath, "/")
		deviceName = devicePathComponents[len(devicePathComponents)-1]
		f := log.Fields{
			"CSIRequestID": reqID,
			"DeviceID":     devID,
			"DeviceName":   deviceName,
			"DevicePath":   devicePath,
			"Retry":        i,
			"SymmetrixID":  symID,
			"WWN":          volumeWWN,
		}
		log.WithFields(f).Info("NodeUnstageVolume disconnectVolume")
		nodeUnstageCtx, cancel := context.WithTimeout(context.Background(), time.Second*120)
		nodeUnstageCtx = setLogFields(nodeUnstageCtx, f)
		switch s.arrayTransportProtocolMap[symID] {
		case FcTransportProtocol:
			s.fcConnector.DisconnectVolumeByDeviceName(nodeUnstageCtx, deviceName)  
		case IscsiTransportProtocol:
			s.iscsiConnector.DisconnectVolumeByDeviceName(nodeUnstageCtx, deviceName)  
		}
		cancel()
		time.Sleep(disconnectVolumeRetryTime)
		if _, err := ioutil.ReadDir(sysBlock + deviceName); err != nil {
			os.Remove(symlinkPath)  
		}
	}
	devPath, _ := gofsutil.WWNToDevicePath(context.Background(), volumeWWN)
	if devPath == "" {
		return nil
	}
	return status.Errorf(codes.Internal, "disconnectVolume exceeded retry limit WWN %s devPath %s", volumeWWN, devPath)
}
func (s *service) NodePublishVolume(
	ctx context.Context,
	req *csi.NodePublishVolumeRequest) (
	*csi.NodePublishVolumeResponse, error) {
	var reqID string
	headers, ok := metadata.FromIncomingContext(ctx)
	if ok {
		if req, ok := headers["csi.requestid"]; ok && len(req) > 0 {
			reqID = req[0]
		}
	}
	id := req.GetVolumeId()
	_, symID, _, _, _, err := s.parseCsiID(id)
	if err != nil {
		log.Errorf("Invalid volumeid: %s", id)
		return nil, status.Errorf(codes.InvalidArgument, "Invalid volume id: %s", id)
	}
	pmaxClient, err := s.GetPowerMaxClient(symID)
	if err != nil {
		log.Error(err.Error())
		return nil, status.Error(codes.InvalidArgument, err.Error())
	}
	err = s.nodeProbe(ctx)
	if err != nil {
		log.Error("nodeProbe failed with error :" + err.Error())
		return nil, err
	}
	symID, devID, _, err := s.GetVolumeByID(ctx, id, pmaxClient)
	if err != nil {
		return nil, err
	}
	volumeContext := req.GetVolumeContext()
	if volumeContext != nil {
		log.Infof("VolumeContext:")
		for key, value := range volumeContext {
			log.Infof("    [%s]=%s", key, value)
		}
	}
	publishContext := req.GetPublishContext()
	deviceWWN := publishContext[PublishContextDeviceWWN]
	volumeLUNAddress := publishContext[PublishContextLUNAddress]
	if deviceWWN == "" {
		log.Error("Device WWN required to be in PublishContext")
		return nil, status.Error(codes.InvalidArgument, "Device WWN required to be in PublishContext")
	}
	var keyCount int
	targetIdentifiers := ""
	if count, ok := publishContext[PortIdentifierKeyCount]; ok {
		keyCount, _ = strconv.Atoi(count)
		for i := 1; i <= keyCount; i++ {
			portIdentifierKey := fmt.Sprintf("%s_%d", PortIdentifiers, i)
			targetIdentifiers += publishContext[portIdentifierKey]
		}
	} else {
		targetIdentifiers = publishContext[PortIdentifiers]
	}
	log.WithField("CSIRequestID", reqID).Infof("node publishing volume: %s lun: %s", deviceWWN, volumeLUNAddress)
	var symlinkPath string
	var devicePath string
	symlinkPath, devicePath, err = gofsutil.WWNToDevicePathX(context.Background(), deviceWWN)
	if err != nil || symlinkPath == "" {
		errmsg := fmt.Sprintf("Device path not found for WWN %s: %s", deviceWWN, err)
		log.Error(errmsg)
		return nil, status.Error(codes.NotFound, errmsg)
	}
	f := log.Fields{
		"CSIRequestID":      reqID,
		"DeviceID":          devID,
		"DevicePath":        devicePath,
		"ID":                req.VolumeId,
		"Name":              volumeContext["Name"],
		"WWN":               deviceWWN,
		"PrivateDir":        s.privDir,
		"SymlinkPath":       symlinkPath,
		"SymmetrixID":       symID,
		"TargetPath":        req.GetTargetPath(),
		"TargetIdentifiers": targetIdentifiers,
	}
	log.WithFields(f).Info("Calling publishVolume")
	if err := publishVolume(req, s.privDir, symlinkPath, reqID); err != nil {
		return nil, err
	}
	return &csi.NodePublishVolumeResponse{}, nil
}
func (s *service) NodeUnpublishVolume(
	ctx context.Context,
	req *csi.NodeUnpublishVolumeRequest) (
	*csi.NodeUnpublishVolumeResponse, error) {
	var reqID string
	var err error
	headers, ok := metadata.FromIncomingContext(ctx)
	if ok {
		if req, ok := headers["csi.requestid"]; ok && len(req) > 0 {
			reqID = req[0]
		}
	}
	target := req.GetTargetPath()
	if target == "" {
		log.Error("target path required")
		return nil, status.Error(codes.InvalidArgument,
			"target path required")
	}
	var targetMount gofsutil.Info
	if targetMount, err = s.getTargetMount(target); err != nil {
		return nil, err
	}
	if targetMount.Device == "" {
		log.Debug(fmt.Sprintf("No target mount found... waiting %v to re-verify no target %s mount", targetMountRecheckSleepTime, target))
		time.Sleep(targetMountRecheckSleepTime)
		if targetMount, err = s.getTargetMount(target); err != nil {
			log.Info(fmt.Sprintf("Still no mount entry for target, so assuming this is an idempotent call: %s", target))
			return &csi.NodeUnpublishVolumeResponse{}, nil
		}
	}
	log.Infof("targetMount: %#v", targetMount)
	devicePath := targetMount.Device
	if devicePath == "devtmpfs" || devicePath == "" {
		devicePath = targetMount.Source
	}
	id := req.GetVolumeId()
	f := log.Fields{
		"CSIRequestID": reqID,
		"DevicePath":   devicePath,
		"ID":           id,
		"PrivateDir":   s.privDir,
		"TargetPath":   req.GetTargetPath(),
	}
	log.WithFields(f).Info("Calling Unmount of TargetPath")
	err = gofsutil.Unmount(context.Background(), req.GetTargetPath())
	if err != nil {
		log.WithFields(f).Info("TargetPath unmount: " + err.Error())
	}
	log.WithFields(f).Info("Calling unpublishVolume")
	var lastUnmounted bool
	if lastUnmounted, err = unpublishVolume(req, s.privDir, devicePath, reqID); err != nil {
		return nil, err
	}
	log.Infof("lastUmounted %v\n", lastUnmounted)
	if lastUnmounted {
		removeWithRetry(target)  
		return &csi.NodeUnpublishVolumeResponse{}, nil
	}
	log.WithFields(f).Info("Not the last mount - rechecking target mount is gone")
	if targetMount, err = s.getTargetMount(target); err != nil {
		return nil, err
	}
	if targetMount.Device != "" {
		log.WithFields(f).Error("Target mount still present... returning failure")
		return nil, status.Error(codes.Internal, "Target Mount still present")
	}
	dev, err := GetDevice(devicePath)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	log.WithFields(f).Info("rechecking dev mounts")
	mnts, err := getDevMounts(dev)
	if err != nil {
		return nil, status.Error(codes.Internal, err.Error())
	}
	if len(mnts) > 0 {
		log.WithFields(f).Infof("Device mounts still present: %#v", mnts)
	}
	removeWithRetry(target)  
	return &csi.NodeUnpublishVolumeResponse{}, nil
}
func (s *service) getTargetMount(target string) (gofsutil.Info, error) {
	var targetMount gofsutil.Info
	mounts, err := gofsutil.GetMounts(context.Background())
	if err != nil {
		log.Error("could not reliably determine existing mount status")
		return targetMount, status.Error(codes.Internal, "could not reliably determine existing mount status")
	}
	for _, mount := range mounts {
		if mount.Path == target {
			targetMount = mount
			log.Infof("matching targetMount %s target %s", target, mount.Path)
		}
	}
	return targetMount, nil
}
func (s *service) nodeProbe(ctx context.Context) error {
	log.Debug("Entering nodeProbe")
	defer log.Debug("Exiting nodeProbe")
	if s.opts.NodeName == "" {
		return status.Errorf(codes.FailedPrecondition,
			"Error getting NodeName from the environment")
	}
	err := s.createPowerMaxClients(ctx)
	if err != nil {
		return err
	}
	if s.nodeIsInitialized {
		if s.opts.TransportProtocol != FcTransportProtocol {
			s.nodeProbeMutex.Lock()
			defer s.nodeProbeMutex.Unlock()
			_ = s.ensureLoggedIntoEveryArray(ctx, false)
		}
	}
	return nil
}
func (s *service) NodeGetCapabilities(
	ctx context.Context,
	req *csi.NodeGetCapabilitiesRequest) (
	*csi.NodeGetCapabilitiesResponse, error) {
	return &csi.NodeGetCapabilitiesResponse{
		Capabilities: []*csi.NodeServiceCapability{
			{
				Type: &csi.NodeServiceCapability_Rpc{
					Rpc: &csi.NodeServiceCapability_RPC{
						Type: csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
					},
				},
			},
			{
				Type: &csi.NodeServiceCapability_Rpc{
					Rpc: &csi.NodeServiceCapability_RPC{
						Type: csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
					},
				},
			},
		},
	}, nil
}
func (s *service) getIPInterfaces(ctx context.Context, symID string, portGroups []string, pmaxClient pmax.Pmax) ([]string, error) {
	ipInterfaces := make([]string, 0)
	for _, pg := range portGroups {
		portGroup, err := pmaxClient.GetPortGroupByID(ctx, symID, pg)
		if err != nil {
			return nil, err
		}
		for _, portKey := range portGroup.SymmetrixPortKey {
			port, err := pmaxClient.GetPort(ctx, symID, portKey.DirectorID, portKey.PortID)
			if err != nil {
				return nil, err
			}
			ipInterfaces = append(ipInterfaces, port.SymmetrixPort.IPAddresses...)
		}
	}
	return ipInterfaces, nil
}
func (s *service) isISCSIConnected(err error) bool {
	errMsg := err.Error()
	if strings.Contains(errMsg, "exit status 24") ||  
		strings.Contains(errMsg, "exit status 15") {  
		return true
	}
	return false
}
func (s *service) createTopologyMap(ctx context.Context) (map[string]string, error) {
	topology := map[string]string{}
	iscsiArrays := make([]string, 0)
	arrays, err := s.retryableGetSymmetrixIDList()
	if err != nil {
		return nil, err
	}
	for _, id := range arrays.SymmetrixIDs {
		pmaxClient, err := s.GetPowerMaxClient(id)
		if err != nil {
			log.Error(err.Error())
			continue
		}
		if s.arrayTransportProtocolMap != nil {
			if protocol, ok := s.arrayTransportProtocolMap[id]; ok && protocol == FcTransportProtocol {
				continue
			}
		}
		if s.loggedInArrays != nil {
			if isLoggedIn, ok := s.loggedInArrays[id]; ok && isLoggedIn {
				iscsiArrays = append(iscsiArrays, id)
				continue
			}
		}
		ipInterfaces, err := s.getIPInterfaces(ctx, id, s.opts.PortGroups, pmaxClient)
		if err != nil {
			log.Errorf("unable to fetch ip interfaces for %s: %s", id, err.Error())
			continue
		}
		if len(ipInterfaces) == 0 {
			log.Errorf("Couldn't find any ip interfaces on any of the port-groups")
		}
		for _, ip := range ipInterfaces {
			isArrayConnected := false
			_, err := s.iscsiClient.DiscoverTargets(ip, false)
			if err != nil {
				if s.isISCSIConnected(err) {
					isArrayConnected = true
				} else {
					log.Errorf("Failed to connect to the IP interface(%s) of array(%s)", ip, id)
				}
			} else {
				isArrayConnected = true
			}
			if isArrayConnected {
				iscsiArrays = append(iscsiArrays, id)
				break
			}
		}
	}
	for array, protocol := range s.arrayTransportProtocolMap {
		if protocol == FcTransportProtocol {
			topology[s.getDriverName()+"/"+array] = s.getDriverName()
			topology[s.getDriverName()+"/"+array+"."+strings.ToLower(FcTransportProtocol)] = s.getDriverName()
		}
	}
	for _, array := range iscsiArrays {
		if _, ok := topology[s.getDriverName()+"/"+array]; !ok {
			topology[s.getDriverName()+"/"+array] = s.getDriverName()
			topology[s.getDriverName()+"/"+array+"."+strings.ToLower(IscsiTransportProtocol)] = s.getDriverName()
		}
	}
	return topology, nil
}
func (s *service) NodeGetInfo(
	ctx context.Context,
	req *csi.NodeGetInfoRequest) (
	*csi.NodeGetInfoResponse, error) {
	if s.opts.NodeName == "" {
		log.Error("Unable to get Node Name from the environment")
		return nil, status.Error(codes.FailedPrecondition,
			"Unable to get Node Name from the environment")
	}
	topology, err := s.createTopologyMap(ctx)
	if err != nil {
		log.Errorf("Unable to get the list of symmetrix ids. (%s)\n", err.Error())
		return nil, status.Error(codes.FailedPrecondition,
			"Unable to get the list of symmetrix ids")
	}
	if len(topology) == 0 {
		log.Errorf("No topology keys could be generated")
		return nil, status.Error(codes.FailedPrecondition, "no topology keys could be generate")
	}
	return &csi.NodeGetInfoResponse{
		NodeId: s.opts.NodeName,
		AccessibleTopology: &csi.Topology{
			Segments: topology,
		},
	}, nil
}
func (s *service) NodeGetVolumeStats(
	ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
	return nil, status.Error(codes.Unimplemented, "")
}
func (s *service) nodeStartup(ctx context.Context) error {
	if s.nodeIsInitialized {
		return nil
	}
	nodePendingState.maxPending = 10
	if s.adminClient == nil {
		return fmt.Errorf("There is no Unisphere connection")
	}
	portWWNs := make([]string, 0)
	IQNs := make([]string, 0)
	var err error
	portWWNs, err = gofsutil.GetFCHostPortWWNs(context.Background())
	if err != nil {
		log.Error("nodeStartup could not GetFCHostPortWWNs")
	}
	IQNs, err = s.iscsiClient.GetInitiators("")
	if err != nil {
		log.Error("nodeStartup could not GetInitiatorIQNs")
	}
	log.Infof("TransportProtocol %s FC portWWNs: %s ... IQNs: %s\n", s.opts.TransportProtocol, portWWNs, IQNs)
	if len(portWWNs) == 0 && len(IQNs) == 0 {
		return fmt.Errorf("No FC or iSCSI initiators were found and at least 1 is required")
	}
	arrays, err := s.retryableGetSymmetrixIDList()
	if err != nil {
		log.Error("Failed to fetch array list. Continuing without initializing node")
		return err
	}
	symmetrixIDs := arrays.SymmetrixIDs
	log.Debug(fmt.Sprintf("GetSymmetrixIDList returned: %v", symmetrixIDs))
	s.nodeHostSetup(ctx, portWWNs, IQNs, symmetrixIDs)  
	return err
}
func isValidHostID(hostID string) bool {
	rx := regexp.MustCompile("^[a-zA-Z0-9]{1}[a-zA-Z0-9_\\-]*$")
	return rx.MatchString(hostID)
}