From c2d3285e0f964a5ad5ded446062259969d25ecad Mon Sep 17 00:00:00 2001 From: catborise Date: Thu, 15 Oct 2026 20:42:50 +0300 Subject: [PATCH] feat(instances): a running migration can be cancelled; its time limit goes --- gunicorn.conf.py | 2 - instances/jobs.py | 32 ++++++++-- instances/tests/test_migrate_job.py | 37 ++++++++++- instances/tests/test_migration_bookkeeping.py | 12 +--- instances/utils.py | 6 -- vrtManager/instance.py | 44 ++----------- vrtManager/tests/test_migration.py | 63 ------------------- 7 files changed, 69 insertions(+), 127 deletions(-) diff --git a/gunicorn.conf.py b/gunicorn.conf.py index 32e6f804..372e4777 100644 --- a/gunicorn.conf.py +++ b/gunicorn.conf.py @@ -87,8 +87,6 @@ def get_workers(): workers = get_workers() # worker_class = 'egg:gunicorn#eventlet' worker_connections = 1000 -# Must exceed instances.utils.MIGRATION_TIME_LIMIT (540 s) plus time to answer; -# a lower value would kill a worker in the middle of a live migration. timeout = 600 keepalive = 2 diff --git a/instances/jobs.py b/instances/jobs.py index e5c16417..f26db6b3 100644 --- a/instances/jobs.py +++ b/instances/jobs.py @@ -2,6 +2,7 @@ import math import re +import time from django.core.exceptions import PermissionDenied from django.utils.translation import gettext_noop @@ -19,7 +20,7 @@ from computes.models import Compute from instances import utils from instances.models import Instance from operations import claims -from operations.jobs import Kind, Outcome, Plan, register +from operations.jobs import Cancelled, Kind, Outcome, Plan, register from operations.models import Claim, Operation @@ -610,6 +611,8 @@ def _domain(compute_pk, uuid): class MigrateInstance(Kind): name = "instance.migrate" label = gettext_noop("Migrate VM") + cancellable = True # libvirt aborts a migration; the VM keeps running on the source + progress_bytes = True # libvirt job data processed/total def normalize(self, raw): return {"instance": int(raw["instance"]), "target": int(raw["target"]), @@ -643,11 +646,32 @@ class MigrateInstance(Kind): instance = _instance(p["instance"]) ctx.record_intent() ctx.set_phase(gettext_noop("Migrating the VM")) - # migrate_instance logs the outcome - utils.migrate_instance(Compute.objects.get(pk=p["target"]), instance, op.user, - **{k: p[k] for k in MIGRATE_OPTIONS}) + try: + # migrate_instance logs the outcome + utils.migrate_instance(Compute.objects.get(pk=p["target"]), instance, op.user, + **{k: p[k] for k in MIGRATE_OPTIONS}) + except libvirtError: + op.refresh_from_db(fields=["cancel_requested"]) + if op.cancel_requested: + raise Cancelled from None + raise return {} + def progress(self, op): + info = _proxy(_instance(op.params["instance"])).instance.jobInfo() + return (info[4], info[3]) if info[3] else None + + def cancel(self, op): + dom = _proxy(_instance(op.params["instance"])).instance + # the worker sends a cancel once: one that comes before libvirt began + # the migration waits for it, or for the job to end without it + while not _libvirt_busy(dom): + if not Operation.objects.filter(pk=op.pk, state__in=Operation.ACTIVE).exists(): + return + time.sleep(1) + # an error (the migration ended meanwhile) goes to the worker log + dom.abortJob() + def reconcile(self, op): source = _domain(op.intent["compute"], op.intent["uuid"]) dest = _domain(op.params["target"], op.intent["uuid"]) diff --git a/instances/tests/test_migrate_job.py b/instances/tests/test_migrate_job.py index 09c2f2ba..ea2c2aba 100644 --- a/instances/tests/test_migrate_job.py +++ b/instances/tests/test_migrate_job.py @@ -11,7 +11,8 @@ from appsettings.settings import get_settings from computes.models import Compute from instances.jobs import MigrateInstance from instances.models import Instance -from operations.jobs import Context, get_kind +from libvirt import libvirtError +from operations.jobs import Cancelled, Context, get_kind from operations.models import Claim, Operation, acquire UUID = "31313131-4242-5353-6464-757575757575" @@ -74,3 +75,37 @@ class MigrateJobTestCase(TransactionTestCase): self.assertEqual(MigrateInstance().reconcile(op).state, Operation.SUCCEEDED) self.vm.refresh_from_db() self.assertEqual(self.vm.compute, self.dest) + + def test_a_cancelled_migration_ends_cancelled_and_the_vm_stays(self): + op = self.queue(live_migrate="true") + ctx = self.start(op) + self.dest_conn.moveto.side_effect = libvirtError("operation aborted: migration out: canceled by client") + Operation.objects.filter(pk=op.pk).update(cancel_requested=True) + with self.assertRaises(Cancelled): + get_kind(op.kind).run(op, ctx) + self.vm.refresh_from_db() + self.assertEqual(self.vm.compute, self.source) + + def test_cancel_and_progress_ask_the_source_vm(self): + op = self.queue(live_migrate="true") + with patch("instances.jobs.wvmInstance") as proxy: + proxy.return_value.instance.jobInfo.return_value = [2, 0, 0, 1000, 250] + [0] * 7 + self.assertEqual(get_kind(op.kind).progress(op), (250, 1000)) + get_kind(op.kind).cancel(op) + proxy.return_value.instance.abortJob.assert_called_once_with() + + def test_a_cancel_before_libvirt_starts_the_migration_waits_for_it(self): + op = self.queue(live_migrate="true") + self.start(op) + with patch("instances.jobs.wvmInstance") as proxy, patch("instances.jobs.time.sleep") as sleep: + dom = proxy.return_value.instance + dom.isActive.return_value = True + dom.jobInfo.side_effect = [[0] * 12, [2] + [0] * 11] # no job yet, then the migration + get_kind(op.kind).cancel(op) + sleep.assert_called_once() + dom.abortJob.assert_called_once_with() + Operation.objects.filter(pk=op.pk).update(state=Operation.SUCCEEDED) # ended before libvirt began + with patch("instances.jobs.wvmInstance") as proxy: + proxy.return_value.instance.isActive.return_value = False + get_kind(op.kind).cancel(op) + proxy.return_value.instance.abortJob.assert_not_called() diff --git a/instances/tests/test_migration_bookkeeping.py b/instances/tests/test_migration_bookkeeping.py index 99f0deb5..9586e421 100644 --- a/instances/tests/test_migration_bookkeeping.py +++ b/instances/tests/test_migration_bookkeeping.py @@ -1,20 +1,17 @@ """What migrate_instance does around the migration itself: it records a finished migration even when autostart fails.""" -import re -from pathlib import Path from unittest.mock import MagicMock, patch from accounts.models import UserInstance from computes.models import Compute -from django.conf import settings from django.contrib.auth import get_user_model from django.db import DatabaseError from django.test import TestCase from logs.models import Logs from instances.models import Instance -from instances.utils import MIGRATION_TIME_LIMIT, migrate_instance +from instances.utils import migrate_instance from vrtManager import util @@ -71,10 +68,3 @@ class MigrationBookkeepingTests(TestCase): found.refresh_from_db() self.assertTrue(found.is_template) self.assertEqual(UserInstance.objects.get(user=owner).instance_id, found.id) - - def test_a_migration_is_cancelled_before_the_web_server_ends_the_request(self): - self.migrate() - self.assertEqual(self.moveto.call_args.kwargs["timeout"], MIGRATION_TIME_LIMIT) - gunicorn = (Path(settings.BASE_DIR) / "gunicorn.conf.py").read_text() - request_timeout = int(re.search(r"^timeout = (\d+)", gunicorn, re.M).group(1)) - self.assertLess(MIGRATION_TIME_LIMIT, request_timeout - 30) # time to answer and log diff --git a/instances/utils.py b/instances/utils.py index 3278bbd5..22586149 100644 --- a/instances/utils.py +++ b/instances/utils.py @@ -148,11 +148,6 @@ def get_network_tuple(network_source_str): return network_source_pack[0], "net" -# A migration that has not finished by then is cancelled: below the 600 s -# after which gunicorn (gunicorn.conf.py) ends the request. -MIGRATION_TIME_LIMIT = 540 - - def migrate_instance( new_compute, instance, @@ -186,7 +181,6 @@ def migrate_instance( autoconverge, compress, uri=new_compute.migration_uri, - timeout=MIGRATION_TIME_LIMIT, ) except Exception as error: msg = _("Instance migration to %(target)s failed: %(error)s") % { diff --git a/vrtManager/instance.py b/vrtManager/instance.py index fd734ba3..74310c7f 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -5,7 +5,6 @@ import logging import os.path import re import string -import threading import time logger = logging.getLogger(__name__) @@ -77,7 +76,7 @@ def _rate(before, after, seconds, scale=1): class wvmInstances(wvmConnect): - def moveto(self, conn, name, live, unsafe, offline, autoconverge=False, compress=False, uri=None, timeout=None): + def moveto(self, conn, name, live, unsafe, offline, autoconverge=False, compress=False, uri=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 @@ -86,9 +85,8 @@ 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. timeout: seconds after which a migration - that has not finished is cancelled (None: no limit). Returns the mode - and options used, for the log.""" + destination's own hostname. Returns the mode and options used, for + the log.""" state = conn.get_status() if state == 5: if live: @@ -150,7 +148,7 @@ class wvmInstances(wvmConnect): # dom.migrate. self.move_definition(dom) else: - self.migrate_within(dom, flags, uri, timeout) + dom.migrate(self.wvm, flags, None, uri, 0) return ", ".join(mode) @staticmethod @@ -172,40 +170,6 @@ class wvmInstances(wvmConnect): "host. Start the VM and migrate it live." ) - 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, diff --git a/vrtManager/tests/test_migration.py b/vrtManager/tests/test_migration.py index 8e46fcb2..4b09cd31 100644 --- a/vrtManager/tests/test_migration.py +++ b/vrtManager/tests/test_migration.py @@ -1,9 +1,7 @@ """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 @@ -427,64 +425,3 @@ class DestinationDisksTestCase(unittest.TestCase): xml = with_disks("") 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)