Merge pull request #3075 from dnephin/reactive_logs

Display logs for new containers started after the command is run

compose/cli/log_printer.py

@@ -2,67 +2,152 @@ from __future__ import absolute_import
 from __future__ import unicode_literals
 
 import sys
+from collections import namedtuple
 from itertools import cycle
+from threading import Thread
+
+from six.moves import _thread as thread
+from six.moves.queue import Empty
+from six.moves.queue import Queue
 
 from . import colors
-from .multiplexer import Multiplexer
 from compose import utils
+from compose.cli.signals import ShutdownException
 from compose.utils import split_buffer
 
 
+class LogPresenter(object):
+
+    def __init__(self, prefix_width, color_func):
+        self.prefix_width = prefix_width
+        self.color_func = color_func
+
+    def present(self, container, line):
+        prefix = container.name_without_project.ljust(self.prefix_width)
+        return '{prefix} {line}'.format(
+            prefix=self.color_func(prefix + ' |'),
+            line=line)
+
+
+def build_log_presenters(service_names, monochrome):
+    """Return an iterable of functions.
+
+    Each function can be used to format the logs output of a container.
+    """
+    prefix_width = max_name_width(service_names)
+
+    def no_color(text):
+        return text
+
+    for color_func in cycle([no_color] if monochrome else colors.rainbow()):
+        yield LogPresenter(prefix_width, color_func)
+
+
+def max_name_width(service_names, max_index_width=3):
+    """Calculate the maximum width of container names so we can make the log
+    prefixes line up like so:
+
+    db_1  | Listening
+    web_1 | Listening
+    """
+    return max(len(name) for name in service_names) + max_index_width
+
+
 class LogPrinter(object):
     """Print logs from many containers to a single output stream."""
 
     def __init__(self,
                  containers,
+                 presenters,
+                 event_stream,
                  output=sys.stdout,
-                 monochrome=False,
                  cascade_stop=False,
                  log_args=None):
-        log_args = log_args or {}
         self.containers = containers
+        self.presenters = presenters
+        self.event_stream = event_stream
         self.output = utils.get_output_stream(output)
-        self.monochrome = monochrome
         self.cascade_stop = cascade_stop
-        self.log_args = log_args
+        self.log_args = log_args or {}
 
     def run(self):
         if not self.containers:
             return
 
-        prefix_width = max_name_width(self.containers)
-        generators = list(self._make_log_generators(self.monochrome, prefix_width))
-        for line in Multiplexer(generators, cascade_stop=self.cascade_stop).loop():
+        queue = Queue()
+        thread_args = queue, self.log_args
+        thread_map = build_thread_map(self.containers, self.presenters, thread_args)
+        start_producer_thread((
+            thread_map,
+            self.event_stream,
+            self.presenters,
+            thread_args))
+
+        for line in consume_queue(queue, self.cascade_stop):
+            remove_stopped_threads(thread_map)
+
+            if not line:
+                if not thread_map:
+                    # There are no running containers left to tail, so exit
+                    return
+                # We got an empty line because of a timeout, but there are still
+                # active containers to tail, so continue
+                continue
+
             self.output.write(line)
             self.output.flush()
 
-    def _make_log_generators(self, monochrome, prefix_width):
-        def no_color(text):
-            return text
 
-        if monochrome:
-            color_funcs = cycle([no_color])
-        else:
-            color_funcs = cycle(colors.rainbow())
-
-        for color_func, container in zip(color_funcs, self.containers):
-            generator_func = get_log_generator(container)
-            prefix = color_func(build_log_prefix(container, prefix_width))
-            yield generator_func(container, prefix, color_func, self.log_args)
+def remove_stopped_threads(thread_map):
+    for container_id, tailer_thread in list(thread_map.items()):
+        if not tailer_thread.is_alive():
+            thread_map.pop(container_id, None)
 
 
-def build_log_prefix(container, prefix_width):
-    return container.name_without_project.ljust(prefix_width) + ' | '
+def build_thread(container, presenter, queue, log_args):
+    tailer = Thread(
+        target=tail_container_logs,
+        args=(container, presenter, queue, log_args))
+    tailer.daemon = True
+    tailer.start()
+    return tailer
 
 
-def max_name_width(containers):
-    """Calculate the maximum width of container names so we can make the log
-    prefixes line up like so:
+def build_thread_map(initial_containers, presenters, thread_args):
+    return {
+        container.id: build_thread(container, next(presenters), *thread_args)
+        for container in initial_containers
+    }
 
-    db_1  | Listening
-    web_1 | Listening
-    """
-    return max(len(container.name_without_project) for container in containers)
+
+class QueueItem(namedtuple('_QueueItem', 'item is_stop exc')):
+
+    @classmethod
+    def new(cls, item):
+        return cls(item, None, None)
+
+    @classmethod
+    def exception(cls, exc):
+        return cls(None, None, exc)
+
+    @classmethod
+    def stop(cls):
+        return cls(None, True, None)
+
+
+def tail_container_logs(container, presenter, queue, log_args):
+    generator = get_log_generator(container)
+
+    try:
+        for item in generator(container, log_args):
+            queue.put(QueueItem.new(presenter.present(container, item)))
+    except Exception as e:
+        queue.put(QueueItem.exception(e))
+        return
+
+    if log_args.get('follow'):
+        queue.put(QueueItem.new(presenter.color_func(wait_on_exit(container))))
+    queue.put(QueueItem.stop())
 
 
 def get_log_generator(container):
