Review logging to keep all debugging information in RAM and flush only if useful

The main goal of this patch is to keep all DEBUG logs and packet logger enabled
without exploding disk usage.
This is done by keeping the last 16 MB (by default) of debugging information in
a RAM buffer, and to emit it to an SQLite DB upon RTMIN signal or in case of
warning and more severe log.

Implementation is also cleaned up for better integration within a framework
or if run standalone. NEO logger is now a direct child of root handler.
Only warnings and more severe logs are forwarded to root handler.

A new script 'neolog' is added to pretty-print the contents of the SQLite log.

In unit tests, logging events are not buffered but emitted immediately.
When a test passes, payloads of all exchanged packets are discarded to reduce
disk usage on test bots.

This slows down performance tests by about 15 % because even if nothing is
written to disk, debug and packet log records are now always rendered.

neo/client/Storage.py

@@ -20,7 +20,6 @@ import ZODB.interfaces
 
 import neo.lib
 from functools import wraps
-from neo.lib import setupLog
 from neo.lib.util import add64
 from neo.lib.protocol import ZERO_TID
 from .app import Application
@@ -63,7 +62,7 @@ class Storage(BaseStorage.BaseStorage,
     )))
 
     def __init__(self, master_nodes, name, read_only=False,
-            compress=None, logfile=None, verbose=False, _app=None,
+            compress=None, logfile=None, _app=None,
             dynamic_master_list=None, **kw):
         """
         Do not pass those parameters (used internally):
@@ -71,7 +70,8 @@ class Storage(BaseStorage.BaseStorage,
         """
         if compress is None:
             compress = True
-        setupLog('CLIENT', filename=logfile, verbose=verbose)
+        if logfile:
+            neo.lib.logging.setup(logfile)
         BaseStorage.BaseStorage.__init__(self, 'NEOStorage(%s)' % (name, ))
         # Warning: _is_read_only is used in BaseStorage, do not rename it.
         self._is_read_only = read_only
