fix(instances): a migration that does not finish in time is cancelled, not cut off

A VM that changes its memory faster than the network carries it never
finishes a live migration without auto converge: measured with a guest
rewriting about 1.2 GB/s over an 8.5 Gbit/s link. The migration then
ran until gunicorn ended the request after 600 s. libvirt cancelled it
when the request went away and the VM stayed on the source, but the
user got an error page and nothing was logged.

The app now cancels a migration that has not finished after 540 s
(MIGRATION_TIME_LIMIT, below the 600 s of gunicorn.conf.py; a test
keeps it below). The failure is reported and logged like any other:
the migration was cancelled, the VM stays on the source host, and a
VM like that needs auto converge. A timer that fires as the migration
ends finishes before the compute locks are released, and an error
the cancel did not cause keeps its own message.

vrtManager/instance.py

@@ -4,6 +4,7 @@ import json
 import logging
 import os.path
 import string
+import threading
 import time
 
 logger = logging.getLogger(__name__)
@@ -113,7 +114,7 @@ class wvmInstances(wvmConnect):
         dom = self.get_instance(name)
         dom.resume()
 
-    def moveto(self, conn, name, live, unsafe, offline, autoconverge=False, compress=False, uri=None):
+    def moveto(self, conn, name, live, unsafe, offline, autoconverge=False, compress=False, uri=None, timeout=None):
         """Migrates the VM of conn (a wvmInstance on the source) here.
 
         The mode follows the VM's state, read once: a running VM migrates
@@ -122,8 +123,9 @@ class wvmInstances(wvmConnect):
         fit the state is refused. unsafe, autoconverge (live only) and
         compress apply to a running VM. uri: the native migration URI
         (tcp://address) on the destination; None lets libvirt use the
-        destination's own hostname. Returns the mode and options used, for
-        the log."""
+        destination's own hostname. timeout: seconds after which a migration
+        that has not finished is cancelled (None: no limit). Returns the mode
+        and options used, for the log."""
         state = conn.get_status()
         if state == 5:
             if live:
@@ -179,9 +181,43 @@ class wvmInstances(wvmConnect):
             # dom.migrate.
             self.move_definition(dom)
         else:
-            dom.migrate(self.wvm, flags, None, uri, 0)
+            self.migrate_within(dom, flags, uri, timeout)
         return ", ".join(mode)
 
+    def migrate_within(self, dom, flags, uri, timeout):
+        """dom.migrate, cancelled when it has not finished after timeout
+        seconds: a VM that changes its memory faster than the network carries
+        it never finishes, and the request would be ended by the web server
+        with no word. Cancelled, the VM stays on the source."""
+        cancelled = threading.Event()
+
+        def cancel():
+            with contextlib.suppress(libvirtError):  # finished meanwhile
+                dom.abortJob()
+                cancelled.set()
+
+        timer = threading.Timer(timeout, cancel) if timeout else None
+        if timer:
+            timer.daemon = True
+            timer.start()
+        try:
+            dom.migrate(self.wvm, flags, None, uri, 0)
+        except libvirtError as error:
+            if timer:
+                timer.cancel()
+                timer.join()  # a cancel under way ends before the locks go
+            if cancelled.is_set():
+                raise util.OperationError(
+                    f"The migration did not finish within {timeout} s and was cancelled ({error}); the VM stays on "
+                    "the source host. A VM that changes its memory faster than the network carries it needs auto "
+                    "converge."
+                ) from error
+            raise
+        finally:
+            if timer:
+                timer.cancel()
+                timer.join()
+
     def missing_disks(self, definitions):
         """(target, source) of the disks in the VM definitions (XML) that are
         not volumes of this host's active storage pools: the VM's own files,

vrtManager/test_migration.py

@@ -1,7 +1,9 @@
 """Migration: the source definition never stays behind; a shut-off VM's
 definition is moved by the app, not by libvirt's offline migration."""
 
+import threading
 import unittest
+from unittest.mock import patch
 
 import libvirt
 from django.conf import settings
@@ -400,3 +402,64 @@ class DestinationDisksTestCase(unittest.TestCase):
         xml = with_disks("<disk type='volume' device='disk'><source pool='p' volume='v'/><target dev='vdc'/></disk>")
         with self.assertRaisesRegex(util.OperationError, "vdc \\(p/v\\)"):
             self.migrate(host, xml)
+
+
+class SlowDomain(FakeDomain):
+    """A migration that does not finish until its job is aborted."""
+
+    def __init__(self, **kw):
+        super().__init__(**kw)
+        self.aborted = threading.Event()
+
+    def migrate(self, dconn, flags, dname, uri, bandwidth):
+        if not self.aborted.wait(5):
+            raise AssertionError("the migration was not aborted")
+        raise libvirt.libvirtError("operation aborted: job 'migration out' canceled by client")
+
+    def abortJob(self):
+        self.aborted.set()
+
+
+class MigrationTimeLimitTestCase(unittest.TestCase):
+    """A migration that does not converge is cancelled after the time limit,
+    before the web server ends the request; the VM stays on the source."""
+
+    def test_a_migration_over_the_limit_is_cancelled(self):
+        source = FakeSource(1)
+        source.instance = SlowDomain()
+        with self.assertRaisesRegex(util.OperationError, "did not finish within 0.05 s and was cancelled.*auto converge"):
+            destination().moveto(source, "vm", live=True, unsafe=False, offline=False, timeout=0.05)
+        self.assertTrue(source.instance.aborted.is_set())
+
+    def test_a_migration_within_the_limit_stops_its_timer(self):
+        timers = []
+
+        class Timer(threading.Timer):
+            def __init__(self, *args):
+                super().__init__(*args)
+                self.joined = False
+                timers.append(self)
+
+            def join(self, timeout=None):
+                self.joined = True
+                super().join(timeout)
+
+        source = FakeSource(1)
+        with patch("vrtManager.instance.threading.Timer", Timer):
+            destination().moveto(source, "vm", live=True, unsafe=False, offline=False, timeout=5)
+        self.assertEqual(len(source.instance.calls), 1)
+        self.assertTrue(timers[0].finished.is_set() and timers[0].joined)
+        self.assertFalse(timers[0].is_alive())
+
+    def test_a_failure_the_cancel_did_not_cause_keeps_its_error(self):
+        source = FakeSource(1)
+        source.instance = SlowDomain()
+
+        def refused():
+            source.instance.aborted.set()  # migrate() gives up, but the abort itself failed
+            raise libvirt.libvirtError("cannot abort migration in confirm phase")
+
+        source.instance.abortJob = refused
+        with self.assertRaisesRegex(libvirt.libvirtError, "operation aborted") as caught:
+            destination().moveto(source, "vm", live=True, unsafe=False, offline=False, timeout=0.05)
+        self.assertNotIsInstance(caught.exception, util.OperationError)