diff --git a/instances/jobs.py b/instances/jobs.py index 76c5b320..8737d7d9 100644 --- a/instances/jobs.py +++ b/instances/jobs.py @@ -3,6 +3,7 @@ import math import os import re +import threading import time from django.db import transaction @@ -1047,17 +1048,33 @@ def _movable(user, params): return instance, _proxy(instance).disk_move_plan(params["dev"], params["pool"]) +USED_CHECK_WAIT = 60 # seconds a disk move waits for the computes to tell whether its old file is used + + def _used_anywhere(path): """Whether a VM or volume of any compute uses path (paths_in_use), as shared storage holds it for every host that mounts it; a compute that - cannot be read might.""" - for c in Compute.objects.all(): - try: - if path in wvmConnect(c.hostname, c.login, c.password, c.type).paths_in_use(): - return True - except libvirtError: - return True - return False + cannot be read might, and so might one that does not answer within + USED_CHECK_WAIT: libvirt has no call timeout, and a stuck host would + hold the job for as long as it is stuck.""" + computes = list(Compute.objects.all()) + answer = [] + + def check(): + for c in computes: + try: + if path in wvmConnect(c.hostname, c.login, c.password, c.type).paths_in_use(): + answer.append(True) + return + except libvirtError: + answer.append(True) + return + answer.append(False) + + thread = threading.Thread(target=check, daemon=True) + thread.start() + thread.join(USED_CHECK_WAIT) + return answer[0] if answer else True @register diff --git a/instances/tests/test_disk_move_job.py b/instances/tests/test_disk_move_job.py index 6841d6f6..7a421049 100644 --- a/instances/tests/test_disk_move_job.py +++ b/instances/tests/test_disk_move_job.py @@ -2,6 +2,8 @@ it, the worker moves it, recovery ends a copy no worker waits for and reads the outcome from the VM's definitions.""" +import threading +import time from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -129,3 +131,17 @@ class MoveDiskJobTestCase(TransactionTestCase): self.proxy.instance.blockJobAbort.reset_mock() self.assertEqual(self.reconcile({OLD}, job={"cur": 1, "end": 2}, mirror="/other").state, Operation.UNCERTAIN) self.proxy.instance.blockJobAbort.assert_not_called() + + +class UsedAnywhereTestCase(TransactionTestCase): + def test_a_host_that_does_not_answer_counts_as_using_the_file(self): + from instances import jobs + + Compute.objects.create(name="stuck", hostname="192.0.2.62", login="root", password="", type=1) + stuck = threading.Event() + self.addCleanup(stuck.set) + with patch("instances.jobs.wvmConnect") as hosts, patch.object(jobs, "USED_CHECK_WAIT", 0.2): + hosts.return_value.paths_in_use.side_effect = lambda: stuck.wait(10) and set() + started = time.monotonic() + self.assertTrue(jobs._used_anywhere(OLD)) + self.assertLess(time.monotonic() - started, 2)