Implement FluentFormatter

This change implements FluentFormatter, a formatter for fluentd.
It enables oslo_log to output logs to fluentd directly.
FluentFormatter expects it will be used by fluent.handler.FluentHandler.

DocImpact
Implements: blueprint fluent-formatter
Change-Id: I67be5079f9370e93e5e3d4c715d2b8011154a2ce

doc/source/configfiles/example_nova.rst

@@ -24,6 +24,7 @@ Several handlers are created, to send messages to different outputs.
 And two formatters are created to be used based on whether the logging
 location will have OpenStack request context information available or
 not.
+A Fluentd formatter is also shown.
 
 .. literalinclude:: nova_sample.conf
    :language: ini
@@ -79,6 +80,23 @@ configuration settings from ``oslo.config``.
 
 .. literalinclude:: nova_sample.conf
    :language: ini
-   :lines: 80-81
+   :lines: 85-86
 
 The ``stdout`` and ``syslog`` handlers are defined, but not used.
+
+The ``fluent`` handler is useful to send logs to ``fluentd``.
+It is a part of fluent-logger-python and you can install it as following.
+
+::
+
+    $ pip install fluent-logger
+
+This handler is configured to use ``fluent`` formatter.
+
+.. literalinclude:: nova_sample.conf
+   :language: ini
+   :lines: 75-78
+
+.. literalinclude:: nova_sample.conf
+   :language: ini
+   :lines: 91-92

doc/source/configfiles/nova_sample.conf

@@ -2,10 +2,10 @@
 keys = root, nova
 
 [handlers]
-keys = stderr, stdout, watchedfile, syslog, null
+keys = stderr, stdout, watchedfile, syslog, fluent, null
 
 [formatters]
-keys = context, default
+keys = context, default, fluent
 
 [logger_root]
 level = WARNING
@@ -72,6 +72,11 @@ class = handlers.SysLogHandler
 args = ('/dev/log', handlers.SysLogHandler.LOG_USER)
 formatter = context
 
+[handler_fluent]
+class = fluent.handler.FluentHandler
+args = ('openstack.nova', 'localhost', 24224)
+formatter = fluent
+
 [handler_null]
 class = logging.NullHandler
 formatter = default
@@ -82,3 +87,6 @@ class = oslo_log.formatters.ContextFormatter
 
 [formatter_default]
 format = %(message)s
+
+[formatter_fluent]
+class = oslo_log.formatters.FluentFormatter

oslo_log/formatters.py

@@ -28,6 +28,9 @@ from oslo_context import context as context_utils
 from oslo_serialization import jsonutils
 from oslo_utils import encodeutils
 
+if six.PY3:
+    from functools import reduce
+
 
 def _dictify_context(context):
     if getattr(context, 'get_logging_values', None):
@@ -161,6 +164,68 @@ class JSONFormatter(logging.Formatter):
         return jsonutils.dumps(message)
 
 
+class FluentFormatter(logging.Formatter):
+    """A formatter for fluentd.
+
+    format() returns dict, not string.
+    It expects to be used by fluent.handler.FluentHandler.
+    (included in fluent-logger-python)
+
+    .. versionadded:: 3.17
+    """
+
+    def __init__(self, fmt=None, datefmt=None):
+        # NOTE(masaki) we ignore the fmt argument because of the same reason
+        #              with JSONFormatter.
+        self.datefmt = datefmt
+        try:
+            self.hostname = socket.gethostname()
+        except socket.error:
+            self.hostname = None
+
+    def formatException(self, exc_info, strip_newlines=True):
+        lines = traceback.format_exception(*exc_info)
+        if strip_newlines:
+            lines = reduce(lambda a, line: a + line.rstrip().splitlines(),
+                           lines, [])
+        return lines
+
+    def format(self, record):
+        message = {'message': record.getMessage(),
+                   'time': self.formatTime(record, self.datefmt),
+                   'name': record.name,
+                   'level': record.levelname,
+                   'filename': record.filename,
+                   'module': record.module,
+                   'funcname': record.funcName,
+                   'process_name': record.processName,
+                   'hostname': self.hostname,
+                   'traceback': None}
+
+        # Build the extra values that were given to us, including
+        # the context.
+        context = _update_record_with_context(record)
+        if hasattr(record, 'extra'):
+            extra = record.extra.copy()
+        else:
+            extra = {}
+        for key in getattr(record, 'extra_keys', []):
+            if key not in extra:
+                extra[key] = getattr(record, key)
+        # If we saved a context object, explode it into the extra
+        # dictionary because the values are more useful than the
+        # object reference.
+        if 'context' in extra:
+            extra.update(_dictify_context(context))
+            del extra['context']
+        message['extra'] = extra
+
+        if record.exc_info:
+            message['traceback'] = self.formatException(record.exc_info)
+
+        return message
+
+
 class ContextFormatter(logging.Formatter):
     """A context.RequestContext aware formatter configured through flags.
 

oslo_log/tests/unit/test_log.py

@@ -441,6 +441,69 @@ def get_fake_datetime(retval):
     return FakeDateTime
 
 
+class DictStreamHandler(logging.StreamHandler):
+    """Serialize dict in order to avoid TypeError in python 3. It is needed for
+    FluentFormatterTestCase.
+    """
+    def emit(self, record):
+        try:
+            msg = self.format(record)
+            jsonutils.dump(msg, self.stream)
+            self.stream.flush()
+        except AttributeError:
+            self.handleError(record)
+
+
+class FluentFormatterTestCase(LogTestBase):
+    def setUp(self):
+        super(FluentFormatterTestCase, self).setUp()
+        self.log = log.getLogger('test-fluent')
+        self._add_handler_with_cleanup(self.log,
+                                       handler=DictStreamHandler,
+                                       formatter=formatters.FluentFormatter)
+        self._set_log_level_with_cleanup(self.log, logging.DEBUG)
+
+    def test_fluent(self):
+        test_msg = 'This is a %(test)s line'
+        test_data = {'test': 'log'}
+        local_context = _fake_context()
+        self.log.debug(test_msg, test_data, key='value', context=local_context)
+
+        data = jsonutils.loads(self.stream.getvalue())
+        self.assertTrue('extra' in data)
+        extra = data['extra']
+        self.assertEqual('value', extra['key'])
+        self.assertEqual(local_context.auth_token, extra['auth_token'])
+        self.assertEqual(local_context.user, extra['user'])
+        self.assertEqual('test-fluent', data['name'])
+
+        self.assertEqual(test_msg % test_data, data['message'])
+
+        self.assertEqual('test_log.py', data['filename'])
+        self.assertEqual('test_fluent', data['funcname'])
+
+        self.assertEqual('DEBUG', data['level'])
+        self.assertFalse(data['traceback'])
+
+    def test_json_exception(self):
+        test_msg = 'This is %s'
+        test_data = 'exceptional'
+        try:
+            raise Exception('This is exceptional')
+        except Exception:
+            self.log.exception(test_msg, test_data)
+
+        data = jsonutils.loads(self.stream.getvalue())
+        self.assertTrue(data)
+        self.assertTrue('extra' in data)
+        self.assertEqual('test-fluent', data['name'])
+
+        self.assertEqual(test_msg % test_data, data['message'])
+
+        self.assertEqual('ERROR', data['level'])
+        self.assertTrue(data['traceback'])
+
+
 class ContextFormatterTestCase(LogTestBase):
     def setUp(self):
         super(ContextFormatterTestCase, self).setUp()