simplify log printing in localstack start command

localstack · thrau · Sep 18, 2023 · Sep 15, 2023 · Sep 15, 2023 · Sep 15, 2023
commit 2e62049aae83f7d741c3e8a40ed4db5d2a9e2e16
diff --git a/localstack/utils/bootstrap.py b/localstack/utils/bootstrap.py
@@ -10,7 +10,7 @@
 import threading
 import time
 from functools import wraps
-from typing import Any, Dict, Iterable, List, Optional, Set, Union
+from typing import Any, Callable, Dict, Iterable, List, Optional, Set, Union
 
 from localstack import config, constants
 from localstack.config import HostAndPort, default_ip, is_env_true
@@ -38,7 +38,6 @@
 from localstack.utils.serving import Server
 from localstack.utils.strings import short_uid
 from localstack.utils.sync import poll_condition
-from localstack.utils.tail import FileListener
 
 LOG = logging.getLogger(__name__)
 
@@ -907,8 +906,40 @@ def stopped(self) -> Container:
         return Container(container_config=self.config, docker_client=self.container_client)
 
 
+class ContainerLogPrinter:
+    def __init__(self, container: Container, callback: Callable[[str], None] = print):
+        self.container = container
+        self.callback = callback
+
+        self._closed = threading.Event()
+        self._stream: Optional[CancellableStream] = None
+
+    def _can_start_streaming(self):
+        if self._closed.is_set():
+            raise IOError("Already stopped")
+        if not self.container.running_container:
+            return False
+        return self.container.running_container.is_running()
+
+    def run(self):
+        try:
+            poll_condition(self._can_start_streaming)
+        except IOError:
+            return
+        self._stream = self.container.running_container.stream_logs()
+        for line in self._stream:
+            self.callback(line.rstrip(b"\r\n").decode("utf-8"))
+
+    def close(self):
+        self._closed.set()
+        if self._stream:
+            self._stream.close()
+
+
 class LocalstackContainerServer(Server):
-    def __init__(self, container_configuration: ContainerConfiguration | None = None) -> None:
+    def __init__(
+        self, container_configuration: ContainerConfiguration | Container | None = None
+    ) -> None:
         super().__init__(config.EDGE_PORT, config.EDGE_BIND_HOST)
 
         if container_configuration is None:
@@ -927,7 +958,10 @@ def __init__(self, container_configuration: ContainerConfiguration | None = None
                 env_vars={},
             )
 
