diff --git a/operations/management/__init__.py b/operations/management/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/operations/management/commands/__init__.py b/operations/management/commands/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/operations/management/commands/run_worker.py b/operations/management/commands/run_worker.py new file mode 100644 index 00000000..c8c9d760 --- /dev/null +++ b/operations/management/commands/run_worker.py @@ -0,0 +1,11 @@ +from django.core.management.base import BaseCommand, CommandError + +from operations.worker import Worker + + +class Command(BaseCommand): + help = "Run the job worker (one per installation)." + + def handle(self, *args, **options): + if Worker().run() != 0: + raise CommandError("Another worker runs on this host") diff --git a/operations/tests/fakes.py b/operations/tests/fakes.py index db547013..3a8f6313 100644 --- a/operations/tests/fakes.py +++ b/operations/tests/fakes.py @@ -1,9 +1,14 @@ """Kinds used by the operations tests; registered on import. test.scripted takes its behaviour from params and from barriers that tests install before -submitting (keyed by submission key).""" +submitting (keyed by submission key). start_worker and stop_worker run a +worker's loop in a thread.""" + +import threading + +from django.db import connection from operations.jobs import Cancelled, Kind, Outcome, Plan, register -from operations.models import Claim, Operation +from operations.models import Claim, Operation, acquire BARRIERS = {} # submission key -> threading.Event the run waits on CANCELLED = [] @@ -14,6 +19,31 @@ def reset(): CANCELLED.clear() +def start_worker(worker): + """The loop of Worker.run in a thread, without its process parts.""" + acquire(worker.identity, worker.host) + worker.recovery.schedule_orphans() + + def loop(): + try: + while not worker.stopping.is_set(): + worker.tick_once() + worker.stopping.wait(worker.tick) + finally: + connection.close() + + thread = threading.Thread(target=loop, daemon=True) + thread.start() + return thread + + +def stop_worker(worker, thread): + worker.stopping.set() + thread.join(10) + for pool in (worker.pool, worker.control, worker.progress_pool): + pool.shutdown(wait=True, cancel_futures=True) + + @register class NoopKind(Kind): name = "test.noop" diff --git a/operations/tests/test_worker.py b/operations/tests/test_worker.py new file mode 100644 index 00000000..77924d2a --- /dev/null +++ b/operations/tests/test_worker.py @@ -0,0 +1,261 @@ +"""The worker runs queued jobs in its pool, sends errors before effects to +failed and after effects to inspection, honours cancel, and never records or +finishes after losing its lease.""" + +import os +import tempfile +import threading +import time +from unittest.mock import patch + +from django.conf import settings +from django.contrib.auth import get_user_model +from django.db import OperationalError, connection +from django.test import TransactionTestCase + +from operations import jobs +from operations.models import Claim, Operation, WorkerLease +from operations.tests import fakes +from operations.worker import Worker, data_dir, host_identity +from vrtManager.util import ConnectionFailed + + +def wait_for(predicate, seconds=15): + end = time.monotonic() + seconds + while time.monotonic() < end: + if predicate(): + return True + time.sleep(0.05) + return False + + +def state(op): + return Operation.objects.get(pk=op.pk).state + + +class WorkerTestCase(TransactionTestCase): + def setUp(self): + fakes.reset() + self.user = get_user_model().objects.create_user("wk_user", "w@example.com", "pw") + self.worker = None + self.thread = None + + def start(self, concurrency=2): + self.worker = Worker(concurrency=concurrency, tick=0.05, heartbeat_every=0.1, progress_every=0.1, grace=1, + exit_on_lost_lease=False) + self.thread = fakes.start_worker(self.worker) + + def tearDown(self): + for barrier in fakes.BARRIERS.values(): + barrier.set() + if self.thread: + fakes.stop_worker(self.worker, self.thread) + connection.close() + + def submit(self, kind, params, key, barrier=False): + if barrier: + fakes.BARRIERS[key] = threading.Event() # before the worker can see the job + return jobs.submit(kind, self.user, params, key) + + def test_a_queued_job_runs_and_succeeds(self): + op = self.submit("test.noop", {}, "key-00000001") + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.SUCCEEDED)) + self.assertFalse(Claim.objects.exists()) + + def test_an_error_before_effects_fails_the_job(self): + op = self.submit("test.scripted", {"fail_before": True}, "key-00000001") + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.FAILED)) + + def test_a_connection_error_fails_the_job_without_naming_the_host(self): + op = self.submit("test.noop", {}, "key-00000001") + failure = ConnectionFailed("Cannot connect to qemu+ssh://root@wvc-test-host.example.com/system") + with patch.object(fakes.NoopKind, "precheck", side_effect=failure): + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.FAILED)) + error = Operation.objects.get(pk=op.pk).error + self.assertEqual(error, "The host of this virtual machine cannot be reached right now.") + self.assertNotIn("example.com", error) + + def test_an_error_after_effects_goes_to_inspection(self): + op = self.submit("test.scripted", {"fail_after": True, "outcome": "uncertain"}, "key-00000001") + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.UNCERTAIN)) + self.assertTrue(Claim.objects.filter(operation=op).exists()) + + def test_no_more_jobs_than_slots_run_at_once(self): + ops = [self.submit("test.scripted", {"uuid": f"u{i}"}, f"key-0000000{i}", barrier=True) for i in range(3)] + self.start() + self.assertTrue(wait_for(lambda: Operation.objects.filter(state=Operation.RUNNING).count() == 2)) + time.sleep(0.3) + self.assertEqual(Operation.objects.filter(state=Operation.QUEUED).count(), 1) + for barrier in fakes.BARRIERS.values(): + barrier.set() + self.assertTrue(wait_for(lambda: all(state(op) == Operation.SUCCEEDED for op in ops))) + + def test_a_cancelled_running_job_ends_cancelled(self): + op = self.submit("test.scripted", {}, "key-00000001", barrier=True) + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.RUNNING)) + jobs.cancel(op.pk, self.user) + self.assertTrue(wait_for(lambda: state(op) == Operation.CANCELLED)) + self.assertFalse(Claim.objects.exists()) + + def test_a_worker_that_lost_its_lease_does_not_finish_jobs(self): + op = self.submit("test.scripted", {}, "key-00000001", barrier=True) + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.RUNNING)) + WorkerLease.objects.update(holder="other-host:1:x", host="other-host") + self.assertTrue(wait_for(lambda: self.worker.lost_lease)) + fakes.BARRIERS["key-00000001"].set() + time.sleep(0.5) + self.assertEqual(state(op), Operation.RUNNING) # left for recovery + + def test_no_progress_is_written_after_the_lease_is_lost(self): + release = threading.Event() + + def slow_progress(op): + release.wait(5) + return (1, 2) + + op = self.submit("test.scripted", {}, "key-00000001", barrier=True) + fakes.ScriptedKind.progress = staticmethod(slow_progress) + try: + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.RUNNING)) + WorkerLease.objects.update(holder="other-host:1:x", host="other-host") + release.set() + self.assertTrue(wait_for(lambda: self.worker.lost_lease)) + time.sleep(0.3) + self.assertIsNone(Operation.objects.get(pk=op.pk).progress_total) + finally: + release.set() + del fakes.ScriptedKind.progress + + def test_heartbeats_continue_while_a_job_blocks(self): + op = self.submit("test.scripted", {}, "key-00000001", barrier=True) + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.RUNNING)) + first = Operation.objects.get(pk=op.pk).heartbeat_at + self.assertTrue(wait_for(lambda: Operation.objects.get(pk=op.pk).heartbeat_at > first)) + + def test_one_progress_query_per_job_at_a_time(self): + calls, release = [], threading.Event() + + def slow_progress(op): + calls.append(op.pk) + release.wait(5) + return (1, 2) + + self.submit("test.scripted", {}, "key-00000001", barrier=True) + fakes.ScriptedKind.progress = staticmethod(slow_progress) + try: + self.start() + self.assertTrue(wait_for(lambda: len(calls) == 1)) + time.sleep(0.5) # several progress intervals while the query hangs + self.assertEqual(len(calls), 1) + finally: + release.set() + del fakes.ScriptedKind.progress + + def test_progress_queries_never_wait_for_a_thread(self): + calls, release = [], threading.Event() + + def hanging_progress(op): + calls.append(op.pk) + release.wait(5) + return (1, 2) + + for i in range(3): + self.submit("test.scripted", {"uuid": f"u{i}"}, f"key-0000000{i}", barrier=True) + fakes.ScriptedKind.progress = staticmethod(hanging_progress) + try: + self.start(concurrency=3) + self.assertTrue(wait_for(lambda: len(calls) == 2)) + time.sleep(0.5) + self.assertEqual(len(calls), 2) + self.assertEqual(self.worker.progress_pool._work_queue.qsize(), 0) + finally: + release.set() + del fakes.ScriptedKind.progress + + def test_progress_reaches_every_job_in_turn(self): + ops = [self.submit("test.scripted", {"uuid": f"u{i}"}, f"key-0000000{i}", barrier=True) for i in range(3)] + fakes.ScriptedKind.progress = staticmethod(lambda op: (1, 2)) + try: + self.start(concurrency=3) + self.assertTrue(wait_for( + lambda: all(Operation.objects.get(pk=op.pk).progress_total == 2 for op in ops))) + finally: + del fakes.ScriptedKind.progress + + def test_a_cancel_during_the_drain_is_honoured(self): + op = self.submit("test.scripted", {}, "key-00000001", barrier=True) + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.RUNNING)) + self.worker.stopping.set() # as SIGTERM: the loop ends, the drain begins + self.thread.join(10) + jobs.cancel(op.pk, self.user) + self.worker._drain() + self.assertEqual(state(op), Operation.CANCELLED) + + def test_a_job_thread_that_raised_unfinished_goes_to_inspection(self): + op = self.submit("test.noop", {}, "key-00000001") + with patch("operations.jobs.Context", side_effect=RuntimeError("no context")): + self.start() + self.assertTrue(wait_for(lambda: state(op) == Operation.UNCERTAIN)) + self.assertIn("no context", Operation.objects.get(pk=op.pk).error) + + def test_a_failing_manager_loop_ends_the_process(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.worker = Worker(data=directory.name) + with ( + patch.object(self.worker, "tick_once", side_effect=OperationalError("database is down")), + patch("operations.worker.signal.signal"), + patch("operations.worker.logging.shutdown"), + patch("operations.worker.os._exit", side_effect=SystemExit) as exit_, + ): + with self.assertRaises(SystemExit): + self.worker.run() + exit_.assert_called_once_with(1) + + +class DataDirTestCase(TransactionTestCase): + def test_the_data_dir_does_not_follow_the_log_file(self): + with self.settings(WEBVIRTCLOUD_LOG_FILE="/var/log/wvc-test/webvirtcloud.log"): + moved = data_dir() + self.assertEqual(str(moved), str(settings.BASE_DIR / "data")) + + +class HostIdentityTestCase(TransactionTestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.dir = directory.name + + def test_the_same_data_dir_is_the_same_host(self): + self.assertEqual(host_identity(self.dir), host_identity(self.dir)) + self.assertTrue(Worker(data=self.dir).identity.startswith(f"{host_identity(self.dir)}:")) + + def test_another_data_dir_is_another_host(self): + other = tempfile.TemporaryDirectory() + self.addCleanup(other.cleanup) + self.assertNotEqual(host_identity(self.dir), host_identity(other.name)) + + def test_an_id_another_worker_created_first_is_kept(self): + def lose_the_race(src, dst): + open(dst, "w").write("wvc-test-first") + raise FileExistsError(dst) + + with patch("operations.worker.os.link", side_effect=lose_the_race): + self.assertEqual(host_identity(self.dir), "wvc-test-first") + self.assertEqual(sorted(os.listdir(self.dir)), ["node_id"]) + + def test_a_new_hostname_on_the_same_data_dir_is_the_same_host(self): + # a recreated container gets a new hostname but keeps its data volume + with patch("operations.worker.socket.gethostname", return_value="wvc-test-a"): + first = host_identity(self.dir) + with patch("operations.worker.socket.gethostname", return_value="wvc-test-b"): + self.assertEqual(host_identity(self.dir), first) diff --git a/operations/utils.py b/operations/utils.py index fbdfadf3..18b1c0e2 100644 --- a/operations/utils.py +++ b/operations/utils.py @@ -1,16 +1,21 @@ -"""Secrets never reach a job record, a response or a log line. What a refused -request says about the job holding the resource: details for its owner and -superusers, a minimal summary for everyone else. Worker settings from the -Settings page. The worker reads them itself (the settings middleware only runs -in gunicorn).""" +"""Secrets never reach a job record, a response or a log line, and users who +do not see the hosts get "host unreachable" instead of a connection error's +text (which names the host). What a refused request says about the job holding +the resource: details for its owner and superusers, a minimal summary for +everyone else. Worker settings from the Settings page. The worker reads them +itself (the settings middleware only runs in gunicorn).""" import re +import libvirt from django.utils import timezone from django.utils.timesince import timesince from django.utils.translation import gettext as _ +from django.utils.translation import gettext_lazy +from libvirt import libvirtError from appsettings.models import AppSettings +from vrtManager.util import ConnectionFailed from . import jobs @@ -25,6 +30,23 @@ _PATTERNS = [ ] _SECRET_KEY = re.compile(r"pass|secret|token|private|key$", re.I) +# errors of the connection to the host, not of the operation: both the +# domain and the code must match (a daemon that lacks a procedure answers +# VIR_FROM_RPC / VIR_ERR_NO_SUPPORT, which the user should see) +TRANSPORT_ERROR_DOMAINS = {libvirt.VIR_FROM_RPC, libvirt.VIR_FROM_REMOTE, libvirt.VIR_FROM_SSH, libvirt.VIR_FROM_LIBSSH} +TRANSPORT_ERROR_CODES = { + libvirt.VIR_ERR_INTERNAL_ERROR, # "connection closed due to keepalive timeout" + libvirt.VIR_ERR_SYSTEM_ERROR, # "Cannot recv data: ..." + libvirt.VIR_ERR_RPC, + libvirt.VIR_ERR_NO_CONNECT, + libvirt.VIR_ERR_AUTH_FAILED, + libvirt.VIR_ERR_AUTH_UNAVAILABLE, + libvirt.VIR_ERR_AUTH_CANCELLED, + libvirt.VIR_ERR_SSH, + libvirt.VIR_ERR_LIBSSH, +} +UNREACHABLE = gettext_lazy("The host of this virtual machine cannot be reached right now.") + BOUNDS = {"WORKER_CONCURRENCY": (1, 16, 4), "JOB_RETENTION_DAYS": (1, 3650, 30)} @@ -35,6 +57,17 @@ def safe_text(text): return text +def public_error_text(exception): + """The error for users who do not see the hosts: a connection error names + the host and how it is reached, so they learn only that it cannot be reached.""" + connection_error = isinstance(exception, ConnectionFailed) or ( + isinstance(exception, libvirtError) + and exception.get_error_domain() in TRANSPORT_ERROR_DOMAINS + and exception.get_error_code() in TRANSPORT_ERROR_CODES + ) + return str(UNREACHABLE) if connection_error else str(exception) + + def check_params(params): if isinstance(params, dict): for key, value in params.items(): diff --git a/operations/worker.py b/operations/worker.py new file mode 100644 index 00000000..a4885341 --- /dev/null +++ b/operations/worker.py @@ -0,0 +1,310 @@ +"""The job worker (spec section 3). The manager loop never calls libvirt: job +threads run kinds; control threads run cancel hooks and recovery; progress +threads read progress. Every write is fenced by the lease. Recovery of jobs +the worker cannot finish itself. Minimal: a job that raised after its effects +started goes straight to inspection (uncertain).""" + +import fcntl +import logging +import os +import signal +import socket +import threading +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from datetime import timedelta +from pathlib import Path + +from django.conf import settings +from django.db import close_old_connections +from django.utils import timezone + +from . import claims, jobs, utils +from .models import Claim, LeaseLost, Operation, acquire, fenced, renew + +logger = logging.getLogger(__name__) + +MAX_SLOTS = 16 +PROGRESS_THREADS = 2 +HOUSEKEEPING_EVERY = 3600.0 + + +def data_dir(): + """The installation's persistent data directory (the Docker volume).""" + data = settings.BASE_DIR / "data" + os.makedirs(data, exist_ok=True) + return data + + +def host_identity(data): + """The worker's host: a random id created once in the data directory, + which also holds the worker's flock. A recreated container on the same + volume is the same host (the flock proves its old worker is gone); + another installation sharing the database is another host, even with + the same hostname. The hostname is not part of it: Docker gives a + recreated container a new one.""" + path = Path(data) / "node_id" + if not path.exists(): + new = path.with_name(f"node_id.{uuid.uuid4().hex}") # PIDs repeat across containers + new.write_text(uuid.uuid4().hex[:12]) + try: + os.link(new, path) # never replaces an id another process just created + except FileExistsError: + pass + finally: + new.unlink() + return path.read_text().strip() + + +class Worker: + def __init__(self, concurrency=None, tick=1.0, heartbeat_every=10.0, progress_every=5.0, grace=60.0, + data=None, exit_on_lost_lease=True): + data = Path(data or data_dir()) + self.host = host_identity(data) + self.identity = f"{self.host}:{os.getpid()}:{uuid.uuid4().hex[:8]}" + self.fixed_concurrency = concurrency + self.tick = tick + self.heartbeat_every = heartbeat_every + self.progress_every = progress_every + self.grace = grace + self.lock_path = data / "worker.lock" + self.exit_on_lost_lease = exit_on_lost_lease + self.stopping = threading.Event() + self.lost_lease = False + self.jobs = {} + self.signalled = set() + self.progress_pending = set() + self.progress_asked = {} # job -> monotonic time its progress was last asked + self.pool = ThreadPoolExecutor(MAX_SLOTS, thread_name_prefix="wvc-job") + self.control = ThreadPoolExecutor(4, thread_name_prefix="wvc-control") + self.progress_pool = ThreadPoolExecutor(PROGRESS_THREADS, thread_name_prefix="wvc-progress") + self.recovery = Scheduler(self) + self._last = {"heartbeat": 0.0, "progress": 0.0, "housekeeping": 0.0} + + # -- process ------------------------------------------------------------ + + def run(self): + """Until SIGTERM. Returns 1 if another worker runs on this host. Ends + with os._exit, so job threads past the grace period end with it; a + failing manager loop ends the process the same way (status 1).""" + lock = open(self.lock_path, "a") + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + logger.error("Another worker runs on this host; exiting") + return 1 + signal.signal(signal.SIGTERM, lambda *_: self.stopping.set()) + signal.signal(signal.SIGINT, lambda *_: self.stopping.set()) + status = 0 + try: + while not acquire(self.identity, self.host): + if self.stopping.wait(5): + return 0 + logger.info("Worker %s started on %s", self.identity, socket.gethostname()) + self.recovery.schedule_orphans() + while not self.stopping.is_set(): + self.tick_once() + self.stopping.wait(self.tick) + self._drain() + except Exception as err: + logger.error("Worker %s failed: %s", self.identity, utils.safe_text(err)) + status = 1 + logger.info("Worker %s stopped", self.identity) + logging.shutdown() + os._exit(status) + + def _drain(self): + end = time.monotonic() + self.grace + while self.jobs and time.monotonic() < end: + if not self._renew(): + return + self._heartbeat() + self._reap() + self._cancels() + time.sleep(min(self.tick, 1.0)) + + def _lose_lease(self): + logger.error("Worker %s lost its lease", self.identity) + self.lost_lease = True + self.stopping.set() + if self.exit_on_lost_lease: + logging.shutdown() + os._exit(1) # self-fencing: no job thread of ours reaches libvirt again + + def _renew(self): + if not renew(self.identity): + self._lose_lease() + return False + return True + + # -- loop --------------------------------------------------------------- + + def tick_once(self): + now = time.monotonic() + if now - self._last["heartbeat"] >= self.heartbeat_every: + self._last["heartbeat"] = now + if not self._renew(): + return + self._heartbeat() + self._reap() + self._cancels() + self._limits() + if now - self._last["progress"] >= self.progress_every: + self._last["progress"] = now + self._progress() + self.recovery.due() + self._claim() + if now - self._last["housekeeping"] >= HOUSEKEEPING_EVERY: + self._last["housekeeping"] = now + self.control.submit(_guarded, housekeeping) + + def _mine(self, pks): + """This worker's active jobs among pks; used only inside fenced.""" + return Operation.objects.filter(pk__in=list(pks), worker=self.identity, state__in=Operation.ACTIVE) + + def _fenced_update(self, pks, **fields): + try: + with fenced(self.identity): + return self._mine(pks).update(**fields) + except LeaseLost: + self._lose_lease() + return 0 + + def _heartbeat(self): + if self.jobs: + self._fenced_update(self.jobs, heartbeat_at=timezone.now()) + + def _reap(self): + for op_id, future in list(self.jobs.items()): + if future.done(): + del self.jobs[op_id] + self.signalled.discard(op_id) + self.progress_asked.pop(op_id, None) + if (err := future.exception()) is not None: + # execute raised before it could finish the job itself + logger.error("job %s ended unfinished: %s", op_id, utils.safe_text(err)) + self.control.submit(_guarded, self.recovery.inspect_now, op_id, utils.public_error_text(err)) + + def _slots(self): + return (self.fixed_concurrency or utils.concurrency()) - len(self.jobs) + + def _claim(self): + while self._slots() > 0 and not self.stopping.is_set(): + candidate = (Operation.objects.filter(state=Operation.QUEUED).order_by("created_at") + .values_list("pk", flat=True).first()) + if candidate is None: + return + try: + with fenced(self.identity): + now = timezone.now() + taken = Operation.objects.filter(pk=candidate, state=Operation.QUEUED).update( + state=Operation.RUNNING, started_at=now, heartbeat_at=now, worker=self.identity) + except LeaseLost: + self._lose_lease() + return + if taken: + self.jobs[candidate] = self.pool.submit(execute, self, candidate) + + def _cancels(self): + for op in Operation.objects.filter(pk__in=list(self.jobs), cancel_requested=True): + if op.pk not in self.signalled: + self.signalled.add(op.pk) + self.control.submit(_guarded, jobs.get_kind(op.kind).cancel, op) + + def _limits(self): + now = timezone.now() + for op in Operation.objects.filter(pk__in=list(self.jobs), time_limit__isnull=False): + if op.started_at and now - op.started_at > timedelta(seconds=op.time_limit): + if jobs.get_kind(op.kind).cancellable and not op.cancel_requested: + self._fenced_update([op.pk], state=Operation.CANCELLING, cancel_requested=True) + elif "(over its time limit)" not in op.phase: + self._fenced_update([op.pk], phase=f"{op.phase} (over its time limit)"[:255]) + + def _progress(self): + """At most one query per job and none waiting for a thread, so queries + that hang in libvirt do not pile up; the least recently asked first.""" + ops = Operation.objects.filter(pk__in=list(self.jobs)) + for op in sorted(ops, key=lambda op: self.progress_asked.get(op.pk, 0.0)): + if len(self.progress_pending) >= PROGRESS_THREADS: + return + if op.pk not in self.progress_pending: + self.progress_pending.add(op.pk) + self.progress_asked[op.pk] = time.monotonic() + self.progress_pool.submit(self._read_progress, op) + + def _read_progress(self, op): + try: + value = jobs.get_kind(op.kind).progress(op) + if value is not None: + with fenced(self.identity): + self._mine([op.pk]).update(progress_done=value[0], progress_total=value[1]) + except Exception as err: + logger.debug("progress of job %s: %s", op.pk, utils.safe_text(err)) + finally: + self.progress_pending.discard(op.pk) + close_old_connections() + + +def _guarded(fn, *args): + try: + fn(*args) + except Exception as err: + logger.error("worker task %s failed: %s", getattr(fn, "__name__", fn), utils.safe_text(err)) + finally: + close_old_connections() + + +def execute(worker, op_id): + """A job thread: precheck and run. Errors before effects fail the job; + after effects they go to inspection; a lost lease leaves it alone.""" + op = Operation.objects.get(pk=op_id) + kind = jobs.get_kind(op.kind) + ctx = jobs.Context(op, worker) + try: + with claims.adopted(op): + kind.precheck(op, ctx) + result = kind.run(op, ctx) + jobs.finish(op, jobs.Outcome(Operation.SUCCEEDED, result=result), worker=worker) + except LeaseLost: + logger.warning("job %s left for recovery: this worker lost its lease", op_id) + except jobs.Cancelled: + jobs.finish(op, jobs.Outcome(Operation.CANCELLED), worker=worker) + except Exception as err: + logger.warning("job %s %s raised: %s", op.pk, op.kind, utils.safe_text(err)) + try: + # jobs have no request: the error users see never names a host + if ctx.effects_started: + worker.recovery.inspect_now(op.pk, error=utils.public_error_text(err)) + else: + jobs.finish(op, jobs.Outcome(Operation.FAILED, error=utils.public_error_text(err)), worker=worker) + except LeaseLost: + pass + finally: + close_old_connections() + + +def housekeeping(): + cutoff = timezone.now() - timedelta(days=utils.retention_days()) + old = list( + Operation.objects.filter(state__in=Operation.TERMINAL, finished_at__lt=cutoff) + .exclude(state=Operation.UNCERTAIN).exclude(claims__isnull=False) + .values_list("pk", flat=True)[:500] + ) + Operation.objects.filter(pk__in=old).delete() + Claim.objects.filter(operation__isnull=True, expires_at__lt=timezone.now()).delete() + + +class Scheduler: + def __init__(self, worker): + self.worker = worker + + def schedule_orphans(self): + pass + + def due(self): + pass + + def inspect_now(self, op_id, error=""): + op = Operation.objects.get(pk=op_id) + jobs.finish(op, jobs.Outcome(Operation.UNCERTAIN, error=error), worker=self.worker) diff --git a/vrtManager/connection.py b/vrtManager/connection.py index 7fec5de0..b12d510a 100644 --- a/vrtManager/connection.py +++ b/vrtManager/connection.py @@ -98,7 +98,6 @@ NVRAM_DIR = "/var/lib/libvirt/qemu/nvram" class wvmConnection(object): """ class representing a single connection stored in the Connection Manager - # to-do: may also need some locking to ensure to not connect simultaniously in 2 threads """ def __init__(self, host, login, passwd, conn): @@ -185,13 +184,19 @@ class wvmConnection(object): return self.__connect_socket() raise ValueError(f'"{self.type}" is not a valid connection type') - @property - def connected(self): + def alive_connection(self): + """The libvirt connection if it is alive, else None. self.connection + is read once: the close callback can clear it at any moment.""" + connection = self.connection try: - return self.connection is not None and self.connection.isAlive() + return connection if connection is not None and connection.isAlive() else None except libvirtError: # isAlive failed for some reason - return False + return None + + @property + def connected(self): + return self.alive_connection() is not None def __libvirt_auth_credentials_callback(self, credentials, user_data): for credential in credentials: @@ -341,16 +346,14 @@ class wvmConnectionManager(object): finally: self._connections_lock.release() - if not connection.connected: + live = connection.alive_connection() + if live is None: # (re-)connect within CONNECT_TIMEOUT; one open per connection at a time connection.connect() - - if connection.connected: - # return libvirt connection object - return connection.connection - else: - # raise libvirt error + live = connection.alive_connection() + if live is None: raise util.ConnectionFailed(connection.last_error) + return live def host_is_up(self, conn_type, hostname, timeout=1): """True if the libvirt endpoint of hostname accepts a TCP (or Unix diff --git a/vrtManager/tests/test_connection_threads.py b/vrtManager/tests/test_connection_threads.py new file mode 100644 index 00000000..037f2c5f --- /dev/null +++ b/vrtManager/tests/test_connection_threads.py @@ -0,0 +1,117 @@ +"""The worker uses the connection cache from many threads at once: one +cached connection per credentials, one reopen after a close, and never a +connection that the close callback cleared meanwhile.""" + +import threading +import time +import unittest +from unittest.mock import MagicMock, patch + +from django.conf import settings + +if not settings.configured: + settings.configure(MAC_OUI="52:54:10") + +import libvirt +from vrtManager.connection import CONN_TCP, wvmConnectionManager + + +def alive_connection(): + conn = MagicMock() + conn.isAlive.return_value = 1 + return conn + + +class ConnectionThreadsTestCase(unittest.TestCase): + def setUp(self): + patcher = patch("vrtManager.connection.wvmEventLoop") + patcher.start() + self.addCleanup(patcher.stop) + self.manager = wvmConnectionManager() + self.release = threading.Event() + self.addCleanup(self.release.set) + self.opened = [] + + def open_auth(self, *conns): + pending = list(conns) + + def open_auth(uri, auth, flags): + self.release.wait(5) + conn = pending.pop(0) + self.opened.append(conn) + return conn + + return patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth) + + def get(self): + try: + return self.manager.get_connection("192.0.2.10", "u", "p", CONN_TCP) + except libvirt.libvirtError as err: + return err + + def get_from_threads(self, count=8): + results = [None] * count + + def worker(i): + results[i] = self.get() + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)] + for t in threads: + t.start() + time.sleep(0.2) # every thread is waiting for the open + self.release.set() + for t in threads: + t.join(10) + return results + + def entries(self): + return self.manager._connections.get("192.0.2.10", []) + + def test_threads_share_one_cached_connection(self): + conn = alive_connection() + with self.open_auth(conn): + results = self.get_from_threads() + self.assertEqual(results, [conn] * 8) + self.assertEqual(self.opened, [conn]) + self.assertEqual(len(self.entries()), 1) + + def test_after_a_close_threads_reopen_once(self): + old, new = alive_connection(), alive_connection() + with self.open_auth(old, new): + self.release.set() + self.assertIs(self.get(), old) + self.release.clear() + entry = self.entries()[0] + entry._wvmConnection__connection_close_callback(old, 2) + results = self.get_from_threads() + self.assertEqual(results, [new] * 8) + self.assertEqual(self.opened, [old, new]) + self.assertEqual(len(self.entries()), 1) + + def test_a_close_during_the_lookup_never_returns_none(self): + """The close callback runs on the event loop thread, at any moment of + another thread's get_connection.""" + for nth in range(1, 4): + with self.subTest(nth=nth): + self.manager = wvmConnectionManager() + conn = alive_connection() + self.release.set() + with self.open_auth(conn): + self.assertIs(self.get(), conn) + entry = self.entries()[0] + calls = [] + + def is_alive(): + calls.append(1) + if len(calls) == nth and not entry.connection_state_lock.locked(): + entry._wvmConnection__connection_close_callback(conn, 2) + return 1 + + conn.isAlive.side_effect = is_alive + with self.open_auth(alive_connection()): + result = self.get() + self.assertIsNotNone(result) + + +if __name__ == "__main__": + unittest.main() diff --git a/webvirtcloud/middleware.py b/webvirtcloud/middleware.py index cfba43a0..6a2b703d 100644 --- a/webvirtcloud/middleware.py +++ b/webvirtcloud/middleware.py @@ -2,47 +2,23 @@ from django.contrib import messages from django.http import JsonResponse from django.shortcuts import redirect, render from django.utils.translation import gettext_lazy as _ -import libvirt from libvirt import libvirtError from operations import claims -from operations.utils import busy_message, summary -from vrtManager.util import ConnectionFailed +from operations.utils import busy_message, public_error_text, summary API_PREFIX = "/api/" -# errors of the connection to the host, not of the operation: both the -# domain and the code must match (a daemon that lacks a procedure answers -# VIR_FROM_RPC / VIR_ERR_NO_SUPPORT, which the user should see) -TRANSPORT_ERROR_DOMAINS = {libvirt.VIR_FROM_RPC, libvirt.VIR_FROM_REMOTE, libvirt.VIR_FROM_SSH, libvirt.VIR_FROM_LIBSSH} -TRANSPORT_ERROR_CODES = { - libvirt.VIR_ERR_INTERNAL_ERROR, # "connection closed due to keepalive timeout" - libvirt.VIR_ERR_SYSTEM_ERROR, # "Cannot recv data: ..." - libvirt.VIR_ERR_RPC, - libvirt.VIR_ERR_NO_CONNECT, - libvirt.VIR_ERR_AUTH_FAILED, - libvirt.VIR_ERR_AUTH_UNAVAILABLE, - libvirt.VIR_ERR_AUTH_CANCELLED, - libvirt.VIR_ERR_SSH, - libvirt.VIR_ERR_LIBSSH, -} -UNREACHABLE = _("The host of this virtual machine cannot be reached right now.") def error_text(request, exception): """ - The libvirt error to show. A connection error names the host and how it - is reached, which only administrators (superusers and the global - view_instances permission, who see the hosts) get; other users learn - that the host cannot be reached. + The libvirt error to show. Administrators (superusers and the global + view_instances permission, who see the hosts) get the full error; other + users the public text. """ - connection_error = isinstance(exception, ConnectionFailed) or ( - isinstance(exception, libvirtError) - and exception.get_error_domain() in TRANSPORT_ERROR_DOMAINS - and exception.get_error_code() in TRANSPORT_ERROR_CODES - ) user = getattr(request, "user", None) - if connection_error and not (user and (user.is_superuser or user.has_perm("instances.view_instances"))): - return str(UNREACHABLE) - return str(exception) + if user and (user.is_superuser or user.has_perm("instances.view_instances")): + return str(exception) + return public_error_text(exception) class ExceptionMiddleware: