Merge pull request #74 from supriyopaul/bug_fixes

Bug fixes

Author: supriyopaul

.travis.yml

@@ -18,8 +18,8 @@ deploy:
     skip_cleanup: true
     api-key:
       secure: Rxl45qbTHWIbOhst3PS60ETfW5wDByxp0xv4ZbtgRGe4SPvHtOLHRNGiajsQX37pgUFF9ALcCseY2cTk46jNEA1jOzFx4DDSKyH+Wu4H5F4M8JDBBlIsvsgezumLsYMqOL18caZA8J84N9UyuzgdPBDb0B0mMclRa9xRaxWncrUZgXwW9r3N2zU1LvGtd0Su4zLXXP6HC6mKHdOOaNSDONqaesx1njYTGr5fbWy7IXrjSg75wWCtHW1dKDPXmyyWZomwpmhURYfYXn/o9lRaXSDpLWx4xTsbJQdG9EiSPm5fLjfv9tZTxIF7jB0tTrOB63gGAgrLu0zC5Z5MJ1Y0+sbotI8eySI4w0GTffhi4WQjTTyO02vgPuSCm9JV5aW+YeNJtSncEgaVgsuUmZUiWdqMsvPG+bqOjh/i0eIkHr/v7cyf3HndFieZH9H3XdlEDtyr4SRExQSjG+be6mcGOJMWMrXervcW6kGP3pcX7EWgrFxnkz9lSgx/0meNMP4JDo8pZWg50b0xpni3zUcweTgCIeYUBd5aIKUvPaCqSHC1BAyZI5z3Cvdlq0tjCS726drQcV4OJNjrnmb301/K6MBbXhAsyhbkB1NpUZ0k0ZwmGxQ7iE4N1pod2BQbTPxjNUL1KNQJXFvjr9Clrw9Arqo6X9S9t//GP2DDl5Ke5KQ=
-    name: logagg-0.2.8
-    tag_name: 0.2.8
+    name: logagg-0.2.9
+    tag_name: 0.2.9
     on:
         branch: master
         repo: deep-compute/logagg

logagg/collector.py

@@ -24,7 +24,8 @@ class LogCollector(object):
     DESC = 'Collects the log information and sends to NSQTopic'
 
     QUEUE_MAX_SIZE = 2000           # Maximum number of messages in in-mem queue
-    NBYTES_TO_SEND = 5 * (1024**2)  # Number of bytes from in-mem queue minimally required to push
+    MAX_NBYTES_TO_SEND = 4.5 * (1024**2)  # Number of bytes from in-mem queue minimally required to push
+    MIN_NBYTES_TO_SEND = 512 * 1024 # Minimum number of bytes to send to nsq in mpub
     MAX_SECONDS_TO_PUSH = 1         # Wait till this much time elapses before pushing
     LOG_FILE_POLL_INTERVAL = 0.25   # Wait time to pull log file for new lines added
     QUEUE_READ_TIMEOUT = 1          # Wait time when doing blocking read on the in-mem q
@@ -71,7 +72,10 @@ def _remove_redundancy(self, log):
     def validate_log_format(self, log):
         for key in log:
             assert (key in self.LOG_STRUCTURE)
-            assert isinstance(log[key], self.LOG_STRUCTURE[key])
+            try:
+                assert isinstance(log[key], self.LOG_STRUCTURE[key])
+            except AssertionError as e:
+                self.log.exception('formatted_log_structure_rejected' , log=log)
 
     @keeprunning(LOG_FILE_POLL_INTERVAL, on_error=util.log_exception)
     def collect_log_lines(self, log_file):
