Coalesce on remote WIP

Signed-off-by: Damien Thenot <[email protected]>

Author: Nambrok

drivers/blktap2.py

@@ -64,6 +64,7 @@
     LINSTOR_AVAILABLE = False
 
 PLUGIN_TAP_PAUSE = "tapdisk-pause"
+PLUGIN_ON_SLAVE = "on-slave"
 
 SOCKPATH = "/var/xapi/xcp-rrdd"
 
@@ -1658,6 +1659,8 @@ def _activate_locked(self, sr_uuid, vdi_uuid, options):
         if self.tap_wanted():
             if not self._add_tag(vdi_uuid, not options["rdonly"]):
                 return False
+                #TODO: Need to interrupt coalesce on master, the coalesce will check for host_OpaqueRef on the VDI before trying offline coalesce
+                #TODO: The coalesce could happen on another slave in onlinecoalesce, interrupt coalesce on another slave (online coalesce)?
             refresh = True
 
         try:

drivers/cleanup.py

@@ -757,7 +757,7 @@ def getTreeHeight(self):
 
         return maxChildHeight + 1
 
-    def getAllLeaves(self):
+    def getAllLeaves(self) -> List["VDI"]:
         "Get all leaf nodes in the subtree rooted at self"
         if len(self.children) == 0:
             return [self]
@@ -828,6 +828,49 @@ def _clear(self):
     def _clearRef(self):
         self._vdiRef = None
 
