Merge pull request #59 from fluent/save-last-error

Save last error

Author: repeatedly

.travis.yml

@@ -1,7 +1,6 @@
 sudo: false
 language: python
 python:
-  - "2.6"
   - "2.7"
   - "3.2"
   - "3.3"

README.rst

@@ -60,15 +60,60 @@ To quickly test your setup, add a matcher that logs to the stdout:
 Usage
 -----
 
+FluentSender Interface
+~~~~~~~~~~~~~~~~~~~~~~
+
+`sender.FluentSender` is a structured event logger for Fluentd.
+
+By default, the logger assumes fluentd daemon is launched locally. You
+can also specify remote logger by passing the options.
+
+.. code:: python
+
+    from fluent import sender
+
+    # for local fluent
+    logger = sender.FluentSender('app')
+
+    # for remote fluent
+    logger = sender.FluentSender('app', host='host', port=24224)
+
+For sending event, call `emit` method with your event. Following example will send the event to
+fluentd, with tag 'app.follow' and the attributes 'from' and 'to'.
+
+.. code:: python
+
+    # Use current time
+    logger.emit('follow', {'from': 'userA', 'to': 'userB'})
+
+    # Specify optional time
+    cur_time = int(time.time())
+    logger.emit_with_time('follow', cur_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
+
+    if not logger.emit('follow', {'from': 'userA', 'to': 'userB'}):
+        print(logger.last_error)
+        logger.clear_last_error() # clear stored error after handled errors
+
+If you want to shutdown the client, call `close()` method.
+
+.. code:: python
+
+    logger.close()
+
 Event-Based Interface
 ~~~~~~~~~~~~~~~~~~~~~
 
-First, you need to call ``logger.setup()`` to create global logger
+This API is a wrapper for `sender.FluentSender`.
+
+First, you need to call ``sender.setup()`` to create global `sender.FluentSender` logger
 instance. This call needs to be called only once, at the beggining of
 the application for example.
 
-By default, the logger assumes fluentd daemon is launched locally. You
-can also specify remote logger by passing the options.
+Initialization code of Event-Based API is below:
 
 .. code:: python
 
@@ -81,7 +126,7 @@ can also specify remote logger by passing the options.
     sender.setup('app', host='host', port=24224)
 
 Then, please create the events like this. This will send the event to
-fluent, with tag 'app.follow' and the attributes 'from' and 'to'.
+fluentd, with tag 'app.follow' and the attributes 'from' and 'to'.
 
 .. code:: python
 
@@ -93,11 +138,14 @@ fluent, with tag 'app.follow' and the attributes 'from' and 'to'.
       'to':   'userB'
     })
 
-If you want to shutdown the client, call `close()` method.
+`event.Event` has one limitation which can't return success/failure result.
+
+Other methods for Event-Based Interface.
 
 .. code:: python
 
-    sender.close()
+    sender.get_global_sender # get instance of global sender
+    sender.close # Call FluentSender#close
 
 Handler for buffer overflow
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -114,7 +162,7 @@ You can inject your own custom proc to handle buffer overflow in the event of co
         for unpacked in unpacker:
             print(unpacked)
 
-    sender.setup('app', host='host', port=24224, buffer_overflow_handler=handler)
+    logger = sender.FluentSender('app', host='host', port=24224, buffer_overflow_handler=handler)
 
 You should handle any exception in handler. fluent-logger ignores exceptions from ``buffer_overflow_handler``.
 

fluent/handler.py

