from client import StorageConnectorClient, Helper import os, time, socket, threading, sys class fcClient(StorageConnectorClient): """ Implementation of the Fiber Channel Storage Connector client for HANA. It requires multipathing to be set up with Linux dm-multipath and the sg3_utils package to be installed, due to the usage of the sg_persist tool. Note: sufficient rights must be granted without the need for typing the password, this can be achieved by: echo "adm ALL=NOPASSWD: /sbin/multipath, /sbin/multipathd, /etc/init.d/multipathd, /usr/bin/sg_persist, /bin/mount, /bin/umount, /bin/kill, /usr/bin/lsof Note: if using the --storage_cfg parameter during installation (or during addhost with already enabled Storage Connector API, the installer will take care of setting /etc/sudoers correctly) """ apiVersion = 2 # prout-type to fence the devices prType = 6 # exclusive access, registrants only # the persistent reservation key used by all nodes of a system prKey = None interval = 1 retries = 20 def __init__(self, *args, **kwargs): # delegate construction to base class super(fcClient, self).__init__(*args, **kwargs) def about(self): return {"provider_company" : "SAP", "provider_name" : "fcClient", "provider_description" : "Generic Fiber Channel Storage Connector. For any Fiber Channel storages that fulfill the usual Linux kernel interfaces as well as support the SCSI-3 standard.", "provider_version" : "2.0"} @staticmethod def sudoers(): """Gives information about the necessary sudo permission to be set.""" return """ALL=NOPASSWD: /sbin/multipath, /sbin/multipathd, /etc/init.d/multipathd, /usr/bin/sg_persist, /bin/mount, /bin/umount, /bin/kill, /usr/bin/lsof, /usr/bin/systemctl, /usr/sbin/lsof, /usr/sbin/xfs_repair""" def handlePRUnitAttention(self, device): """handles """ retry = 0 while True: if retry > self.retries: break (code, output) = Helper._runOsCommand("sudo /usr/bin/sg_persist -r %s" % device, self.tracer) if code == 0 and "PR generation=" in output: return time.sleep(self.interval) retry = retry+1 msg = "unable to handle UNIT ATTENTION or device not ready after %s seconds" % (self.interval * self.retries) self.tracer.warning(msg) def extractMultipathCfg(self, connectionData): """ Read the OS multipath settings by using the information giving of the HANA configuration (global.ini). There is either a World Wide Identification (wwid) given or an alias given to identify the SAN device. If both are specified the alias is preferred and a wwid in the configuration is overwritten. """ try: wwid = connectionData["wwid"] (code_wwid, output) = Helper._runOsCommand("sudo /sbin/multipath -l %s" % wwid, self.tracer) except: wwid = None code_wwid = -1 try: alias = connectionData["alias"] (code_alias, output) = Helper._runOsCommand("sudo /sbin/multipath -l %s" % alias, self.tracer) except: alias = None code_alias = -1 # at least one of the multipath OS calls must succeed in order to get the correct output if code_wwid != 0 and code_alias != 0: msg = "error reading OS multipath map with wwid:'%s' and alias:'%s'" % (connectionData.get("wwid"), connectionData.get("alias")) print msg self.tracer.info(msg) raise Exception(msg) # override predefined "wwid" if an "alias" is defined on OS level alias_line = "" if not alias is None: lines = output.split("\n") for l in lines: if alias in l: wwid = l.split("(")[1].split(")")[0] alias_line = l break # probe device with wwid first device = "/dev/mapper/%s" % wwid if not os.path.exists(device): # check for an invalid "alias" in the multipath configuration if len(alias_line) == 0: raise Exception("multipath device '%s' not found" % device) else: # probe device with extracted alias device = "/dev/mapper/%s" % alias_line.split("\n")[0].split()[0] # alias if not os.path.exists(device): raise Exception("multipath device '%s' not found (alias:'%s')" % (device, alias)) return (wwid,device,output) # request the storage server to activate a LUN for this host def attach(self, storages): """Attaches storages on this host.""" self.tracer.info("%s.attach method called" % self.__class__.__name__) # a unique key for the host self.prKey = self._generateUniqueHostKey(socket.gethostname()) self.hookAttachBegin(globals(), locals()) threads = [] threadLock = threading.Lock() firstDev = True self._cfg.reload() (self.prType, self.prTypeReservationText) = self.getPRType(defaultPrType = self.prType) # loop over storages given by HDB for storage in storages: self.hookAttachBeginDevice(globals(), locals()) attachMode = "trying to" if storage.get("testOnly") is False else "testing" self.tracer.info("%s attach for partition %s, usage type %s on path %s" % (attachMode, storage.get("partition"), storage.get("usage_type"), storage.get("path"))) # retrieve all parameters for this storage (unique key => (partition, usageType) ) if not self._cfg.hasParameter(storage.get("partition"), storage.get("usage_type"), "wwid") and not self._cfg.hasParameter(storage.get("partition"), storage.get("usage_type"), "alias"): msg = "no storage with key (partiton, usageType) = (%s, %s) configured" % (storage.get("partition"), storage.get("usage_type")) print msg self.tracer.fatal(msg) raise Exception(msg) if firstDev: # some migration stuff when switching from prType 6 to 5: (code, output) = Helper._runOsCommand("sudo /sbin/multipath -r", self.tracer) if code != 0 or "reject" in output.lower(): self._reloadMultipathDaemon() firstDev = False thread = AttachThread(self,threadLock,storage.get("partition"),storage.get("usage_type"),storage.get("path"),storage.get("testOnly")) thread.start() if not self._cfg.hasParameter("*","*","sequentialattach"): threads.append(thread) # save thread for mutual joining else: self.tracer.info("sequential attach is set in global.ini") thread.join() # attach to a usage_type immediately as a fallback for t in threads: t.join() self.flushMultipath(self.prType) self.hookAttachEnd(globals(), locals()) return 0 def detach(self, storages): self.hookDetachBegin(globals(), locals()) self.tracer.info("%s.detach method called" % self.__class__.__name__) # loop over storages given by HDB for storage in storages: self.hookDetachBeginDevice(globals(), locals()) # retrieve all parameters for this storage (unique key => (partition, usageType) ) if not self._cfg.hasParameter(storage.get("partition"), storage.get("usage_type"), "wwid") and not self._cfg.hasParameter(storage.get("partition"), storage.get("usage_type"), "alias"): msg = "no storage with key (partiton, usageType) = (%s, %s) configured" % (storage.get("partition"), storage.get("usage_type")) print msg self.tracer.info(msg) continue connectionData = self._getConnectionDataForLun(storage.get("partition"), storage.get("usage_type")) (wwid, device, output) = self.extractMultipathCfg(connectionData) # extract necessary information path = os.path.realpath(storage.get("path")) ################################################################################################### self.hookDetachPreUmount(globals(), locals()) # check current mounts mountList = self._isMounted(path) # if no device is attached at all, end processing if len(mountList) != 0: # try to unmount device and wait up to 1 minute self._forcedUnmount(device, path, 12) else: msg = "no device attached to path '%s'" % path print msg self.tracer.info(msg) continue self.hookDetachEndDevice(globals(), locals()) ################################################################################################### (self.prType, _) = self.getPRType(defaultPrType = self.prType) self.flushMultipath(self.prType) self.hookDetachEnd(globals(), locals()) return 0 def info(self, paths): self.tracer.info("%s.info method called" % self.__class__.__name__) mappings = [] # retrieve information for each path given by HDB for path in paths: # determine real OS path without symlinks and retrieve the mounted devices path = os.path.realpath(path) mountList = self._isMounted(path) # no device is mounted if len(mountList) == 0: mappings.append({"path" : path, "OS Device Name" : "not_connected"}) continue if len(mountList) > 1: mappings.append({"path" : path, "OS Device Name" : "found_multiple_mounts_on_path"}) continue wwid = iter(mountList).next()[0].split("/")[-1] diskSize = "" (code, output) = Helper._runOsCommand("sudo /sbin/multipath -l %s" % wwid, self.tracer) if code == 0: lines = output.split("\n") for line in lines: if wwid in line: wwid = line.split()[0] singleDevices = self.extractSingleDevices(output, wwid) diskSize = output.split("size=")[1].split()[0] break if len(diskSize) == 0: mappings.append({"path" : path, "OS Device Name" : "wwid_not_found"}) continue # filesystem type (code, output) = Helper._run2PipedOsCommand("cat /proc/mounts", "grep -w %s" % path) if not code == 0: self.tracer.warning("error running cat /proc/mounts: code %s: %s" % (code, output)) fstype = "?" else: fstype = output.split()[2] # combine all extracted information mappings.append({ "path" : path, "OS Multipath Device Name" : iter(mountList).next()[0], "OS Single Devices" : ', '.join(singleDevices), "OS Filesystem Type" : fstype, "WWID" : wwid, "Disk Size" : diskSize, }) return mappings def hookBeginAttach(self, globals, locals): pass def hookAttachBegin(self, globals, locals): pass def hookAttachBeginDevice(self, globals, locals): pass def hookAttachPrePRClear(self, globals, locals): pass def hookAttachPreCleanupUmount(self, globals, locals): pass def hookAttachPrePRRegister(self, globals, locals): pass def hookAttachPreMount(self, globals, locals): pass def hookAttachEndDevice(self, globals, locals): pass def hookAttachEnd(self, globals, locals): pass def hookDetachBegin(self, globals, locals): pass def hookDetachBeginDevice(self, globals, locals): pass def hookDetachPreUmount(self, globals, locals): pass def hookDetachPrePRClear(self, globals, locals): pass def hookDetachEndDevice(self, globals, locals): pass def hookDetachEnd(self, globals, locals): pass class AttachThread(threading.Thread): def __init__(self, caller, attachLock, partition, usageType, path, testOnly): threading.Thread.__init__(self) self.name = usageType + "-" + str(partition) self.caller = caller self.partition = partition self.usageType = usageType self.path = os.path.realpath(path) self.testOnly = testOnly self.attachLock = attachLock self.firstDev = True def join(self): threading.Thread.join(self) if self.exc: msg = "Thread '%s' threw an exception: %s" % (self.getName(), self.exc[1]) new_exc = Exception(msg) raise new_exc.__class__, new_exc, self.exc[2] def run(self): self.exc = None try: self.excRun() except: self.exc = sys.exc_info() def excRun(self): connectionData = self.caller._getConnectionDataForLun(self.partition, self.usageType) (wwid, device, output) = self.caller.extractMultipathCfg(connectionData) ################################################################################################### self.caller.hookAttachPreCleanupUmount(globals(), locals()) # unmount everything if not self.testOnly: mountList = self.caller._isMounted(self.path) mountList.update(self.caller._isMounted(device)) self.caller.unmountEverything(mountList, self.path) ################################################################################################### singleDevices = self.caller.extractSingleDevices(output, wwid) if self.testOnly: return self.caller.hookAttachPrePRClear(globals(), locals()) # register PR key on the device; if it way already registered, this command fails, but the check after that is ok for d in singleDevices: self.caller.handlePRUnitAttention(d) Helper._runOsCommand("sudo /usr/bin/sg_persist --out --register --param-sark=%s %s" % (self.caller.prKey, d), self.caller.tracer) # migration code to handle changed usage of reservation keys (c, o) = Helper._runOsCommand("sudo /usr/bin/sg_persist -i -k %s" % d, self.caller.tracer) if c != 0 or "0x%s" % self.caller.prKey not in o: for l in o.split("\n"): if l.strip().startswith("0x"): oldKey = l.strip()[2:] if oldKey != self.caller.prKey: self.caller.handlePRUnitAttention(d) Helper._runOsCommand("sudo /usr/bin/sg_persist --out --register --param-sark=%s %s" % (oldKey, d), self.caller.tracer) self.caller.handlePRUnitAttention(d) Helper._runOsCommand("sudo /usr/bin/sg_persist --out --clear --param-rk=%s %s" % (oldKey, d), self.caller.tracer) break self.caller.handlePRUnitAttention(d) Helper._runOsCommand("sudo /usr/bin/sg_persist --out --register --param-sark=%s %s" % (self.caller.prKey, d), self.caller.tracer) # check if key was registered for d in singleDevices: (c, o) = Helper._runOsCommand("sudo /usr/bin/sg_persist -i -k %s" % d, self.caller.tracer) if c != 0 or "0x%s" % self.caller.prKey not in o: msg = "reservation key was not registered on device %s" % d print msg self.caller.tracer.fatal(msg) raise Exception(msg) ################################################################################################### self.caller.hookAttachPrePRRegister(globals(), locals()) # the following is only necessary for one device firstDevForReservation = singleDevices[0] self.caller.handlePRUnitAttention(firstDevForReservation) reservations = [] (c, o) = Helper._runOsCommand("sudo /usr/bin/sg_persist -r %s" % firstDevForReservation, self.caller.tracer) for l in o.split("\n"): if l.strip().startswith("Key=0x"): key = l.strip()[6:] if key == self.caller.prKey: self.caller.tracer.info("a reservation for this host is already active, re-write reservation") continue reservations.append(key) self.caller.tracer.info("found another reservation active for key=%s, taking over" % key) if len(reservations) == 0: # reserve with prKey Helper._runOsCommand("sudo /usr/bin/sg_persist --out --reserve --param-rk=%s --prout-type=%s %s" % (self.caller.prKey, self.caller.prType, firstDevForReservation), self.caller.tracer) else: # if there are multiple other keys, there is still only one reservation to take, try to preempt for all key for k in reservations: # preempt with prKey Helper._runOsCommand("sudo /usr/bin/sg_persist --out --preempt --param-sark=%s --param-rk=%s --prout-type=%s %s" % (k, self.caller.prKey, self.caller.prType, firstDevForReservation), self.caller.tracer) for d in singleDevices: self.caller.handlePRUnitAttention(d) # check actual reservation (code, output) = Helper._runOsCommand("sudo /usr/bin/sg_persist -r %s" % firstDevForReservation, self.caller.tracer) if code != 0 or self.caller.prTypeReservationText not in output or "0x%s" % self.caller.prKey not in output: msg = "reservation of Persistent Reservation key failed for device '%s': %s" % (device, output) print msg self.caller.tracer.fatal(msg) raise Exception(msg) ################################################################################################### self.caller.hookAttachPreMount(globals(), locals()) self.caller.attachMount(connectionData, device, self.path) self.caller.hookAttachEndDevice(globals(), locals())