import logging.handlers
from ganeti import constants
+from ganeti import compat
class _ReopenableLogHandler(logging.handlers.BaseRotatingHandler):
return logging.Formatter("".join(parts))
+def _ReopenLogFiles(handlers):
+ """Wrapper for reopening all log handler's files in a sequence.
+
+ """
+ for handler in handlers:
+ handler.RequestReopen()
+
+
def SetupLogging(logfile, program, debug=0, stderr_logging=False,
multithreaded=False, syslog=constants.SYSLOG_USAGE,
- console_logging=False):
+ console_logging=False, root_logger=None):
"""Configures the logging module.
@type logfile: str
@type console_logging: boolean
@param console_logging: if True, will use a FileHandler which falls back to
the system console if logging fails
+ @type root_logger: logging.Logger
+ @param root_logger: Root logger to use (for unittests)
@raise EnvironmentError: if we can't open the log file and
syslog/stderr logging is disabled
formatter = _GetLogFormatter(progname, multithreaded, debug, False)
syslog_fmt = _GetLogFormatter(progname, multithreaded, debug, True)
- root_logger = logging.getLogger("")
+ reopen_handlers = []
+
+ if root_logger is None:
+ root_logger = logging.getLogger("")
root_logger.setLevel(logging.NOTSET)
# Remove all previously setup handlers
else:
logfile_handler.setLevel(logging.INFO)
root_logger.addHandler(logfile_handler)
+
+ reopen_handlers.append(logfile_handler)
+
+ return compat.partial(_ReopenLogFiles, reopen_handlers)
import unittest
import logging
import tempfile
+import shutil
from ganeti import constants
from ganeti import errors
raise Exception
+class TestSetupLogging(unittest.TestCase):
+ def setUp(self):
+ self.tmpdir = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmpdir)
+
+ def testSimple(self):
+ logfile = utils.PathJoin(self.tmpdir, "basic.log")
+ logger = logging.Logger("TestLogger")
+ self.assertTrue(callable(utils.SetupLogging(logfile, "test",
+ console_logging=False,
+ syslog=constants.SYSLOG_NO,
+ stderr_logging=False,
+ multithreaded=False,
+ root_logger=logger)))
+ self.assertEqual(utils.ReadFile(logfile), "")
+ logger.error("This is a test")
+
+ # Ensure SetupLogging used custom logger
+ logging.error("This message should not show up in the test log file")
+
+ self.assertTrue(utils.ReadFile(logfile).endswith("This is a test\n"))
+
+ def testReopen(self):
+ logfile = utils.PathJoin(self.tmpdir, "reopen.log")
+ logfile2 = utils.PathJoin(self.tmpdir, "reopen.log.OLD")
+ logger = logging.Logger("TestLogger")
+ reopen_fn = utils.SetupLogging(logfile, "test",
+ console_logging=False,
+ syslog=constants.SYSLOG_NO,
+ stderr_logging=False,
+ multithreaded=False,
+ root_logger=logger)
+ self.assertTrue(callable(reopen_fn))
+
+ self.assertEqual(utils.ReadFile(logfile), "")
+ logger.error("This is a test")
+ self.assertTrue(utils.ReadFile(logfile).endswith("This is a test\n"))
+
+ os.rename(logfile, logfile2)
+ assert not os.path.exists(logfile)
+
+ # Notify logger to reopen on the next message
+ reopen_fn()
+ assert not os.path.exists(logfile)
+
+ # Provoke actual reopen
+ logger.error("First message")
+
+ self.assertTrue(utils.ReadFile(logfile).endswith("First message\n"))
+ self.assertTrue(utils.ReadFile(logfile2).endswith("This is a test\n"))
+
+
if __name__ == "__main__":
testutils.GanetiTestProgram()