+    def _call_plugin_coalesce(self, hostRef):
+        args = {"path": self.path, "vdi_type": self.vdi_type}
+        self.sr.xapi.session.xenapi.host.call_plugin( \
+                    hostRef, XAPI.PLUGIN_ON_SLAVE, "commit_tapdisk", args)
+
+    def _doCoalesceOnHost(self, hostRef):
+        self.validate()
+        self.parent.validate(True)
+        self.parent._increaseSizeVirt(self.sizeVirt)
+        self.sr._updateSlavesOnResize(self.parent)
+        #TODO: We might need to make the LV RW on the slave directly for coalesce?
+
+        def abortTest():
+            file = self.sr._gc_running_file(self)
+            try:
+                with open(file, "r") as f:
+                    if not f.read():
+                        return True
+            except OSError as e:
+                if e.errno == errno.ENOENT:
+                    util.SMlog("File {} does not exist".format(file))
+                else:
+                    util.SMlog("IOError: {}".format(e))
+                return True
+            return False
+
+        Util.runAbortable(lambda: self._call_plugin_coalesce(hostRef),
+                          None, self.sr.uuid, abortTest, VDI.POLL_INTERVAL, 0)
+
+        self.parent.validate(True)
+        #self._verifyContents(0)
+        self.parent.updateBlockInfo()
+
+    def _isOpenOnHosts(self) -> Optional[str]:
+        for pbdRecord in self.sr.xapi.getAttachedPBDs():
+            hostRef = pbdRecord["host"]
+            args = {"path": self.path}
+            is_openers = util.strtobool(self.sr.xapi.session.xenapi.host.call_plugin( \
+                    hostRef, XAPI.PLUGIN_ON_SLAVE, "is_openers", args))
+            if is_openers:
+                return hostRef
+        return None
+
     def _doCoalesce(self) -> None:
         """Coalesce self onto parent. Only perform the actual coalescing of
         an image, but not the subsequent relinking. We'll do that as the next step,
@@ -914,7 +957,7 @@ def coalesce(self) -> int:
         return self.cowutil.coalesce(self.path)
 
     @staticmethod
-    def _doCoalesceCowImage(vdi):
+    def _doCoalesceCowImage(vdi: "VDI"):
         try:
             startTime = time.time()
             allocated_size = vdi.getAllocatedSize()
@@ -2383,7 +2426,17 @@ def cleanupJournals(self, dryRun=False):
     def cleanupCache(self, maxAge=-1) -> int:
         return 0
 
-    def _coalesce(self, vdi):
+    def _hasLeavesAttachedOn(self, vdi: VDI):
+        leaves = vdi.getAllLeaves()
+        leaves_vdi = [leaf.uuid for leaf in leaves]
+        return util.get_hosts_attached_on(self.xapi.session, leaves_vdi)
+
+    def _gc_running_file(self, vdi):
+        run_file = "gc_running_{}".format(vdi.uuid)
+        return os.path.join(NON_PERSISTENT_DIR, str(self.uuid), run_file)
+
+    def _coalesce(self, vdi: VDI):
+        skipRelink = False
         if self.journaler.get(vdi.JRN_RELINK, vdi.uuid):
             # this means we had done the actual coalescing already and just
             # need to finish relinking and/or refreshing the children
@@ -2393,8 +2446,35 @@ def _coalesce(self, vdi):
             # order to decide whether to abort the coalesce. We remove the
             # journal as soon as the COW coalesce step is done, because we
             # don't expect the rest of the process to take long
+
+            #TODO: Create `gc_running` in `/run/nonpersistent/sm/<sr uuid>/`
+            if os.path.exists(self._gc_running_file(vdi)):
+                util.SMlog("gc_running already exist for {}. Ignoring...".format(self.uuid))
+
+            with open(self._gc_running_file(vdi), "w") as f:
+                f.write("1")
+
             self.journaler.create(vdi.JRN_COALESCE, vdi.uuid, "1")
-            vdi._doCoalesce()
+            host_refs = self._hasLeavesAttachedOn(vdi)
+            #TODO: this check of multiple host_refs should be done earlier in `is_coalesceable` to avoid stopping this late every time
+            if len(host_refs) > 1:
+                util.SMlog("Not coalesceable, chain activated more than once")
+                raise Exception("Not coalesceable, chain activated more than once") #TODO: Use correct error
+
+            try:
+                if host_refs and vdi.cowutil.coalesceOnRemote:
+                    #Leaf opened on another host, we need to call online coalesce
+                    vdi._doCoalesceOnHost(host_refs[0])
+                    skipRelink = True
+                else:
+                    vdi._doCoalesce()
+            except:
+                os.unlink(self._gc_running_file(vdi))
+                raise
+            """
+            vdi._doCoalesce will call vdi._coalesceCowImage (after doing other things).
+            It will then call VDI._doCoalesceCowImage in a runAbortable context
+            """
             self.journaler.remove(vdi.JRN_COALESCE, vdi.uuid)
 
             util.fistpoint.activate("LVHDRT_before_create_relink_journal", self.uuid)
@@ -2403,19 +2483,22 @@ def _coalesce(self, vdi):
             # like SM.clone from manipulating the VDIs we'll be relinking and
             # rescan the SR first in case the children changed since the last
             # scan
-            self.journaler.create(vdi.JRN_RELINK, vdi.uuid, "1")
+            if not skipRelink:
+                self.journaler.create(vdi.JRN_RELINK, vdi.uuid, "1")
 
-        self.lock()
-        try:
-            vdi.parent._tagChildrenForRelink()
-            self.scan()
-            vdi._relinkSkip()
-        finally:
-            self.unlock()
-            # Reload the children to leave things consistent
-            vdi.parent._reloadChildren(vdi)
+        if not skipRelink:
+            self.lock()
+            try:
+                vdi.parent._tagChildrenForRelink()
+                self.scan()
+                vdi._relinkSkip()
+            finally:
+                self.unlock()
+                # Reload the children to leave things consistent
+                vdi.parent._reloadChildren(vdi)
+            self.journaler.remove(vdi.JRN_RELINK, vdi.uuid)
 
-        self.journaler.remove(vdi.JRN_RELINK, vdi.uuid)
+        os.unlink(self._gc_running_file(vdi))
         self.deleteVDI(vdi)
 
     class CoalesceTracker:
@@ -3718,8 +3801,7 @@ def _gc_init_file(sr_uuid):
 
 def _create_init_file(sr_uuid):
     util.makedirs(os.path.join(NON_PERSISTENT_DIR, str(sr_uuid)))
-    with open(os.path.join(
-            NON_PERSISTENT_DIR, str(sr_uuid), 'gc_init'), 'w+') as f:
+    with open(os.path.join(_gc_init_file(sr_uuid)), 'w+') as f:
         f.write('1')
 
 
@@ -3748,7 +3830,7 @@ def abortTest():
         Util.log("GC active, quiet period ended")
 
 
-def _gcLoop(sr, dryRun=False, immediate=False):
+def _gcLoop(sr: SR, dryRun=False, immediate=False):
     if not lockGCActive.acquireNoblock():
         Util.log("Another GC instance already active, exiting")
         return

drivers/cowutil.py

@@ -266,6 +266,14 @@ def getKeyHash(self, path: str) -> Optional[str]:
     def setKey(self, path: str, key_hash: str) -> None:
         pass
 
+    @abstractmethod
+    def coalesceOnRemote(self) -> bool:
+        pass
+
+    @abstractmethod
+    def coalesceOnline(self, path: str) -> int:
+        pass
+
     def getParentChain(self, lvName: str, extractUuidFunction: Callable[[str], str], vgName: str) -> Dict[str, str]:
         """
         Get the chain of all parents of 'path'. Safe to call for raw VDI's as well.

drivers/on_slave.py

@@ -148,7 +148,6 @@ def is_open(session, args):
         util.logException("is_open")
         raise
 
-
 def refresh_lun_size_by_SCSIid(session, args):
     """Refresh the size of LUNs backing the SCSIid on the local node."""
     util.SMlog("on-slave.refresh_lun_size_by_SCSIid(,%s)" % args)
@@ -160,10 +159,35 @@ def refresh_lun_size_by_SCSIid(session, args):
         util.SMlog("on-slave.refresh_lun_size_by_SCSIid with %s failed" % args)
         return "False"
 
