in progress

Signed-off-by: Ronan Abhamon <ronan.abhamon@vates.tech>

Author: Wescoeur

drivers/LinstorSR.py

@@ -1838,7 +1838,7 @@ def attach(self, sr_uuid, vdi_uuid) -> str:
             return self._attach_using_http_nbd()
 
         # Ensure we have a path...
-        self.sr._vhdutil.create_chain_paths(self.uuid, readonly=not writable)
+        self.sr._vhdutil.create_chain_paths(self.uuid, not writable, cleanup.LinstorSR.abort_from_openers)
 
         self.attached = True
         return VDI.VDI.attach(self, self.sr.uuid, self.uuid)
@@ -2375,7 +2375,7 @@ def _snapshot(self, snap_type, cbtlog=None, cbt_consistency=None):
             raise xs_errors.XenError('SnapshotChainTooLong')
 
         # Ensure we have a valid path if we don't have a local diskful.
-        self.sr._vhdutil.create_chain_paths(self.uuid, readonly=True)
+        self.sr._vhdutil.create_chain_paths(self.uuid, True, cleanup.LinstorSR.abort_from_openers)
 
         volume_path = self.path
         if not util.pathexists(volume_path):

drivers/VDI.py

@@ -123,14 +123,7 @@ def __init__(self, sr, uuid):
 
     @staticmethod
     def from_uuid(session, vdi_uuid):
-
-        _VDI = session.xenapi.VDI
-        vdi_ref = _VDI.get_by_uuid(vdi_uuid)
-        sr_ref = _VDI.get_SR(vdi_ref)
-
-        _SR = session.xenapi.SR
-        sr_uuid = _SR.get_uuid(sr_ref)
-
+        sr_uuid = util.get_sr_from_vdi_uuid(session, vdi)
         sr = SR.SR.from_uuid(session, sr_uuid)
 
         sr.srcmd.params['vdi_ref'] = vdi_ref

drivers/blktap2.py