-        self.container: Container | RunningContainer = Container(container_configuration)
+        if isinstance(container_configuration, Container):
+            self.container = container_configuration
+        else:
+            self.container: Container | RunningContainer = Container(container_configuration)
 
     def is_up(self) -> bool:
         """
@@ -1098,10 +1132,7 @@ def _init_log_printer(line):
         print(line)
         log_printer.callback = print
 
-    logfile = get_container_default_logfile_location(container_config.name)
-    log_printer = FileListener(logfile, callback=_init_log_printer)
-    log_printer.truncate_file()
-    log_printer.start()
+    log_printer = ContainerLogPrinter(container, callback=_init_log_printer)
 
     # Set up signal handler, to enable clean shutdown across different operating systems.
     #  There are subtle differences across operating systems and terminal emulators when it
@@ -1116,16 +1147,19 @@ def shutdown_handler(*args):
             shutdown_event.set()
         print("Shutting down...")
         server.shutdown()
-        log_printer.close()
 
     shutdown_event = threading.Event()
     shutdown_event_lock = threading.RLock()
     signal.signal(signal.SIGINT, shutdown_handler)
 
     # start the Localstack container as a Server
-    server = LocalstackContainerServer(container_config)
+    server = LocalstackContainerServer(container)
+    log_printer_thread = threading.Thread(
+        target=log_printer.run, name="container-log-printer", daemon=True
+    )
     try:
         server.start()
+        log_printer_thread.start()
         server.join()
         error = server.get_error()
         if error:
@@ -1134,6 +1168,8 @@ def shutdown_handler(*args):
     except KeyboardInterrupt:
         print("ok, bye!")
         shutdown_handler()
+    finally:
+        log_printer.close()
 
 
 def ensure_container_image(console, container: Container):

diff --git a/localstack/utils/common.py b/localstack/utils/common.py
@@ -176,9 +176,6 @@
     wait_until,
 )
 
-# TODO: remove imports from here (need to update any client code that imports these from utils.common)
-from localstack.utils.tail import FileListener  # noqa
-
 # TODO: remove imports from here (need to update any client code that imports these from utils.common)
 from localstack.utils.threads import (  # noqa
     TMP_PROCESSES,

diff --git a/localstack/utils/tail.py b/localstack/utils/tail.py
diff --git a/tests/unit/utils/test_common.py b/tests/unit/utils/test_common.py
@@ -1,7 +1,5 @@
-import os.path
 import re
 import threading
-import time
 import unittest
 from unittest.mock import MagicMock
 
@@ -18,8 +16,7 @@
 from localstack.utils.files import load_file, new_tmp_file, rm_rf
 from localstack.utils.json import FileMappedDocument
 from localstack.utils.run import run
-from localstack.utils.sync import poll_condition, synchronized
-from localstack.utils.tail import FileListener
+from localstack.utils.sync import synchronized
 
 
 class SynchronizedTest(unittest.TestCase):
@@ -77,84 +74,6 @@ def _run(cmd):
         _run(["echo 'foo bar 123'"])
 
 
-@pytest.mark.parametrize("tail_engine", ["command", "tailer"])
-class TestFileListener:
-    def test_basic_usage(self, tail_engine, tmp_path):
-        lines = []
-
-        file = tmp_path / "log.txt"
-        file.touch()
-        fd = open(file, "a")
-        listener = FileListener(str(file), lines.append)
-        listener.use_tail_command = tail_engine != "tailer"
-
-        try:
-            listener.start()
-            assert listener.started.is_set()
-            fd.write("hello" + os.linesep)
-            fd.write("pytest" + os.linesep)
-            fd.flush()
-
-            assert poll_condition(lambda: len(lines) == 2, timeout=3), (
-                "expected two lines to appear. %s" % lines
-            )
-
-            assert lines[0] == "hello"
-            assert lines[1] == "pytest"
-        finally:
-            listener.close()
-
-        try:
-            fd.write("foobar" + os.linesep)
-            time.sleep(0.5)
-            assert len(lines) == 2, "expected listener.stop() to stop listening on new "
-        finally:
-            fd.close()
-
-    def test_callback_exception_ignored(self, tail_engine, tmp_path):
-        lines = []
-
-        def callback(line):
-            if "throw" in line:
-                raise ValueError("oh noes")
-
-            lines.append(line)
-
-        file = tmp_path / "log.txt"
-        file.touch()
-        fd = open(file, "a")
-        listener = FileListener(str(file), callback)
-        listener.use_tail_command = tail_engine != "tailer"
-
-        try:
-            listener.start()
-            assert listener.started.is_set()
-            fd.write("hello" + os.linesep)
-            fd.flush()
-            fd.write("throw" + os.linesep)
-            fd.write("pytest" + os.linesep)
-            fd.flush()
-
-            assert poll_condition(lambda: len(lines) == 2, timeout=3), (
-                "expected two lines to appear. %s" % lines
-            )
-
-            assert lines[0] == "hello"
-            assert lines[1] == "pytest"
-        finally:
-            fd.close()
-            listener.close()
-
-    def test_open_missing_file(self, tail_engine):
-        lines = []
-
-        listener = FileListener("/tmp/does/not/exist", lines.append)
-        listener.use_tail_command = tail_engine != "tailer"
-
-        with pytest.raises(FileNotFoundError):
-            listener.start()
-
-
 class TestFileMappedDocument:
     def test_load_without_file_succeeds(self, tmp_path):
         path = tmp_path / "doc.json"