+def commit_tapdisk(session, args):
+    path = args["path"]
+    vdi_type = args["vdi_type"]
+    #TODO: Miss activating/changing readonly, naming should reflect that it does more than coalesceing
+    from cowutil import getCowUtil
+    cowutil = getCowUtil(vdi_type)
+    return str(cowutil.coalesceOnline(path))
+
+def commit_cancel(session, args):
+    path = args["path"]
+    vdi_type = args["vdi_type"]
+    #TODO: HERE CANCEL
+    # blktap2.cancel()
+
+def is_openers(session, args):
+    path = args["path"]
+    openers_pid= util.get_openers_pid(path)
+    if openers_pid:
+        return str(True)
+    else:
+        return str(False)
 
 if __name__ == "__main__":
     import XenAPIPlugin
     XenAPIPlugin.dispatch({
         "multi": multi,
         "is_open": is_open,
-        "refresh_lun_size_by_SCSIid": refresh_lun_size_by_SCSIid})
+        "refresh_lun_size_by_SCSIid": refresh_lun_size_by_SCSIid,
+        "is_openers": is_openers,
+        "commit_tapdisk": commit_tapdisk,
+        "commit_cancel": commit_cancel,
+        })

drivers/qcow2util.py

@@ -718,17 +718,17 @@ def getBlockBitmap(self, path: str) -> bytes:
         return zlib.compress(self._create_bitmap())
 
     @override
-    def coalesce(self, path: str) -> int:
-        pid_openers = util.get_openers_pid(path)
+    def coalesceOnline(self, path: str) -> int:
+        pid_openers = util.get_openers_pid(path) # TODO: need to check other hosts and call tap-ctl on the host where it runs
         if pid_openers:
             if len(pid_openers) > 1:
-                util.SMlog("Multiple openers for {}".format(path)) #TODO: There might be multiple PID?
+                raise xs_errors.XenError("Multiple openers for {}".format(path)) # TODO: There might be multiple PID? Yes, we can have the chain enabled for multiple leaf (i.e. after a clone)
             pid = pid_openers[0]
             l = TapCtl.list(pid=pid)
             if len(l) > 1: #TODO: There might more than one minor for this blktap?
                 raise xs_errors.XenError("TapdiskAlreadyRunning", "There is multiple minor for this tapdisk process")
             minor = l[0]["minor"]
-            TapCtl.commit(pid, minor, QCOW2_TYPE, path) #TODO: Handle commit call failing
+            TapCtl.commit(pid, minor, QCOW2_TYPE, path) #TODO: Handle commit call failing, it's needed if the tapdisk hasn't started yet or crashed
             #We need to wait for query to return concluded
             #TODO: We are technically ininterruptible since being interrupted will only stop checking if the job is done
             # We should call `tap-ctl cancel` if we are interrupted
@@ -742,11 +742,11 @@ def coalesce(self, path: str) -> int:
                     util.SMlog("Got status {} for tapdisk {} minor: {}".format(status, pid, minor))
                 return nb
             except TapCtl.CommandFailure as e:
-                # util.SMlog("Query command failed on tapdisk instance {}. Retrying with offline coalesce...".format(pid))
-                # return self.coalesce(path)
                 util.SMlog("Query command failed on tapdisk instance {}. Raising...".format(pid))
                 raise
-        else:
+
+    @override
+    def coalesce(self, path: str) -> int:
             allocated_blocks = self.getAllocatedSize(path)
             # -d on commit make it not empty the original image since we don't intend to keep it
             cmd = [QEMU_IMG, "commit", "-f", QCOW2_TYPE, path, "-d"]
@@ -822,3 +822,7 @@ def getKeyHash(self, path: str) -> Optional[str]:
     @override
     def setKey(self, path: str, key_hash: str) -> None:
         pass
+
+    @override
+    def coalesceOnRemote(self) -> bool:
+        return True

drivers/util.py

@@ -2115,7 +2115,7 @@ def gen_path(path):
         SMlog('* End profiling of {} ({}) *'.format(name, filename))
 
 
-def strtobool(str):
+def strtobool(str: str) -> bool:
     # Note: `distutils` package is deprecated and slated for removal in Python 3.12.
     # There is not alternative for strtobool.
     # See: https://peps.python.org/pep-0632/#migration-advice

drivers/vhdutil.py

@@ -442,6 +442,14 @@ def setKey(self, path: str, key_hash: str) -> None:
         """
         self._ioretry([VHD_UTIL, "key", "-s", "-n", path, "-H", key_hash])
 
+    @override
+    def coalesceOnRemote(self) -> bool:
+        return False
+
+    @abstractmethod
+    def coalesceOnline(self, path: str) -> int:
+        raise NotImplementedError("Online coalesce not implemented for vhdutil")
+
     @staticmethod
     def _convertAllocatedSizeToBytes(size: int):
         # Assume we have standard 2MB allocation blocks