fix(console): the serial console worker keeps pastes whole and says what failed

- A paste reaches the VM whole: the non-blocking console stream takes it
  in parts, or none while the VM is busy, and the rest was dropped. The
  worker sends the rest until the VM has it, for at most five seconds.
- A libvirt error while the console is open (the host gone, the VM
  undefined or migrated) ends the session with one line in the terminal
  instead of a Python traceback; libvirt no longer prints its errors on
  the terminal. Errors are shown as users see them elsewhere: a host that
  cannot be reached is not named.
- The terminal says when the VM is not running and its console will open
  when it starts. The terminal is raw before the console opens, and the
  stream is dropped once when both threads see the VM stop.

conf/daemon/consolecallback

@@ -20,8 +20,18 @@ django.setup()
 import libvirt  # noqa: E402
 
 from instances.models import Instance  # noqa: E402
+from operations.utils import public_error_text  # noqa: E402
 from vrtManager.connection import connection_manager  # noqa: E402
 
+# libvirt prints its errors on stderr, which is the user's terminal: they are
+# raised and told in one line instead
+libvirt.registerErrorHandler(lambda context, error: None, None)
+SEND_WAIT = 5  # seconds a paste may wait for the VM to take it
+
+
+def say(text):
+    os.write(1, ("\r\n[%s]\r\n" % text).encode())
+
 
 class Console:
     def __init__(self, connection, uuid):
@@ -34,14 +44,18 @@ class Console:
         """Open the stream while the VM runs, drop it while it does not."""
         active = self.domain.state()[0] in (libvirt.VIR_DOMAIN_RUNNING, libvirt.VIR_DOMAIN_PAUSED)
         if active and self.stream is None:
-            self.stream = self.connection.newStream(libvirt.VIR_STREAM_NONBLOCK)
-            self.domain.openConsole(None, self.stream, 0)  # a console in use is refused
-            self.stream.eventAddCallback(libvirt.VIR_STREAM_EVENT_READABLE, from_vm, self)
+            stream = self.connection.newStream(libvirt.VIR_STREAM_NONBLOCK)
+            self.domain.openConsole(None, stream, 0)  # a console in use is refused
+            stream.eventAddCallback(libvirt.VIR_STREAM_EVENT_READABLE, from_vm, self)
+            self.stream = stream
         elif not active and self.stream is not None:
             self.close_stream()
+            say("The VM is not running: its console opens again when it starts")
 
     def close_stream(self):
         stream, self.stream = self.stream, None
+        if stream is None:  # the other thread closed it
+            return
         try:
             stream.eventRemoveCallback()
             stream.abort()
@@ -56,11 +70,19 @@ def from_client(watch, fd, events, console):
         data = b""
     if not data:  # the client is gone
         console.running = False
-    elif console.stream is not None:
-        try:
-            console.stream.send(data)
-        except libvirt.libvirtError:
-            console.close_stream()
+        return
+    stream = console.stream  # the main thread may drop it meanwhile
+    deadline = time.monotonic() + SEND_WAIT
+    try:
+        # a non-blocking stream takes part of a paste, or none (-2) while the VM is busy
+        while data and stream is not None and time.monotonic() < deadline:
+            sent = stream.send(data)
+            if sent == -2:
+                time.sleep(0.01)
+            else:
+                data = data[sent:]
+    except libvirt.libvirtError:
+        console.close_stream()
 
 
 def from_vm(stream, events, console):
@@ -80,15 +102,22 @@ def main():
     c = instance.compute
     try:
         console = Console(connection_manager.get_connection(c.hostname, c.login, c.password, c.type), instance.uuid)
+        tty.setraw(0)  # before the console opens: nothing typed is echoed by the terminal
         console.check()
     except libvirt.libvirtError as err:
-        sys.stdout.write("Cannot open the console of %s: %s\r\n" % (instance.name, err.get_error_message()))
+        say("Cannot open the console of %s: %s" % (instance.name, public_error_text(err)))
         sys.exit(1)
-    tty.setraw(0)
+    if console.stream is None:
+        say("The VM is not running: its console opens when it starts")
     libvirt.virEventAddHandle(0, libvirt.VIR_EVENT_HANDLE_READABLE, from_client, console)
-    while console.running:
-        time.sleep(0.2)
-        console.check()
+    try:
+        while console.running:
+            time.sleep(0.2)
+            console.check()
+    except libvirt.libvirtError as err:  # the host went away, the VM was undefined or migrated
+        say("The console of %s closed: %s" % (instance.name, public_error_text(err)))
+        sys.exit(1)
 
 
-main()
+if __name__ == "__main__":
+    main()

console/tests/test_consolecallback.py

@@ -0,0 +1,59 @@
+"""The serial console worker: a paste reaches the VM whole although its
+stream takes it in parts or is busy, and the stream is dropped once."""
+
+from importlib.machinery import SourceFileLoader
+from importlib.util import module_from_spec, spec_from_loader
+from pathlib import Path
+from unittest import TestCase
+from unittest.mock import MagicMock, patch
+
+CONSOLECALLBACK = Path(__file__).resolve().parent.parent.parent / "conf" / "daemon" / "consolecallback"
+
+
+def load():
+    loader = SourceFileLoader("consolecallback_under_test", str(CONSOLECALLBACK))
+    module = module_from_spec(spec_from_loader(loader.name, loader))
+    loader.exec_module(module)
+    return module
+
+
+class ConsoleWorkerTests(TestCase):
+    def setUp(self):
+        self.cb = load()
+        self.console = MagicMock(running=True)
+
+    def test_a_paste_reaches_the_vm_whole(self):
+        received = []
+
+        def send(data):
+            if len(received) % 2 == 0:  # busy every other time
+                received.append(b"")
+                return -2
+            received.append(data[:3])  # takes three bytes at a time
+            return 3
+
+        self.console.stream.send.side_effect = send
+        with patch.object(self.cb.os, "read", return_value=b"echo hello\r"), patch.object(self.cb.time, "sleep"):
+            self.cb.from_client(None, 0, None, self.console)
+        self.assertEqual(b"".join(received), b"echo hello\r")
+
+    def test_a_vm_that_takes_nothing_does_not_hold_the_console_for_ever(self):
+        self.console.stream.send.return_value = -2
+        clock = iter(range(0, 100))
+        with patch.object(self.cb.os, "read", return_value=b"x"), patch.object(self.cb.time, "sleep"), \
+                patch.object(self.cb.time, "monotonic", side_effect=lambda: next(clock)):
+            self.cb.from_client(None, 0, None, self.console)
+        self.assertLessEqual(self.console.stream.send.call_count, self.cb.SEND_WAIT + 1)
+
+    def test_the_stream_is_dropped_once(self):
+        console = self.cb.Console.__new__(self.cb.Console)
+        console.stream = stream = MagicMock()
+        console.close_stream()
+        console.close_stream()  # the other thread, a moment later
+        stream.abort.assert_called_once_with()
+
+    def test_a_closed_client_ends_the_worker(self):
+        with patch.object(self.cb.os, "read", return_value=b""):
+            self.cb.from_client(None, 0, None, self.console)
+        self.assertFalse(self.console.running)
+        self.console.stream.send.assert_not_called()