mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
10b46afd36
libvirt has no timeout for opening a connection. Measured in the test setup: an unreachable host blocked qemu+tcp for 135 s and qemu+ssh for 133 s; a daemon that accepts and never answers blocks forever. Gunicorn runs sync workers and the instance list connects to every compute, so one unreachable compute held a worker for minutes per page load. - A connection is opened in a helper thread, at most one per connection at a time. The request waits LIBVIRT_CONNECT_TIMEOUT (default 5 s). An open nobody waits for any more keeps running; until it ends, requests get its error at once, and if it succeeds later the connection is used. A failed open is not repeated for LIBVIRT_RETRY_AFTER (default 30 s). A stuck host holds one thread and does not keep other hosts from connecting. - qemu+ssh runs ssh through vrtManager/libvirt-ssh (the URI's command parameter), which adds ConnectTimeout=10 and BatchMode=yes, so an abandoned ssh process ends and no prompt can hang. The user's ssh configuration (aliases, ProxyJump, keys) still applies. - The close callback acts only for the connection it was registered for: a late callback of a replaced connection dropped its successor. The reason is reported as text. A connection that closed before it was published is an error. - Compute.status is a bool (it returned the exception object, so the "host is down" check in refresh_instance_database never fired); the message is Compute.connection_error. Unit tests cover the timeout, retry delay, late success, single open, a stuck host next to a healthy one and the close callback. Live tests (vrtManager/test_live_connection.py), for TCP and SSH computes of the test setup: an unreachable address (no ssh left after 10 s), an endpoint that never answers, a slow open completing later, a host that goes silent on an established connection (keepalive ends calls within 45 s, the next request reconnects), an ssh_config alias and a ProxyJump host.
210 lines
7.6 KiB
Python
210 lines
7.6 KiB
Python
"""Reachability probe and connections that do not block each other."""
|
|
|
|
import os
|
|
import socket
|
|
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_SSH, CONN_TCP, SSH_COMMAND, 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()
|
|
|
|
|
|
|
|
def alive_connection():
|
|
conn = MagicMock()
|
|
conn.isAlive.return_value = 1
|
|
return conn
|
|
|
|
|
|
class BoundedOpenTestCase(unittest.TestCase):
|
|
"""Opening waits at most CONNECT_TIMEOUT; a failed host is not retried
|
|
for RETRY_AFTER; one open per connection at a time."""
|
|
|
|
def setUp(self):
|
|
for name, value in (("wvmEventLoop", MagicMock()), ("CONNECT_TIMEOUT", 0.3), ("RETRY_AFTER", 1.0)):
|
|
patcher = patch(f"vrtManager.connection.{name}", value)
|
|
patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
self.manager = wvmConnectionManager()
|
|
self.release = threading.Event()
|
|
self.addCleanup(self.release.set)
|
|
self.calls = 0
|
|
|
|
def open_auth(self, result):
|
|
def open_auth(uri, auth, flags):
|
|
self.calls += 1
|
|
self.release.wait(5)
|
|
if isinstance(result, Exception):
|
|
raise result
|
|
return result
|
|
return patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth)
|
|
|
|
def get(self, host="h"):
|
|
started = time.monotonic()
|
|
try:
|
|
return self.manager.get_connection(host, "u", "p", CONN_TCP), time.monotonic() - started
|
|
except libvirt.libvirtError as err:
|
|
return err, time.monotonic() - started
|
|
|
|
def test_a_hanging_open_is_given_up_after_the_timeout(self):
|
|
with self.open_auth(alive_connection()):
|
|
result, took = self.get()
|
|
self.assertIsInstance(result, libvirt.libvirtError)
|
|
self.assertIn("did not answer within", str(result))
|
|
self.assertLess(took, 1.0)
|
|
|
|
def test_within_retry_after_the_last_error_comes_at_once(self):
|
|
with self.open_auth(alive_connection()):
|
|
self.get()
|
|
result, took = self.get()
|
|
self.assertIn("did not answer within", str(result))
|
|
self.assertLess(took, 0.1)
|
|
self.assertEqual(self.calls, 1)
|
|
|
|
def test_a_late_success_is_used(self):
|
|
conn = alive_connection()
|
|
with self.open_auth(conn):
|
|
self.get()
|
|
self.release.set()
|
|
time.sleep(0.1)
|
|
result, took = self.get()
|
|
self.assertIs(result, conn)
|
|
self.assertEqual(self.calls, 1)
|
|
|
|
def test_a_failed_host_is_tried_again_after_retry_after(self):
|
|
self.release.set()
|
|
with self.open_auth(libvirt.libvirtError("refused")):
|
|
self.get()
|
|
self.get()
|
|
self.assertEqual(self.calls, 1)
|
|
time.sleep(1.1)
|
|
self.get()
|
|
self.assertEqual(self.calls, 2)
|
|
|
|
def test_concurrent_requests_share_one_open(self):
|
|
with self.open_auth(alive_connection()):
|
|
threads = [threading.Thread(target=self.get) for _ in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
self.assertEqual(self.calls, 1)
|
|
|
|
def test_a_stuck_host_does_not_keep_other_hosts_from_connecting(self):
|
|
healthy = alive_connection()
|
|
|
|
def open_auth(uri, auth, flags):
|
|
if "stuck" in uri:
|
|
self.release.wait(5) # an open that never returns
|
|
return healthy
|
|
|
|
with patch("vrtManager.connection.libvirt.openAuth", side_effect=open_auth):
|
|
for i in range(12):
|
|
self.get(f"stuck{i}")
|
|
result, took = self.get("healthy")
|
|
self.assertIs(result, healthy)
|
|
|
|
def test_an_abandoned_open_is_not_waited_for_again_nor_duplicated(self):
|
|
conn = alive_connection()
|
|
with self.open_auth(conn):
|
|
self.get()
|
|
time.sleep(1.1) # past RETRY_AFTER, the open still runs
|
|
result, took = self.get()
|
|
self.assertIsInstance(result, libvirt.libvirtError)
|
|
self.assertLess(took, 0.1)
|
|
self.release.set() # the host answers at last
|
|
time.sleep(0.1)
|
|
result, _ = self.get()
|
|
self.assertIs(result, conn)
|
|
self.assertEqual(self.calls, 1)
|
|
|
|
def test_a_late_close_callback_does_not_drop_the_new_connection(self):
|
|
self.release.set()
|
|
old, new = alive_connection(), alive_connection()
|
|
with self.open_auth(old):
|
|
self.get()
|
|
entry = self.manager._search_connection("h", "u", "p", CONN_TCP)
|
|
old.isAlive.return_value = 0 # seen dead, reconnected
|
|
with self.open_auth(new):
|
|
self.assertIs(self.get()[0], new)
|
|
entry._wvmConnection__connection_close_callback(old, 1)
|
|
self.assertIs(self.get()[0], new)
|
|
entry._wvmConnection__connection_close_callback(new, 2)
|
|
self.assertIsNone(entry.connection)
|
|
self.assertEqual(entry.last_error, "Connection closed (keepalive timeout)")
|
|
|
|
def test_a_connection_closed_before_it_is_published_is_an_error(self):
|
|
self.release.set()
|
|
dead = alive_connection()
|
|
dead.isAlive.return_value = 0
|
|
with self.open_auth(dead):
|
|
result, _ = self.get()
|
|
self.assertIn("closed right after opening", str(result))
|
|
|
|
def test_ssh_runs_ssh_with_a_connect_timeout(self):
|
|
self.release.set()
|
|
with patch("vrtManager.connection.libvirt.open", return_value=alive_connection()) as libvirt_open:
|
|
self.manager.get_connection("host", "root", "", CONN_SSH)
|
|
uri = libvirt_open.call_args[0][0]
|
|
self.assertTrue(uri.startswith("qemu+ssh://root@host/system?command="))
|
|
self.assertTrue(os.access(SSH_COMMAND, os.X_OK))
|
|
with open(SSH_COMMAND) as script:
|
|
self.assertIn("ConnectTimeout", script.read())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|