mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-09 03:36:48 +00:00
feat(instances): a running migration can be cancelled; its time limit goes
This commit is contained in:
catborise
parent
31cbe60aa4
commit
c2d3285e0f
7 changed files with 69 additions and 127 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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"])
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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") % {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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("<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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue