diff --git a/computes/test_quota_down_host.py b/computes/test_quota_down_host.py new file mode 100644 index 00000000..933388f4 --- /dev/null +++ b/computes/test_quota_down_host.py @@ -0,0 +1,25 @@ +import socket + +from accounts.models import UserAttributes, UserInstance +from appsettings.settings import get_settings +from computes.models import Compute +from django.contrib.auth import get_user_model +from django.test import TestCase +from instances.models import Instance +from instances.utils import check_user_quota + + +class QuotaWithDownHostTestCase(TestCase): + def test_vm_on_an_unreachable_host_does_not_break_the_quota_check(self): + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + user = get_user_model().objects.create_user("quota_down", password="pw") + UserAttributes.objects.create(user=user, max_instances=5, max_cpus=8, max_memory=8192, max_disk_size=100) + compute = Compute.objects.create(name="down", hostname=f"127.0.0.1:{port}", login="", password="", type=1) + vm = Instance.objects.create(compute=compute, name="vm", uuid="55555555-5555-5555-5555-555555555555") + UserInstance.objects.create(user=user, instance=vm) + get_settings() # app settings are normally loaded by middleware + + self.assertEqual(check_user_quota(user, 1, 1, 128, 1), "") diff --git a/vrtManager/connection.py b/vrtManager/connection.py index 34ff87b9..bb69fbf6 100644 --- a/vrtManager/connection.py +++ b/vrtManager/connection.py @@ -65,9 +65,6 @@ class wvmConnection(object): self.passwd = passwd self.type = conn - # connect - self.connect() - def connect(self): self.connection_state_lock.acquire() try: @@ -281,19 +278,15 @@ class wvmConnectionManager(object): # as the thread previously holding the write lock may have already added our connection connection = self._search_connection(host, login, passwd, conn) if connection is None: - # create a new connection if a matching connection does not already exist + # Only register it here: connecting under the global lock + # would make every other host wait for a slow one. connection = wvmConnection(host, login, passwd, conn) - - # add new connection to connection dict - if host in self._connections: - self._connections[host].append(connection) - else: - self._connections[host] = [connection] + self._connections.setdefault(host, []).append(connection) finally: self._connections_lock.release() - elif not connection.connected: - # try to (re-)connect if connection is closed + if not connection.connected: + # (re-)connect; the connection's own lock serializes concurrent attempts connection.connect() if connection.connected: @@ -303,34 +296,33 @@ class wvmConnectionManager(object): # raise libvirt error raise util.OperationError(connection.last_error) - def host_is_up(self, conn_type, hostname): - """ - returns True if the given host is up and we are able to establish - a connection using the given credentials. - """ + def host_is_up(self, conn_type, hostname, timeout=1): + """True if the libvirt endpoint of hostname accepts a TCP (or Unix + socket) connection. This is a reachability probe only: it does not + check credentials or libvirt itself. hostname may carry a port + ("host:port", "[v6]:port").""" + if conn_type == CONN_SOCKET: + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + try: + sock.settimeout(timeout) + sock.connect("/var/run/libvirt/libvirt-sock") + return True + except OSError: + return False + finally: + sock.close() + default_port = {CONN_SSH: SSH_PORT, CONN_TCP: TCP_PORT, CONN_TLS: TLS_PORT}.get(conn_type) + if default_port is None: + return False + host, port = str(hostname), default_port + match = re.fullmatch(r"\[?([^\[\]]+?)\]?:(\d+)", host) + if match and (host.startswith("[") or host.count(":") == 1): + host, port = match.group(1), int(match.group(2)) try: - socket_host = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - socket_host.settimeout(1) - if conn_type == CONN_SSH: - if ":" in hostname: - libvirt_host, PORT = hostname.split(":") - PORT = int(PORT) - else: - PORT = SSH_PORT - libvirt_host = hostname - socket_host.connect((libvirt_host, PORT)) - if conn_type == CONN_TCP: - socket_host.connect((hostname, TCP_PORT)) - if conn_type == CONN_TLS: - socket_host.connect((hostname, TLS_PORT)) - if conn_type == CONN_SOCKET: - socket_host = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - socket_host.connect("/var/run/libvirt/libvirt-sock") - socket_host.close() - return True - except Exception as err: - return err - + with socket.create_connection((host.strip("[]"), port), timeout=timeout): + return True + except OSError: + return False connection_manager = wvmConnectionManager( settings.LIBVIRT_KEEPALIVE_INTERVAL if hasattr(settings, "LIBVIRT_KEEPALIVE_INTERVAL") else 5, diff --git a/vrtManager/test_connection.py b/vrtManager/test_connection.py new file mode 100644 index 00000000..acfb1063 --- /dev/null +++ b/vrtManager/test_connection.py @@ -0,0 +1,65 @@ +"""R-09: reachability probe and connections that do not block each other.""" + +import socket +import threading +import time +import unittest +from unittest.mock import patch + +from django.conf import settings + +if not settings.configured: + settings.configure(MAC_OUI="52:54:10") + +from vrtManager.connection import CONN_TCP, wvmConnectionManager + + +def closed_port(): + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + port = sock.getsockname()[1] + sock.close() + return port + + +class HostIsUpTestCase(unittest.TestCase): + def setUp(self): + self.manager = wvmConnectionManager.__new__(wvmConnectionManager) + + def test_reachable_host_is_true(self): + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.listen(1) + self.addCleanup(server.close) + self.assertIs(self.manager.host_is_up(CONN_TCP, f"127.0.0.1:{server.getsockname()[1]}"), True) + + def test_unreachable_host_is_false_not_an_exception(self): + self.assertIs(self.manager.host_is_up(CONN_TCP, f"127.0.0.1:{closed_port()}"), False) + + +class ConnectionManagerTestCase(unittest.TestCase): + def setUp(self): + patcher = patch("vrtManager.connection.wvmEventLoop") + patcher.start() + self.addCleanup(patcher.stop) + self.manager = wvmConnectionManager() + + def test_a_slow_host_does_not_block_other_hosts(self): + def open_auth(uri, auth, flags): + if "slow" in uri: + time.sleep(1.5) + raise __import__("libvirt").libvirtError("refused") + + with patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth): + slow = threading.Thread(target=lambda: self.assertRaises(Exception, self.manager.get_connection, "slow", "u", "p", CONN_TCP)) + slow.start() + time.sleep(0.2) + started = time.monotonic() + with self.assertRaises(Exception): + self.manager.get_connection("fast", "u", "p", CONN_TCP) + self.assertLess(time.monotonic() - started, 1.0) + slow.join() + + +if __name__ == "__main__": + unittest.main()