@@ -84,8 +84,6 @@ class Storage(BaseStorage.BaseStorage,
         self._init_kw = {
             'read_only': read_only,
             'compress': compress,
-            'logfile': logfile,
-            'verbose': verbose,
             'dynamic_master_list': dynamic_master_list,
             '_app': _app,
         }

neo/client/component.xml

@@ -27,9 +27,9 @@
         and is still allowed on a read-only neostorage.
       </description>
     </key>
-    <key name="verbose" datatype="boolean">
+    <key name="logfile" datatype="path">
       <description>
-        Log debugging information
+        Log debugging information to specified SQLite DB.
       </description>
     </key>
     <key name="dynamic_master_list" datatype="path">

neo/lib/__init__.py

 #
-# Copyright (C) 2006-2011  Nexedi SA
+# Copyright (C) 2006-2012  Nexedi SA
 #
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU General Public License
@@ -14,49 +14,4 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-import logging as logging_std
-
-FMT = ('%(asctime)s %(levelname)-9s %(name)-10s'
-       ' [%(module)14s:%(lineno)3d] \n%(message)s')
-
-class Formatter(logging_std.Formatter):
-
-    def formatTime(self, record, datefmt=None):
-        return logging_std.Formatter.formatTime(self, record,
-           '%Y-%m-%d %H:%M:%S') + '.%04d' % (record.msecs * 10)
-
-    def format(self, record):
-        lines = iter(logging_std.Formatter.format(self, record).splitlines())
-        prefix = lines.next()
-        return '\n'.join(prefix + line for line in lines)
-
-def setupLog(name='NEO', filename=None, verbose=False):
-    global logging
-    if verbose:
-        level = logging_std.DEBUG
-    else:
-        level = logging_std.INFO
-    if logging is not None:
-        for handler in logging.handlers:
-            handler.acquire()
-            try:
-                handler.close()
-            finally:
-                handler.release()
-        del logging.manager.loggerDict[logging.name]
-    logging = logging_std.getLogger(name)
-    for handler in logging.handlers[:]:
-        logging.removeHandler(handler)
-    if filename is None:
-        handler = logging_std.StreamHandler()
-    else:
-        handler = logging_std.FileHandler(filename)
-    handler.setFormatter(Formatter(FMT))
-    logging.setLevel(level)
-    logging.addHandler(handler)
-    logging.propagate = 0
-
-# Create default logger
-logging = None
-setupLog()
-
+from .logger import logging

neo/lib/config.py

@@ -72,9 +72,6 @@ class ConfigurationManager(object):
         assert cluster != '', "Cluster name must be non-empty"
         return cluster
 
-    def getName(self):
-        return self.__get('name')
-
     def getReplicas(self):
         return int(self.__get('replicas'))
 

neo/lib/connection.py

@@ -23,7 +23,6 @@ from .connector import ConnectorException, ConnectorTryAgainException, \
         ConnectorInProgressException, ConnectorConnectionRefusedException, \
         ConnectorConnectionClosedException
 from .locking import RLock
-from .logger import PACKET_LOGGER
 from .profiling import profiler_decorator
 from .protocol import Errors, PacketMalformedError, Packets, ParserState
 from .util import dump, ReadBuffer
@@ -166,7 +165,7 @@ class HandlerSwitcher(object):
     @profiler_decorator
     def _handle(self, connection, packet):
         assert len(self._pending) == 1 or self._pending[0][0]
-        PACKET_LOGGER.dispatch(connection, packet, False)
+        neo.lib.logging.packet(connection, packet, False)
         if connection.isClosed() and packet.ignoreOnClosedConnection():
             neo.lib.logging.debug('Ignoring packet %r on closed connection %r',
                 packet, connection)
@@ -638,7 +637,7 @@ class Connection(BaseConnection):
         if was_empty:
             # enable polling for writing.
             self.em.addWriter(self)
-        PACKET_LOGGER.dispatch(self, packet, True)
+        neo.lib.logging.packet(self, packet, True)
 
     @not_closed
     def notify(self, packet):

neo/lib/debug.py

 #
-# Copyright (C) 2010  Nexedi SA
+# Copyright (C) 2010-2012  Nexedi SA
 #
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU General Public License
@@ -22,34 +22,23 @@ import sys
 from functools import wraps
 import neo
 
-# WARNING: This module should only be used for live application debugging.
-# It - by purpose - allows code injection in a running neo process.
-# You don't want to enable it in a production environment. Really.
-ENABLED = False
+# kill -RTMIN+1 <pid>
+#   Dump information to logs.
+# kill -RTMIN+2 <pid>
+#   Loads (or reloads) neo.debug module.
+#   The content is up to you (it's only imported). It can be a breakpoint.
 
-# How to include in python code:
-#   from neo.debug import register
-#   register()
-#
-# How to trigger it:
-#   Kill python process with:
-#     SIGUSR1:
-#       Loads (or reloads) neo.debug module.
-#       The content is up to you (it's only imported).
-#     SIGUSR2:
-#       Triggers a pdb prompt on process' controlling TTY.
-
-def decorate(func):
-    def decorator(sig, frame):
+def safe_handler(func):
+    def wrapper(sig, frame):
         try:
             func(sig, frame)
         except:
             # Prevent exception from exiting signal handler, so mistakes in
             # "debug" module don't kill process.
             traceback.print_exc()
-    return wraps(func)(decorator)
+    return wraps(func)(wrapper)
 
-@decorate
+@safe_handler
 def debugHandler(sig, frame):
     file, filename, (suffix, mode, type) = imp.find_module('debug',
         neo.__path__)
@@ -95,24 +84,10 @@ def winpdb(depth=0):
         finally:
             os.abort()
 
-@decorate
-def pdbHandler(sig, frame):
-    try:
-        winpdb(2) # depth is 2, because of the decorator
-    except ImportError:
-        global _debugger
-        if _debugger is None:
-            _debugger = getPdb()
-        debugger.set_trace(frame)
-
 def register(on_log=None):
-    if ENABLED:
-        signal.signal(signal.SIGUSR1, debugHandler)
-        signal.signal(signal.SIGUSR2, pdbHandler)
-        if on_log is not None:
-            # use 'kill -RTMIN <pid>
-            @decorate
-            def on_log_signal(signum, signal):
-                on_log()
-            signal.signal(signal.SIGRTMIN, on_log_signal)
-
+    if on_log is not None:
+        @safe_handler
+        def on_log_signal(signum, signal):
+            on_log()
+        signal.signal(signal.SIGRTMIN+1, on_log_signal)
+    signal.signal(signal.SIGRTMIN+2, debugHandler)

neo/lib/logger.py

 #
-# Copyright (C) 2006-2011  Nexedi SA
+# Copyright (C) 2006-2012  Nexedi SA
 #
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU General Public License
@@ -14,57 +14,203 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from base64 import b64encode
-import neo
-from .protocol import PacketMalformedError
-from .util import dump
-from .handler import EventHandler
-from .profiling import profiler_decorator
+# WARNING: Log rotating should not be implemented here.
+#          SQLite does not access database only by file descriptor,
+#          and an OperationalError exception would be raised if a log is emitted
+#          between a rename and a reopen.
+#          Fortunately, SQLite allow multiple process to access the same DB,
+#          so an external tool should be able to dump and empty tables.
 
-LOGGER_ENABLED = False
+from binascii import b2a_hex
+from collections import deque
+from functools import wraps
+from logging import getLogger, Formatter, Logger, LogRecord, StreamHandler, \
+    DEBUG, WARNING
+from time import time
+from traceback import format_exception
+import neo, os, signal, sqlite3, threading
 
-class PacketLogger(object):
-    """ Logger at packet level (for debugging purpose) """
+# Stats for storage node of matrix test (py2.7:SQLite)
+RECORD_SIZE = ( 234360832 # extra memory used
+              - 16777264  # sum of raw data ('msg' attribute)
+              ) // 187509 # number of records
+
+FMT = ('%(asctime)s %(levelname)-9s %(name)-10s'
+       ' [%(module)14s:%(lineno)3d] \n%(message)s')
+
+class _Formatter(Formatter):
+
+    def formatTime(self, record, datefmt=None):
+        return Formatter.formatTime(self, record,
+           '%Y-%m-%d %H:%M:%S') + '.%04d' % (record.msecs * 10)
+
+    def format(self, record):
+        lines = iter(Formatter.format(self, record).splitlines())
+        prefix = lines.next()
+        return '\n'.join(prefix + line for line in lines)
+
+
+class PacketRecord(object):
+
+    args = None
+    levelno = DEBUG
+    __init__ = property(lambda self: self.__dict__.update)
+
+
+class logging(Logger):
+
+    default_root_handler = StreamHandler()
+    default_root_handler.setFormatter(_Formatter(FMT))
 
     def __init__(self):
-        self.enable(LOGGER_ENABLED)
-
-    def enable(self, enabled):
-        self.dispatch = enabled and self._dispatch or (lambda *args, **kw: None)
-
-    def _dispatch(self, conn, packet, outgoing):
-        """This is a helper method to handle various packet types."""
-        # default log message
-        uuid = dump(conn.getUUID())
-        ip, port = conn.getAddress()
-        packet_name = packet.__class__.__name__
-        neo.lib.logging.debug('#0x%04x %-30s %s %s (%s:%d) %s', packet.getId(),
-                packet_name, outgoing and '>' or '<', uuid, ip, port,
-                b64encode(packet._body[:96]))
-        # look for custom packet logger
-        logger = getattr(self, packet.handler_method_name, None)
-        if logger is None:
+        Logger.__init__(self, None)
+        self.parent = root = getLogger()
+        if not root.handlers:
+            root.addHandler(self.default_root_handler)
+        self.db = None
+        self._record_queue = deque()
+        self._record_size = 0
+        self._async = set()
+        l = threading.Lock()
+        self._acquire = l.acquire
+        release = l.release
+        def _release():
+            try:
+                while self._async:
+                    self._async.pop()(self)
+            finally:
+                release()
+        self._release = _release
+        self.backlog()
+
+    def __async(wrapped):
+        def wrapper(self):
+            self._async.add(wrapped)
+            if self._acquire(0):
+                self._release()
+        return wraps(wrapped)(wrapper)
+
+    @__async
+    def flush(self):
+        if self.db is None:
             return
-        # enhanced log
+        self.db.execute("BEGIN")
+        for r in self._record_queue:
+            self._emit(r)
+        self.db.commit()
+        self._record_queue.clear()
+        self._record_size = 0
+
+    def backlog(self, max_size=1<<24):
+        self._acquire()
         try:
-            args = packet.decode() or ()
-        except PacketMalformedError:
-            neo.lib.logging.warning("Can't decode packet for logging")
-            return
-        log_message = logger(conn, *args)
-        if log_message is not None:
-            neo.lib.logging.debug('#0x%04x %s', packet.getId(), log_message)
+            self._max_size = max_size
+            if max_size is None:
+                self.flush()
+            else:
+                q = self._record_queue
+                while max_size < self._record_size:
+                    self._record_size -= RECORD_SIZE + len(q.popleft().msg)
+        finally:
+            self._release()
+
+    def setup(self, filename=None, reset=False):
+        self._acquire()
+        try:
+            if self.db is not None:
+                self.db.close()
+                if not filename:
+                    self.db = None
+                    self._record_queue.clear()
+                    self._record_size = 0
+                    return
+            if filename:
+                self.db = sqlite3.connect(filename, isolation_level=None,
+                                                    check_same_thread=False)
+                q = self.db.execute
+                if reset:
+                    for t in 'log', 'packet':
+                        q('DROP TABLE IF EXISTS ' + t)
+                q("""CREATE TABLE IF NOT EXISTS log (
+                        date REAL NOT NULL,
+                        name TEXT,
+                        level INTEGER NOT NULL,
+                        pathname TEXT,
+                        lineno INTEGER,
+                        msg TEXT)
+                  """)
+                q("""CREATE INDEX IF NOT EXISTS _log_i1 ON log(date)""")
+                q("""CREATE TABLE IF NOT EXISTS packet (
+                        date REAL NOT NULL,
+                        name TEXT,
+                        msg_id INTEGER NOT NULL,
+                        code INTEGER NOT NULL,
+                        peer TEXT NOT NULL,
+                        body BLOB)
+                  """)
+                q("""CREATE INDEX IF NOT EXISTS _packet_i1 ON packet(date)""")
+        finally:
+            self._release()
+    __del__ = setup
 
-    def error(self, conn, code, message):
-        return "%s (%s)" % (code, message)
+    def isEnabledFor(self, level):
+        return True
 
-    def notifyNodeInformation(self, conn, node_list):
-        for node_type, address, uuid, state in node_list:
-            if address is not None:
-                address = '%s:%d' % address
+    def _emit(self, r):
+        if type(r) is PacketRecord:
+            ip, port = r.addr
+            peer = '%s %s (%s:%u)' % ('>' if r.outgoing else '<',
+                                      r.uuid and b2a_hex(r.uuid), ip, port)
+            self.db.execute("INSERT INTO packet VALUES (?,?,?,?,?,?)",
+                (r.created, r._name, r.msg_id, r.code, peer, buffer(r.msg)))
+        else:
+            pathname = os.path.relpath(r.pathname, *neo.__path__)
+            self.db.execute("INSERT INTO log VALUES (?,?,?,?,?,?)",
+                (r.created, r._name, r.levelno, pathname, r.lineno, r.msg))
+
+    def _queue(self, record):
+        record._name = self.name and str(self.name)
+        self._acquire()
+        try:
+            if self._max_size is None:
+                self._emit(record)
             else:
-                address = '?'
-            node = (dump(uuid), node_type, address, state)
-            neo.lib.logging.debug(' ! %s | %8s | %22s | %s' % node)
+                self._record_size += RECORD_SIZE + len(record.msg)
+                q = self._record_queue
+                q.append(record)
+                if record.levelno < WARNING:
+                    while self._max_size < self._record_size:
+                        self._record_size -= RECORD_SIZE + len(q.popleft().msg)
+                else:
+                    self.flush()
+        finally:
+            self._release()
+
+    def callHandlers(self, record):
+        if self.db is not None:
+            record.msg = record.getMessage()
+            record.args = None
+            if record.exc_info:
+                record.msg += '\n' + ''.join(
+                    format_exception(*record.exc_info)).strip()
+                record.exc_info = None
+            self._queue(record)
+        if Logger.isEnabledFor(self, record.levelno):
+            record.name = self.name or 'NEO'
+            self.parent.callHandlers(record)
+
+    def packet(self, connection, packet, outgoing):
+        if self.db is not None:
+            ip, port = connection.getAddress()
+            self._queue(PacketRecord(
+                created=time(),
+                msg_id=packet._id,
+                code=packet._code,
+                outgoing=outgoing,
+                uuid=connection.getUUID(),
+                addr=connection.getAddress(),
+                msg=packet._body))
+
 
-PACKET_LOGGER = PacketLogger()
+logging = logging()
+signal.signal(signal.SIGRTMIN, lambda signum, frame: logging.flush())

neo/scripts/neoadmin.py

@@ -2,7 +2,7 @@
 #
 # neoadmin - run an administrator  node of NEO
 #
-# Copyright (C) 2009  Nexedi SA
+# Copyright (C) 2009-2012  Nexedi SA
 #
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU General Public License
@@ -18,26 +18,22 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from optparse import OptionParser
-from neo.lib import setupLog
+from neo.lib import logging
 from neo.lib.config import ConfigurationManager
 
 parser = OptionParser()
 parser.add_option('-u', '--uuid', help='specify an UUID to use for this ' \
                   'process')
-parser.add_option('-v', '--verbose', action = 'store_true',
-                  help = 'print verbose messages')
 parser.add_option('-f', '--file', help = 'specify a configuration file')
 parser.add_option('-s', '--section', help = 'specify a configuration section')
 parser.add_option('-l', '--logfile', help = 'specify a logging file')
 parser.add_option('-c', '--cluster', help = 'the cluster name')
 parser.add_option('-m', '--masters', help = 'master node list')
 parser.add_option('-b', '--bind', help = 'the local address to bind to')
-parser.add_option('-n', '--name', help = 'the node name (improve logging)')
 parser.add_option('-D', '--dynamic-master-list', help='path of the file '
     'containing dynamic master node list')
 
 defaults = dict(
-    name = 'admin',
     bind = '127.0.0.1:9999',
     masters = '127.0.0.1:10000',
 )
@@ -47,7 +43,6 @@ def main(args=None):
     (options, args) = parser.parse_args(args=args)
     arguments = dict(
         uuid = options.uuid,
-        name = options.name or options.section,
         cluster = options.cluster,
         masters = options.masters,
         bind = options.bind,
@@ -61,7 +56,7 @@ def main(args=None):
     )
 
     # setup custom logging
-    setupLog(config.getName().upper(), options.logfile or None, options.verbose)
+    logging.setup(options.logfile)
 
     # and then, load and run the application
     from neo.admin.app import Application

neo/scripts/neoctl.py

@@ -2,7 +2,7 @@
 #
 # neoadmin - run an administrator node of NEO
 #
-# Copyright (C) 2009  Nexedi SA
+# Copyright (C) 2009-2012  Nexedi SA
 #
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU General Public License
@@ -19,15 +19,14 @@
 
 import sys
 from optparse import OptionParser
-from neo.lib import setupLog
+from neo.lib import logging
 from neo.lib.util import parseNodeAddress
 
 parser = OptionParser()
-parser.add_option('-v', '--verbose', action = 'store_true',
-                  help = 'print verbose messages')
 parser.add_option('-a', '--address', help = 'specify the address (ip:port) ' \
     'of an admin node', default = '127.0.0.1:9999')
 parser.add_option('--handler', help = 'specify the connection handler')
+parser.add_option('-l', '--logfile', help = 'specify a logging file')
 
 def main(args=None):
     (options, args) = parser.parse_args(args=args)
@@ -36,7 +35,7 @@ def main(args=None):
     else:
         address = ('127.0.0.1', 9999)
 
-    setupLog('NEOCTL', options.verbose)
+    logging.setup(options.logfile)
     from neo.neoctl.app import Application
 
     print Application(address).execute(args)

neo/scripts/neolog.py

+#!/usr/bin/env python
+#
+# neostorage - run a storage node of NEO
+#
+# Copyright (C) 2012  Nexedi SA
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import logging, os, sqlite3, sys, time
+from neo.lib.protocol import Packets, PacketMalformedError
+from neo.lib.util import dump
+
+def emit(date, name, levelname, msg_list):
+    d = int(date)
+    prefix = '%s.%04u %-9s %-10s ' % (time.strftime('%F %T', time.localtime(d)),
+                                      int((date - d) * 10000), levelname,
+                                      name or default_name)
+    for msg in msg_list:
+        print prefix + msg
+
+class packet(object):
+
+    def __new__(cls, date, name, msg_id, code, peer, body):
+        try:
+            p = Packets[code]
+        except KeyError:
+            Packets[code] = p = type('UnknownPacket[%u]' % code, (object,), {})
+        msg = ['#0x%04x %-30s %s' %