@@ -71,32 +156,75 @@ def get_log_generator(container):
     return build_no_log_generator
 
 
-def build_no_log_generator(container, prefix, color_func, log_args):
+def build_no_log_generator(container, log_args):
     """Return a generator that prints a warning about logs and waits for
     container to exit.
     """
-    yield "{} WARNING: no logs are available with the '{}' log driver\n".format(
-        prefix,
+    yield "WARNING: no logs are available with the '{}' log driver\n".format(
         container.log_driver)
-    if log_args.get('follow'):
-        yield color_func(wait_on_exit(container))
 
 
-def build_log_generator(container, prefix, color_func, log_args):
+def build_log_generator(container, log_args):
     # if the container doesn't have a log_stream we need to attach to container
     # before log printer starts running
     if container.log_stream is None:
         stream = container.logs(stdout=True, stderr=True, stream=True, **log_args)
-        line_generator = split_buffer(stream)
     else:
-        line_generator = split_buffer(container.log_stream)
+        stream = container.log_stream
 
-    for line in line_generator:
-        yield prefix + line
-    if log_args.get('follow'):
-        yield color_func(wait_on_exit(container))
+    return split_buffer(stream)
 
 
 def wait_on_exit(container):
     exit_code = container.wait()
     return "%s exited with code %s\n" % (container.name, exit_code)
+
+
+def start_producer_thread(thread_args):
+    producer = Thread(target=watch_events, args=thread_args)
+    producer.daemon = True
+    producer.start()
+
+
+def watch_events(thread_map, event_stream, presenters, thread_args):
+    for event in event_stream:
+        if event['action'] == 'stop':
+            thread_map.pop(event['id'], None)
+
+        if event['action'] != 'start':
+            continue
+
+        if event['id'] in thread_map:
+            if thread_map[event['id']].is_alive():
+                continue
+            # Container was stopped and started, we need a new thread
+            thread_map.pop(event['id'], None)
+
+        thread_map[event['id']] = build_thread(
+            event['container'],
+            next(presenters),
+            *thread_args)
+
+
+def consume_queue(queue, cascade_stop):
+    """Consume the queue by reading lines off of it and yielding them."""
+    while True:
+        try:
+            item = queue.get(timeout=0.1)
+        except Empty:
+            yield None
+            continue
+        # See https://github.com/docker/compose/issues/189
+        except thread.error:
+            raise ShutdownException()
+
+        if item.exc:
+            raise item.exc
+
+        if item.is_stop:
+            if cascade_stop:
+                raise StopIteration
+            else:
+                continue
+
+        yield item.item

compose/cli/main.py

@@ -35,6 +35,7 @@ from .docopt_command import NoSuchCommand
 from .errors import UserError
 from .formatter import ConsoleWarningFormatter
 from .formatter import Formatter
+from .log_printer import build_log_presenters
 from .log_printer import LogPrinter
 from .utils import get_version_info
 from .utils import yesno
@@ -279,6 +280,7 @@ class TopLevelCommand(object):
 
         def json_format_event(event):
             event['time'] = event['time'].isoformat()
+            event.pop('container')
             return json.dumps(event)
 
         for event in self.project.events():
@@ -376,7 +378,6 @@ class TopLevelCommand(object):
         """
         containers = self.project.containers(service_names=options['SERVICE'], stopped=True)
 
-        monochrome = options['--no-color']
         tail = options['--tail']
         if tail is not None:
             if tail.isdigit():
@@ -389,7 +390,11 @@ class TopLevelCommand(object):
             'timestamps': options['--timestamps']
         }
         print("Attaching to", list_containers(containers))
-        LogPrinter(containers, monochrome=monochrome, log_args=log_args).run()
+        log_printer_from_project(
+            self.project,
+            containers,
+            options['--no-color'],
+            log_args).run()
 
     def pause(self, options):
         """
@@ -671,7 +676,6 @@ class TopLevelCommand(object):
             --remove-orphans           Remove containers for services not
                                        defined in the Compose file
         """
-        monochrome = options['--no-color']
         start_deps = not options['--no-deps']
         cascade_stop = options['--abort-on-container-exit']
         service_names = options['SERVICE']
@@ -694,8 +698,14 @@ class TopLevelCommand(object):
 
             if detached:
                 return
-            log_args = {'follow': True}
-            log_printer = build_log_printer(to_attach, service_names, monochrome, cascade_stop, log_args)
+
+            log_printer = log_printer_from_project(
+                self.project,
+                filter_containers_to_service_names(to_attach, service_names),
+                options['--no-color'],
+                {'follow': True},
+                cascade_stop,
+                event_stream=self.project.events(service_names=service_names))
             print("Attaching to", list_containers(log_printer.containers))
             log_printer.run()
 
@@ -842,13 +852,30 @@ def run_one_off_container(container_options, project, service, options):
     sys.exit(exit_code)
 
 
-def build_log_printer(containers, service_names, monochrome, cascade_stop, log_args):
-    if service_names:
-        containers = [
-            container
-            for container in containers if container.service in service_names
-        ]
-    return LogPrinter(containers, monochrome=monochrome, cascade_stop=cascade_stop, log_args=log_args)
+def log_printer_from_project(
+    project,
+    containers,
+    monochrome,
+    log_args,
+    cascade_stop=False,
+    event_stream=None,
+):
+    return LogPrinter(
+        containers,
+        build_log_presenters(project.service_names, monochrome),
+        event_stream or project.events(),
+        cascade_stop=cascade_stop,
+        log_args=log_args)
+
+
+def filter_containers_to_service_names(containers, service_names):
+    if not service_names:
+        return containers
+
+    return [
+        container
+        for container in containers if container.service in service_names
+    ]
 
 
 @contextlib.contextmanager

compose/cli/multiplexer.py

@@ -1,66 +0,0 @@
-from __future__ import absolute_import
-from __future__ import unicode_literals
-
-from threading import Thread
-
-from six.moves import _thread as thread
-
-try:
-    from Queue import Queue, Empty
-except ImportError:
-    from queue import Queue, Empty  # Python 3.x
-
-from compose.cli.signals import ShutdownException
-
-STOP = object()
-
-
-class Multiplexer(object):
-    """
-    Create a single iterator from several iterators by running all of them in
-    parallel and yielding results as they come in.
-    """
-
-    def __init__(self, iterators, cascade_stop=False):
-        self.iterators = iterators
-        self.cascade_stop = cascade_stop
-        self._num_running = len(iterators)
-        self.queue = Queue()
-
-    def loop(self):
-        self._init_readers()
-
-        while self._num_running > 0:
-            try:
-                item, exception = self.queue.get(timeout=0.1)
-
-                if exception:
-                    raise exception
-
-                if item is STOP:
-                    if self.cascade_stop is True:
-                        break
-                    else:
-                        self._num_running -= 1
-                else:
-                    yield item
-            except Empty:
-                pass
-            # See https://github.com/docker/compose/issues/189
-            except thread.error:
-                raise ShutdownException()
-
-    def _init_readers(self):
-        for iterator in self.iterators:
-            t = Thread(target=_enqueue_output, args=(iterator, self.queue))
-            t.daemon = True
-            t.start()
-
-
-def _enqueue_output(iterator, queue):
-    try:
-        for item in iterator:
-            queue.put((item, None))
-        queue.put((STOP, None))
-    except Exception as e:
-        queue.put((None, e))

compose/project.py

@@ -297,7 +297,7 @@ class Project(object):
                 detached=True,
                 start=False)
 
-    def events(self):
+    def events(self, service_names=None):
         def build_container_event(event, container):
             time = datetime.datetime.fromtimestamp(event['time'])
             time = time.replace(
@@ -311,10 +311,11 @@ class Project(object):
                 'attributes': {
                     'name': container.name,
                     'image': event['from'],
-                }
+                },
+                'container': container,
             }
 
-        service_names = set(self.service_names)
+        service_names = set(service_names or self.service_names)
         for event in self.client.events(
             filters={'label': self.labels()},
             decode=True
@@ -325,7 +326,11 @@ class Project(object):
                 continue
 
             # TODO: get labels from the API v1.22 , see github issue 2618
-            container = Container.from_id(self.client, event['id'])
+            try:
+                # this can fail if the conatiner has been removed
+                container = Container.from_id(self.client, event['id'])
+            except APIError:
+                continue
             if container.service not in service_names:
                 continue
             yield build_container_event(event, container)