Merge pull request fluent#82 from luhn/nanosecond-time

repeatedly · web-flow · commit fffa39a5e120 · 2017-04-04T13:50:30.000+09:00
Nanosecond-precision timestamps
diff --git a/README.rst b/README.rst
@@ -90,6 +90,16 @@ fluentd, with tag 'app.follow' and the attributes 'from' and 'to'.
     cur_time = int(time.time())
     logger.emit_with_time('follow', cur_time, {'from': 'userA', 'to':'userB'})
 
+To send events with nanosecond-precision timestamps (Fluent 0.14 and up),
+specify `nanosecond_precision` on `FluentSender`.
+
+.. code:: python
+
+    # Use nanosecond
+    logger = sender.FluentSender('app', nanosecond_precision=True)
+    logger.emit('follow', {'from': 'userA', 'to': 'userB'})
+    logger.emit_with_time('follow', time.time(), {'from': 'userA', 'to': 'userB'})
+
 You can detect an error via return value of `emit`. If an error happens in `emit`, `emit` returns `False` and get an error object using `last_error` method.
 
 .. code:: python
diff --git a/fluent/sender.py b/fluent/sender.py
@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 
 from __future__ import print_function
+import struct
 import socket
 import threading
 import time
@@ -30,6 +31,18 @@ def get_global_sender():
 def close():
     get_global_sender().close()
 
+
+class EventTime(msgpack.ExtType):
+    def __new__(cls, timestamp):
+        seconds = int(timestamp)
+        nanoseconds = int(timestamp % 1 * 10 ** 9)
+        return super(EventTime, cls).__new__(
+            cls,
+            code=0,
+            data=struct.pack(">II", seconds, nanoseconds),
+        )
+
+
 class FluentSender(object):
     def __init__(self,
                  tag,
@@ -39,6 +52,7 @@ def __init__(self,
                  timeout=3.0,
                  verbose=False,
                  buffer_overflow_handler=None,
+                 nanosecond_precision=False,
                  **kwargs):
 
         self.tag = tag
@@ -48,6 +62,7 @@ def __init__(self,
         self.timeout = timeout
         self.verbose = verbose
         self.buffer_overflow_handler = buffer_overflow_handler
+        self.nanosecond_precision = nanosecond_precision
 
         self.socket = None
         self.pendings = None
@@ -61,10 +76,15 @@ def __init__(self,
             self._close()
 
     def emit(self, label, data):
-        cur_time = int(time.time())
+        if self.nanosecond_precision:
+            cur_time = EventTime(time.time())
+        else:
+            cur_time = int(time.time())
         return self.emit_with_time(label, cur_time, data)
 
     def emit_with_time(self, label, timestamp, data):
+        if self.nanosecond_precision and isinstance(timestamp, float):
+            timestamp = EventTime(timestamp)
         try:
             bytes_ = self._make_packet(label, timestamp, data)
         except Exception as e:
diff --git a/tests/test_sender.py b/tests/test_sender.py
@@ -3,6 +3,7 @@
 from __future__ import print_function
 import unittest
 import socket
+import msgpack
 
 import fluent.sender
 from tests import mockserver
@@ -64,6 +65,36 @@ def test_simple(self):
         self.assertTrue(data[0][1])
         self.assertTrue(isinstance(data[0][1], int))
 
+    def test_nanosecond(self):
+        sender = self._sender
+        sender.nanosecond_precision = True
+        sender.emit('foo', {'bar': 'baz'})
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(isinstance(data[0][1], msgpack.ExtType))
+        eq(data[0][1].code, 0)
+
+    def test_nanosecond_coerce_float(self):
+        time = 1490061367.8616468906402588
+        sender = self._sender
+        sender.nanosecond_precision = True
+        sender.emit_with_time('foo', time, {'bar': 'baz'})
+        sender._close()
+        data = self.get_data()
+        eq = self.assertEqual
+        eq(1, len(data))
+        eq(3, len(data[0]))
+        eq('test.foo', data[0][0])
+        eq({'bar': 'baz'}, data[0][2])
+        self.assertTrue(isinstance(data[0][1], msgpack.ExtType))
+        eq(data[0][1].code, 0)
+        eq(data[0][1].data, b'X\xd0\x8873[\xb0*')
+
     def test_no_last_error_on_successful_emit(self):
         sender = self._sender
         sender.emit('foo', {'bar': 'baz'})
@@ -93,3 +124,10 @@ def test_connect_exception_during_sender_init(self, mock_socket):
         mock_connect.side_effect = socket.error(EXCEPTION_MSG)
 
         self.assertEqual(self._sender.last_error.args[0], EXCEPTION_MSG)
+
+
+class TestEventTime(unittest.TestCase):
+    def test_event_time(self):
+        time = fluent.sender.EventTime(1490061367.8616468906402588)
+        self.assertEqual(time.code, 0)
+        self.assertEqual(time.data, b'X\xd0\x8873[\xb0*')