@@ -83,7 +87,6 @@ def collect_log_lines(self, log_file):
         for line_info in freader:
             line = line_info['line'][:-1] # remove new line char at the end
             log = dict(
-                    id=None,
                     file=fpath,
                     host=self.HOST,
                     formatter=L['formatter'],
@@ -123,58 +126,67 @@ def collect_log_lines(self, log_file):
             self.log.debug('waiting_for_pygtail_to_fully_ack', wait_time=t)
             time.sleep(t)
 
-    def _get_msgs_from_queue(self, msgs, msgs_nbytes, timeout):
+    def _get_msgs_from_queue(self, msgs, timeout):
+        msgs_pending = []
         read_from_q = False
         ts = time.time()
 
+        msgs_nbytes = sum(len(m['log']) for m in msgs)
+
         while 1:
             try:
                 msg = self.queue.get(block=True, timeout=self.QUEUE_READ_TIMEOUT)
                 read_from_q = True
                 self.log.debug("tally:get_from_self.queue")
-                msgs.append(msg)
-                msgs_nbytes += len(msg['log'])
 
-                if msgs_nbytes > self.NBYTES_TO_SEND: 
+                _msgs_nbytes = msgs_nbytes + len(msg['log'])
+                _msgs_nbytes += 1 # for newline char
+
+                if _msgs_nbytes > self.MAX_NBYTES_TO_SEND: 
+                    msgs_pending.append(msg)
                     self.log.debug('msg_bytes_read_mem_queue_exceeded')
                     break
+
+                msgs.append(msg)
+                msgs_nbytes = _msgs_nbytes
+
                 #FIXME condition never met
                 if time.time() - ts >= timeout and msgs:
                     self.log.debug('msg_reading_timeout_from_mem_queue_got_exceeded')
                     break
                     # TODO: What if a single log message itself is bigger than max bytes limit?
+
             except Queue.Empty:
                 self.log.debug('queue_empty')
                 time.sleep(self.QUEUE_READ_TIMEOUT)
                 if not msgs:
                     continue
                 else:
-                    return msgs, msgs_nbytes, read_from_q
+                    return msgs_pending, msgs_nbytes, read_from_q
+
         self.log.debug('got_msgs_from_mem_queue')
-        return msgs, msgs_nbytes, read_from_q
+        return msgs_pending, msgs_nbytes, read_from_q
 
 
     @keeprunning(0, on_error=util.log_exception) # FIXME: what wait time var here?
     def send_to_nsq(self, state):
         self.log.debug('send_to_nsq')
         msgs = []
-        msgs_nbytes = 0
         should_push = False
 
         while not should_push:
             cur_ts = time.time()
             self.log.debug('should_push', should_push=should_push)
             time_since_last_push = cur_ts - state.last_push_ts
 
-            msgs, msgs_nbytes, read_from_q = self._get_msgs_from_queue(msgs,
-                                                                        msgs_nbytes,
+            msgs_pending, msgs_nbytes, read_from_q = self._get_msgs_from_queue(msgs,
                                                                         self.MAX_SECONDS_TO_PUSH)
 
-            have_enough_msgs = msgs_nbytes >= self.NBYTES_TO_SEND
+            have_enough_msgs = msgs_nbytes >= self.MIN_NBYTES_TO_SEND
             is_max_time_elapsed = time_since_last_push >= self.MAX_SECONDS_TO_PUSH
 
             should_push = len(msgs) > 0 and (is_max_time_elapsed or have_enough_msgs)
-            self.log.debug('desciding_to_push', should_push=should_push,
+            self.log.debug('deciding_to_push', should_push=should_push,
                             time_since_last_push=time_since_last_push,
                             msgs_nbytes=msgs_nbytes)
 
@@ -183,7 +195,7 @@ def send_to_nsq(self, state):
             self.nsq_sender.handle_logs(msgs)
             self.confirm_success(msgs)
             self.log.debug('pushed_to_nsq', msgs_length=len(msgs))
-            msgs = []
+            msgs = msgs_pending
             state.last_push_ts = time.time()
         except (SystemExit, KeyboardInterrupt): raise
         finally:

logagg/nsqsender.py

@@ -1,4 +1,5 @@
 import time
+import ujson as json
 
 import requests
 from deeputil import keeprunning
@@ -15,7 +16,7 @@ def __init__(self, http_loc, nsq_topic, nsq_max_depth, log=util.DUMMY_LOGGER):
         self.topic_name = nsq_topic
         self.nsq_max_depth = nsq_max_depth
         self.log = log
-        
+
         self.session = requests.Session()
         self._ensure_topic(self.topic_name)
         self._ensure_topic(self.HEARTBEAT_TOPIC)
@@ -64,7 +65,7 @@ def _is_ready(self, topic_name):
             depth += sum(c.get('depth', 0) for c in topic['channels'])
             self.log.debug('nsq_depth_check', topic=topic_name,
                             depth=depth, max_depth=self.nsq_max_depth)
-            
+
             if depth < self.nsq_max_depth:
                 return
             else:
@@ -76,23 +77,20 @@ def _is_ready(self, topic_name):
                  exit_on_success=True,
                  on_error=util.log_exception)
     def _send_messages(self, msgs, topic_name):
-        if not isinstance(msgs, list):
-            data = msgs
-        else:
-            data = '\n'.join(m['log'] for m in msgs) #FIXME only works if 'log' is there
         url = self.MPUB_URL % (self.nsqd_http_address, topic_name)
         try:
-            self.session.post(url, data=data, timeout=5) # TODO What if session expires?
+            self.session.post(url, data=msgs, timeout=5) # TODO What if session expires?
         except (SystemExit, KeyboardInterrupt): raise
         except requests.exceptions.RequestException as e:
             raise
-        self.log.debug('nsq push done ', nmsgs=len(msgs), nbytes=len(data))
+        self.log.debug('nsq push done ', nmsgs=len(msgs), nbytes=len(msgs))
 
     def handle_logs(self, msgs):
         self._is_ready(topic_name=self.topic_name)
+        msgs = '\n'.join(m['log'] for m in msgs)
         self._send_messages(msgs, topic_name=self.topic_name)
 
-    def handle_heartbeat(self, msgs):
+    def handle_heartbeat(self, heartbeat):
+        msgs = json.dumps(heartbeat)
         self._is_ready(topic_name=self.HEARTBEAT_TOPIC)
         self._send_messages(msgs, topic_name=self.HEARTBEAT_TOPIC)
-

setup.py

@@ -3,7 +3,7 @@
 
 setup(
     name="logagg",
-    version="0.2.8",
+    version="0.2.9",
     description="logs aggregation framework",
     keywords="logagg",
     author="Deep Compute, LLC",
@@ -18,7 +18,7 @@
         "pymongo==3.6.0",
         "nsq-py==0.1.10",
         "influxdb==4.1.1",
-        "deeputil==0.1.2",
+        "deeputil==0.2.5",
         "ujson==1.35",
     ],
     package_dir={'logagg': 'logagg'},