m-labs · Oct 16, 2015 · Oct 16, 2015 · Oct 16, 2015 · Oct 16, 2015
Showing with 102 additions and 28 deletions.

+56 −24 artiq/frontend/artiq_ctlmgr.py

+1 −1 artiq/master/log.py

+45 −3 artiq/protocols/logging.py
diff --git a/artiq/frontend/artiq_ctlmgr.py b/artiq/frontend/artiq_ctlmgr.py
@@ -10,34 +10,13 @@
 
 from artiq.protocols.sync_struct import Subscriber
 from artiq.protocols.pc_rpc import AsyncioClient, Server
-from artiq.tools import verbosity_args, init_logger
+from artiq.protocols.logging import LogForwarder
 from artiq.tools import TaskObject, Condition
 
 
 logger = logging.getLogger(__name__)
 
 
-def get_argparser():
-    parser = argparse.ArgumentParser(description="ARTIQ controller manager")
-    verbosity_args(parser)
-    parser.add_argument(
-        "-s", "--server", default="::1",
-        help="hostname or IP of the master to connect to")
-    parser.add_argument(
-        "--port", default=3250, type=int,
-        help="TCP port to use to connect to the master")
-    parser.add_argument(
-        "--retry-master", default=5.0, type=float,
-        help="retry timer for reconnecting to master")
-    parser.add_argument(
-        "--bind", default="::1",
-        help="hostname or IP address to bind to")
-    parser.add_argument(
-        "--bind-port", default=3249, type=int,
-        help="TCP port to listen to for control (default: %(default)d)")
-    return parser
-
-
 class Controller:
     def __init__(self, name, ddb_entry):
         self.name = name
@@ -252,9 +231,54 @@ def retry_now(self, k):
         self.controller_db.current_controllers.active[k].retry_now.notify()
 
 
+def get_argparser():
+    parser = argparse.ArgumentParser(description="ARTIQ controller manager")
+
+    group = parser.add_argument_group("verbosity")
+    group.add_argument("-v", "--verbose", default=0, action="count",
+                       help="increase logging level")
+    group.add_argument("-q", "--quiet", default=0, action="count",
+                       help="decrease logging level")
+
+    parser.add_argument(
+        "-s", "--server", default="::1",
+        help="hostname or IP of the master to connect to")
+    parser.add_argument(
+        "--port-notify", default=3250, type=int,
+        help="TCP port to connect to for notifications")
+    parser.add_argument(
+        "--port-logging", default=1066, type=int,
+        help="TCP port to connect to for logging")
+    parser.add_argument(
+        "--retry-master", default=5.0, type=float,
+        help="retry timer for reconnecting to master")
+    parser.add_argument(
+        "--bind", default="::1",
+        help="hostname or IP address to bind to")
+    parser.add_argument(
+        "--bind-port", default=3249, type=int,
+        help="TCP port to listen to for control (default: %(default)d)")
+    return parser
+
+
+class SourceAdder:
+    def filter(self, record):
+        if not hasattr(record, "source"):
+            record.source = "ctlmgr"
+        return True
+
+
 def main():
     args = get_argparser().parse_args()
-    init_logger(args)
+
+    root_logger = logging.getLogger()
+    root_logger.setLevel(logging.WARNING + args.quiet*10 - args.verbose*10)
+    source_adder = SourceAdder()
+    console_handler = logging.StreamHandler()
+    console_handler.setFormatter(logging.Formatter(
+        "%(levelname)s:%(source)s:%(name)s:%(message)s"))
+    console_handler.addFilter(source_adder)
+    root_logger.addHandler(console_handler)
 
     if os.name == "nt":
         loop = asyncio.ProactorEventLoop()
@@ -263,7 +287,15 @@ def main():
         loop = asyncio.get_event_loop()
     atexit.register(lambda: loop.close())
 
-    ctlmgr = ControllerManager(args.server, args.port, args.retry_master)
+    logfwd = LogForwarder(args.server, args.port_logging,
+                          args.retry_master)
+    logfwd.addFilter(source_adder)
+    root_logger.addHandler(logfwd)
+    logfwd.start()
+    atexit.register(lambda: loop.run_until_complete(logfwd.stop()))
+
+    ctlmgr = ControllerManager(args.server, args.port_notify,
+                               args.retry_master)
     ctlmgr.start()
     atexit.register(lambda: loop.run_until_complete(ctlmgr.stop()))
 

diff --git a/artiq/master/log.py b/artiq/master/log.py
@@ -20,6 +20,7 @@ class LogBufferHandler(logging.Handler):
     def __init__(self, log_buffer, *args, **kwargs):
         logging.Handler.__init__(self, *args, **kwargs)
         self.log_buffer = log_buffer
+        self.setFormatter(logging.Formatter("%(name)s:%(message)s"))
 
     def emit(self, record):
         message = self.format(record)
@@ -87,7 +88,6 @@ def init_log(args):
 
     log_buffer = LogBuffer(1000)
     buffer_handler = LogBufferHandler(log_buffer)
-    buffer_handler.setFormatter(logging.Formatter("%(name)s:%(message)s"))
     handlers.append(buffer_handler)
 
     for handler in handlers:

diff --git a/artiq/protocols/logging.py b/artiq/protocols/logging.py
@@ -2,6 +2,7 @@
 import logging
 
 from artiq.protocols.asyncio_server import AsyncioServer
+from artiq.tools import TaskObject
 
 
 _fwd_logger = logging.getLogger("fwd")
@@ -38,6 +39,11 @@ def parse_log_message(msg):
 
 
 class Server(AsyncioServer):
+    """Remote logging TCP server.
+
+    Takes one log entry per line, in the format:
+        source:levelno:name:message
+    """
     async def _handle_connection_cr(self, reader, writer):
         try:
             line = await reader.readline()
@@ -56,12 +62,48 @@ async def _handle_connection_cr(self, reader, writer):
                 linesplit = line.split(":", 4)
                 if len(linesplit) != 4:
                     return
-                source, levelname, name, message = linesplit
+                source, level, name, message = linesplit
                 try:
-                    level = _name_to_level[levelname]
-                except KeyError:
+                    level = int(level)
+                except:
                     return
                 log_with_name(name, level, message,
                               extra={"source": source})
         finally:
             writer.close()
+
+
+class LogForwarder(logging.Handler, TaskObject):
+    def __init__(self, host, port, reconnect_timer=5.0, queue_size=1000,
+                 **kwargs):
+        logging.Handler.__init__(self, **kwargs)
+        self.host = host
+        self.port = port
+        self.setFormatter(logging.Formatter(
+            "%(source)s:%(levelno)d:%(name)s:%(message)s"))
+        self._queue = asyncio.Queue(queue_size)
+        self.reconnect_timer = reconnect_timer
+
+    def emit(self, record):
+        message = self.format(record)
+        try:
+            self._queue.put_nowait(message)
+        except asyncio.QueueFull:
+            pass
+
+    async def _do(self):
+        while True:
+            try:
+                reader, writer = await asyncio.open_connection(self.host,
+                                                               self.port)
+                writer.write(_init_string)
+                while True:
+                    message = await self._queue.get() + "\n"
+                    writer.write(message.encode())
+                    await writer.drain()
+            except asyncio.CancelledError:
+                return
+            except:
+                await asyncio.sleep(self.reconnect_timer)
+            finally:
+                writer.close()