m-labs · Jan 26, 2016 · Jan 26, 2016 · Jan 26, 2016 · Jan 26, 2016
Showing with 84 additions and 41 deletions.

+2 −2 artiq/master/worker_impl.py

+59 −38 artiq/protocols/logging.py

+23 −1 artiq/tools.py
diff --git a/artiq/master/worker_impl.py b/artiq/master/worker_impl.py
@@ -7,7 +7,7 @@
 
 import artiq
 from artiq.protocols import pipe_ipc, pyon
-from artiq.tools import file_import
+from artiq.tools import multiline_log_config, file_import
 from artiq.master.worker_db import DeviceManager, DatasetManager, get_hdf5_output
 from artiq.language.environment import is_experiment
 from artiq.language.core import set_watchdog_factory, TerminationRequested
@@ -187,7 +187,7 @@ def render_diagnostic(self, diagnostic):
 def main():
     global ipc
 
-    logging.basicConfig(level=int(sys.argv[2]))
+    multiline_log_config(level=int(sys.argv[2]))
     ipc = pipe_ipc.ChildComm(sys.argv[1])
 
     start_time = None

diff --git a/artiq/protocols/logging.py b/artiq/protocols/logging.py
@@ -1,8 +1,9 @@
 import asyncio
 import logging
+import re
 
 from artiq.protocols.asyncio_server import AsyncioServer
-from artiq.tools import TaskObject
+from artiq.tools import TaskObject, MultilineFormatter
 
 
 logger = logging.getLogger(__name__)
@@ -25,25 +26,51 @@ def log_with_name(name, *args, **kwargs):
 
 
 def parse_log_message(msg):
-    for name, level in _name_to_level.items():
-        if msg.startswith(name + ":"):
-            remainder = msg[len(name) + 1:]
-            try:
-                idx = remainder.index(":")
-            except:
-                continue
-            return level, remainder[:idx], remainder[idx+1:]
-    return logging.INFO, "print", msg
+    lr = "|".join(_name_to_level.keys())
+    m = re.fullmatch('('+lr+')(<\d+>)?:([^:]*):(.*)', msg)
+    if m is None:
+        return 0, logging.INFO, "print", msg
+    level = _name_to_level[m.group(1)]
+    if m.group(2):
+        multiline = int(m.group(2)[1:-1]) - 1
+    else:
+        multiline = 0
+    name = m.group(3)
+    message = m.group(4)
+    return multiline, level, name, message
 
 
 class LogParser:
     def __init__(self, source_cb):
         self.source_cb = source_cb
+        self.multiline_count = 0
+        self.multiline_level = None
+        self.multiline_name = None
+        self.multiline_message = None
 
     def line_input(self, msg):
-        level, name, message = parse_log_message(msg)
-        log_with_name(name, level, message,
-                      extra={"source": self.source_cb()})
+        if self.multiline_count:
+            self.multiline_message += "\n" + msg
+            self.multiline_count -= 1
+            if not self.multiline_count:
+                log_with_name(
+                    self.multiline_name,
+                    self.multiline_level,
+                    self.multiline_message,
+                    extra={"source": self.source_cb()})
+                self.multiline_level = None
+                self.multiline_name = None
+                self.multiline_message = None
+        else:
+            multiline, level, name, message = parse_log_message(msg)
+            if multiline:
+                self.multiline_count = multiline
+                self.multiline_level = level
+                self.multiline_name = name
+                self.multiline_message = message
+            else:
+                log_with_name(name, level, message,
+                              extra={"source": self.source_cb()})
 
     async def stream_task(self, stream):
         while True:
@@ -65,15 +92,20 @@ async def stream_task(self, stream):
 class Server(AsyncioServer):
     """Remote logging TCP server.
 
-    Takes one log entry per line, in the format:
-        source:levelno:name:message
+    Log entries are in the format:
+        source:levelno<total_lines>:name:message
+        continuation...
+        ...continuation
     """
     async def _handle_connection_cr(self, reader, writer):
         try:
             line = await reader.readline()
             if line != _init_string:
                 return
 
+            source = None
+            parser = LogParser(lambda: source)
+
             while True:
                 line = await reader.readline()
                 if not line:
@@ -83,20 +115,16 @@ async def _handle_connection_cr(self, reader, writer):
                 except:
                     return
                 line = line[:-1]
-                linesplit = line.split(":", 3)
-                if len(linesplit) != 4:
-                    logger.warning("received improperly formatted message, "
-                                   "dropping connection")
-                    return
-                source, level, name, message = linesplit
-                try:
-                    level = int(level)
-                except:
-                    logger.warning("received improperly formatted level, "
-                                   "dropping connection")
-                    return
-                log_with_name(name, level, message,
-                              extra={"source": source})
+                if parser.multiline_count:
+                    parser.line_input(line)
+                else:
+                    linesplit = line.split(":", maxsplit=1)
+                    if len(linesplit) != 2:
+                        logger.warning("received improperly formatted message, "
+                                       "dropping connection")
+                        return
+                    source, remainder = linesplit
+                    parser.line_input(remainder)
         finally:
             writer.close()
 
@@ -123,19 +151,12 @@ def __init__(self, host, port, reconnect_timer=5.0, queue_size=1000,
         logging.Handler.__init__(self, **kwargs)
         self.host = host
         self.port = port
-        self.setFormatter(logging.Formatter(
-            "%(name)s:%(message)s"))
+        self.setFormatter(MultilineFormatter())
         self._queue = asyncio.Queue(queue_size)
         self.reconnect_timer = reconnect_timer
 
     def emit(self, record):
-        message = self.format(record)
-        for part in message.split("\n"):
-            part = "{}:{}:{}".format(record.source, record.levelno, part)
-            try:
-                self._queue.put_nowait(part)
-            except asyncio.QueueFull:
-                break
+        self._queue.put_nowait(record.source + ":" + self.format(record))
 
     async def _do(self):
         while True:

diff --git a/artiq/tools.py b/artiq/tools.py
@@ -133,8 +133,30 @@ def simple_network_args(parser, default_port):
             group.add_argument("--port-" + name, default=default, type=int,
                            help=h)
 
+class MultilineFormatter(logging.Formatter):
+    def __init__(self):
+        logging.Formatter.__init__(
+            self, "%(levelname)s:%(name)s:%(message)s")
+
+    def format(self, record):
+        r = logging.Formatter.format(self, record)
+        linebreaks = r.count("\n")
+        if linebreaks:
+            i = r.index(":")
+            r = r[:i] + "<" + str(linebreaks + 1) + ">" + r[i:]
+        return r
+
+
+def multiline_log_config(level):
+    root_logger = logging.getLogger()
+    root_logger.setLevel(level)
+    handler = logging.StreamHandler()
+    handler.setFormatter(MultilineFormatter())
+    root_logger.addHandler(handler)
+
+
 def init_logger(args):
-    logging.basicConfig(level=logging.WARNING + args.quiet*10 - args.verbose*10)
+    multiline_log_config(level=logging.WARNING + args.quiet*10 - args.verbose*10)
 
 
 def bind_address_from_args(args):