@@ -106,7 +106,7 @@ def __init__(self,
 
     def emit(self, record):
         data = self.format(record)
-        self.sender.emit(None, data)
+        return self.sender.emit(None, data)
 
     def close(self):
         self.acquire()

fluent/sender.py

@@ -52,6 +52,7 @@ def __init__(self,
         self.socket = None
         self.pendings = None
         self.lock = threading.Lock()
+        self._last_error_threadlocal = threading.local()
 
         try:
             self._reconnect()
@@ -61,17 +62,18 @@ def __init__(self,
 
     def emit(self, label, data):
         cur_time = int(time.time())
-        self.emit_with_time(label, cur_time, data)
+        return self.emit_with_time(label, cur_time, data)
 
     def emit_with_time(self, label, timestamp, data):
         try:
             bytes_ = self._make_packet(label, timestamp, data)
-        except Exception:
+        except Exception as e:
+            self.last_error = e
             bytes_ = self._make_packet(label, timestamp,
                                        {"level": "CRITICAL",
                                         "message": "Can't output to log",
                                         "traceback": traceback.format_exc()})
-        self._send(bytes_)
+        return self._send(bytes_)
 
     def close(self):
         self.lock.acquire()
@@ -101,7 +103,7 @@ def _make_packet(self, label, timestamp, data):
     def _send(self, bytes_):
         self.lock.acquire()
         try:
-            self._send_internal(bytes_)
+            return self._send_internal(bytes_)
         finally:
             self.lock.release()
 
@@ -116,16 +118,24 @@ def _send_internal(self, bytes_):
 
             # send finished
             self.pendings = None
+
+            return True
         except socket.error as e:
+        #except Exception as e:
+            self.last_error = e
+
             # close socket
             self._close()
+
             # clear buffer if it exceeds max bufer size
             if self.pendings and (len(self.pendings) > self.bufmax):
                 self._call_buffer_overflow_handler(self.pendings)
                 self.pendings = None
             else:
                 self.pendings = bytes_
 
+            return False
+
     def _send_data(self, bytes_):
         # reconnect if possible
         self._reconnect()
@@ -152,6 +162,18 @@ def _call_buffer_overflow_handler(self, pending_events):
             # User should care any exception in handler
             pass
 
+    @property
+    def last_error(self):
+        return getattr(self._last_error_threadlocal, 'exception', None)
+
+    @last_error.setter
+    def last_error(self, err):
+        self._last_error_threadlocal.exception  =  err
+
+    def clear_last_error(self, _thread_id = None):
+        if hasattr(self._last_error_threadlocal, 'exception'):
+            delattr(self._last_error_threadlocal, 'exception')
+
     def _close(self):
         if self.socket:
             self.socket.close()

tests/test_event.py

@@ -3,21 +3,64 @@
 import unittest
 
 from fluent import event, sender
+from tests import mockserver
 
+class TestException(BaseException): pass
 
-sender.setup(server='localhost', tag='app')
+class TestEvent(unittest.TestCase):
+    def setUp(self):
+        self._server = mockserver.MockRecvServer('localhost')
+        sender.setup('app', port=self._server.port)
 
+    def tearDown(self):
+        from fluent.sender import _set_global_sender
+        sender.close()
+        _set_global_sender(None)
 
-class TestEvent(unittest.TestCase):
     def test_logging(self):
+        # XXX: This tests succeeds even if the fluentd connection failed
         # send event with tag app.follow
         event.Event('follow', {
             'from': 'userA',
             'to':   'userB'
         })
 
+    def test_logging_with_timestamp(self):
+        # XXX: This tests succeeds even if the fluentd connection failed
+
         # send event with tag app.follow, with timestamp
         event.Event('follow', {
             'from': 'userA',
             'to':   'userB'
         }, time=int(0))
+
+    def test_no_last_error_on_successful_event(self):
+        global_sender = sender.get_global_sender()
+        event.Event('unfollow', {
+            'from': 'userC',
+            'to':   'userD'
+        })
+
+        self.assertEqual(global_sender.last_error, None)
+        sender.close()
+
+    @unittest.skip("This test failed with 'TypeError: catching classes that do not inherit from BaseException is not allowed' so skipped")
+    #@patch('fluent.sender.socket')
+    def test_connect_exception_during_event_send(self, mock_socket):
+        # Make the socket.socket().connect() call raise a custom exception
+        mock_connect = mock_socket.socket.return_value.connect
+        EXCEPTION_MSG = "a event send socket connect() exception"
+        mock_connect.side_effect = TestException(EXCEPTION_MSG)
+
+        # Force the socket to reconnect while trying to emit the event
+        global_sender = sender.get_global_sender()
+        global_sender._close()
+
+        event.Event('unfollow', {
+            'from': 'userE',
+            'to':   'userF'
+        })
+
+        ex = global_sender.last_error
+        self.assertEqual(ex.args, EXCEPTION_MSG)
+        global_sender.clear_last_error()

tests/test_sender.py

@@ -2,9 +2,9 @@
 
 from __future__ import print_function
 import unittest
+import socket
 
 import fluent.sender
-
 from tests import mockserver
 
 
@@ -45,6 +45,9 @@ def setUp(self):
         self._sender = fluent.sender.FluentSender(tag='test',
                                                   port=self._server.port)
 
+    def tearDown(self):
+        self._sender.close()
+
     def get_data(self):
         return self._server.get_recieved()
 
@@ -60,3 +63,33 @@ def test_simple(self):
         eq({'bar': 'baz'}, data[0][2])
         self.assertTrue(data[0][1])
         self.assertTrue(isinstance(data[0][1], int))
+
+    def test_no_last_error_on_successful_emit(self):
+        sender = self._sender
+        sender.emit('foo', {'bar': 'baz'})
+        sender._close()
+
+        self.assertEqual(sender.last_error, None)
+
+    def test_last_error_property(self):
+        EXCEPTION_MSG = "custom exception for testing last_error property"
+        self._sender.last_error = socket.error(EXCEPTION_MSG)
+
+        self.assertEqual(self._sender.last_error.args[0], EXCEPTION_MSG)
+
+    def test_clear_last_error(self):
+        EXCEPTION_MSG = "custom exception for testing clear_last_error"
+        self._sender.last_error = socket.error(EXCEPTION_MSG)
+        self._sender.clear_last_error()
+
+        self.assertEqual(self._sender.last_error, None)
+
+    @unittest.skip("This test failed with 'TypeError: catching classes that do not inherit from BaseException is not allowed' so skipped")
+    #@patch('fluent.sender.socket')
+    def test_connect_exception_during_sender_init(self, mock_socket):
+        # Make the socket.socket().connect() call raise a custom exception
+        mock_connect = mock_socket.socket.return_value.connect
+        EXCEPTION_MSG = "a sender init socket connect() exception"
+        mock_connect.side_effect = socket.error(EXCEPTION_MSG)
+
+        self.assertEqual(self._sender.last_error.args[0], EXCEPTION_MSG)

tox.ini

@@ -1,6 +1,6 @@
 [tox]
 minversion = 1.7.2
-envlist = py26, py27, py32, py33, py34, py35
+envlist = py27, py32, py33, py34, py35
 skip_missing_interpreters = True
 
 [testenv]