Supports json-packaging of messages sent to fluentd.

panaetov · panaetov · commit 8c0b4b8fe51a · 2017-03-15T14:02:13.000+03:00
diff --git a/fluent/handler.py b/fluent/handler.py
@@ -115,12 +115,14 @@ def __init__(self,
                  host='localhost',
                  port=24224,
                  timeout=3.0,
+                 packager='msgpack',
                  verbose=False,
                  buffer_overflow_handler=None):
 
         self.tag = tag
         self.sender = sender.FluentSender(tag,
                                           host=host, port=port,
+                                          packager=packager,
                                           timeout=timeout, verbose=verbose,
                                           buffer_overflow_handler=buffer_overflow_handler)
         logging.Handler.__init__(self)
diff --git a/fluent/sender.py b/fluent/sender.py
@@ -5,6 +5,7 @@
 import time
 import traceback
 
+import json
 import msgpack
 
 from fluent.transport import Transport, TransportError
@@ -38,6 +39,7 @@ def __init__(self,
                  tag,
                  host='localhost',
                  port=24224,
+                 packager="msgpack",
                  bufmax=1 * 1024 * 1024,
                  timeout=3.0,
                  verbose=False,
@@ -51,6 +53,7 @@ def __init__(self,
         self.timeout = timeout
         self.verbose = verbose
         self.buffer_overflow_handler = buffer_overflow_handler
+        self.packager = self.get_packager(packager)
 
         self.pendings = None
         self.lock = threading.Lock()
@@ -100,7 +103,16 @@ def _make_packet(self, label, timestamp, data):
         packet = (tag, timestamp, data)
         if self.verbose:
             print(packet)
-        return msgpack.packb(packet)
+        return self.packager(packet)
+
+    def get_packager(self, name):
+        if name == 'json':
+            return json.dumps
+
+        if name == 'msgpack':
+            return msgpack.packb
+
+        raise RuntimeError("Unknown packager: {}", name)
 
     def _send(self, bytes_):
         self.lock.acquire()