@@ -424,6 +424,8 @@ def unpause(cls, pid, minor, _type=None, _file=None, mirror=None,
             args += ["-a", str(params)]
         if cbtlog:
             args.extend(["-c", cbtlog])
+
+        # TODO: Handle issue.
         cls._pread(args)
 
     @classmethod
@@ -844,11 +846,16 @@ def launch_on_tap(cls, blktap, path, _type, options):
                             err = (
                                 'status' in e.info and e.info['status']
                             ) or None
-                            if err in (errno.EIO, errno.EROFS, errno.EAGAIN):
+                            if err in (errno.EIO, errno.EROFS, errno.EMEDIUMTYPE, errno.EAGAIN):
                                 if retry_open < 5:
                                     retry_open += 1
                                     time.sleep(1)
                                     continue
+
+                                import cleanup
+                                cleanup.LinstorSR.abort_from_openers()
+
+                                # TODO: create chain
                                 if LINSTOR_AVAILABLE and err == errno.EROFS:
                                     log_drbd_openers(path)
                             raise

drivers/cleanup.py

@@ -3656,9 +3656,9 @@ def _finishInterruptedCoalesceLeaf(self, childUuid, parentUuid):
 
     def _checkSlaves(self, vdi):
         try:
-            all_openers = self._linstor.get_volume_openers(vdi.uuid)
-            for openers in all_openers.values():
-                for opener in openers.values():
+            openers = self._linstor.get_volume_openers(vdi.uuid)
+            for host_openers in openers.values():
+                for opener in host_openers.values():
                     if opener['process-name'] != 'tapdisk':
                         raise util.SMException(
                             'VDI {} is in use: {}'.format(vdi.uuid, all_openers)
@@ -3667,6 +3667,34 @@ def _checkSlaves(self, vdi):
             if e.code != LinstorVolumeManagerError.ERR_VOLUME_NOT_EXISTS:
                 raise
 
+    @staticmethod
+    def abort_from_openers(vdi_uuid, openers) -> None:
+        from linstorvhdutil import MANAGER_PLUGIN
+
+        node_name = None
+
+        for host_openers in openers.values():
+            for hostname, opener in host_openers.values():
+                if opener['process-name'] != 'vhd-util' or 'coalesce' not in opener['cmdline']:
+                    continue
+
+                if not node_name:
+                    import socket
+                    node_name = socket.gethostname()
+
+                if node_name == hostname:
+                    continue
+
+                session = self.xapi.getSession()
+                try:
+                    sr_uuid = util.get_sr_from_vdi_uuid(session, vdi_uuid)
+                    ret = session.xenapi.host.call_plugin(
+                        util.get_master_ref(session), self.MANAGER_PLUGIN, command, args
+                    )
+                finally:
+                    self.session.xenapi.session.logout()
+
+
 
 ################################################################################
 #

drivers/linstorvhdutil.py

@@ -169,7 +169,7 @@ def __init__(self, session, linstor):
         self._session = session
         self._linstor = linstor
 
-    def create_chain_paths(self, vdi_uuid, readonly=False):
+    def create_chain_paths(self, vdi_uuid, readonly=False, cb_openers=None):
         # OPTIMIZE: Add a limit_to_first_allocated_block param to limit vhdutil calls.
         # Useful for the snapshot code algorithm.
 
@@ -192,9 +192,10 @@ def check_volume_usable():
                             time.sleep(2)
                             continue
                         if e.errno == errno.EROFS or e.errno == errno.EMEDIUMTYPE:
-                            util.SMlog('Volume not attachable because used. Openers: {}'.format(
-                                self._linstor.get_volume_openers(vdi_uuid)
-                            ))
+                            openers = self._linstor.get_volume_openers(vdi_uuid)
+                            util.SMlog(f'Volume not attachable because RO. Openers: {openers}')
+                            if cb_openers:
+                                cb_openers(vdi_uuid, openers)
                         raise
                     break
             util.retry(check_volume_usable, 15, 2)
@@ -554,7 +555,7 @@ def _call_method(self, local_method, remote_method, device_path, use_parent, *ar
         # B.3. Call!
         def remote_call():
             try:
-                all_openers = self._linstor.get_volume_openers(openers_uuid)
+                openers = self._linstor.get_volume_openers(openers_uuid)
             except Exception as e:
                 raise xs_errors.XenError(
                     'VDIUnavailable',
@@ -563,8 +564,8 @@ def remote_call():
                 )
 
             no_host_found = True
-            for hostname, openers in all_openers.items():
-                if not openers:
+            for hostname, host_openers in all_openers.items():
+                if not host_openers:
                     continue
 
                 host_ref = self._find_host_ref_from_hostname(hosts, hostname)

drivers/linstorvolumemanager.py

@@ -71,9 +71,16 @@ def get_local_volume_openers(resource_name, volume):
         process_name = groups[0]
         pid = groups[1]
         open_duration_ms = groups[2]
+
+        try:
+            cmdline = util.get_process_cmdline()
+        except Exception:
+            cmdline = []
+
         result[pid] = {
             'process-name': process_name,
-            'open-duration': open_duration_ms
+            'open-duration': open_duration_ms,
+            'cmdline': cmdline
         }
 
     return json.dumps(result)

drivers/tapdisk-pause

@@ -168,7 +168,11 @@ class Tapdisk:
                 group_name,
                 logger=util.SMlog
             )
-            device_path = LinstorVhdUtil(session, linstor).create_chain_paths(self.vdi_uuid)
+
+            import cleanup
+            device_path = LinstorVhdUtil(session, linstor).create_chain_paths(
+                self.vdi_uuid, False, cleanup.LinstorSR.abort_from_openers
+            )
 
             if realpath != device_path:
                 util.SMlog(

drivers/util.py

@@ -16,6 +16,8 @@
 # Miscellaneous utility functions
 #
 
+from sm_typing import List
+
 import os
 import re
 import sys
@@ -1210,6 +1212,12 @@ def _getVDIs(srobj):
     return VDIs
 
 
+def get_sr_from_vdi_uuid(session, vdi_uuid):
+    vdi_ref = session.xenapi.VDI.get_by_uuid(vdi_uuid)
+    sr_ref = session.xenapi.VDI.get_SR(vdi_ref)
+    return session.xenapi.get_uuid(sr_ref)
+
+
 def _getVDI(srobj, vdi_uuid):
     vdi = srobj.session.xenapi.VDI.get_by_uuid(vdi_uuid)
     ref = srobj.session.xenapi.VDI.get_record(vdi)
@@ -1484,6 +1492,17 @@ def pid_is_alive(pid):
         return False
 
 
+def get_process_cmdline(pid: int) -> List[str]:
+    try:
+        with open(os.path.join('/proc', str(pid), 'cmdline'), 'rb') as f:
+            line = f.read().split(b'\0')
+        return [arg.decode() for arg in line]
+    except IOError as e:
+        if e.errno == errno.ENOENT:
+            raise
+        return []
+
+
 # Looks at /proc and figures either
 #   If a process is still running (default), returns open file names
 #   If any running process has open handles to the given file (process = False)