Merge pull request #76 from supriyopaul/bug_fixes

Fixed high usage of cpu by collector and metrics for forwarders.py
deep-compute · Mar 27, 2018 · 64d21d6 · 64d21d6
2 parents b049635 + a89e75c
commit 64d21d6
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 11 deletions.
diff --git a/.travis.yml b/.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.9
-    tag_name: 0.2.9
+    name: logagg-0.3.0
+    tag_name: 0.3.0
     on:
         branch: master
         repo: deep-compute/logagg

diff --git a/logagg/collector.py b/logagg/collector.py
@@ -87,6 +87,7 @@ 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'],
@@ -107,7 +108,7 @@ def collect_log_lines(self, log_file):
                     _log = util.load_object(formatter)(raw_log)
 
                 log.update(_log)
-                if 'id' not in log:
+                if log['id'] == None:
                     log['id'] = uuid.uuid1().hex
                 log = self._remove_redundancy(log)
                 self.validate_log_format(log)
@@ -125,6 +126,7 @@ def collect_log_lines(self, log_file):
             t = self.PYGTAIL_ACK_WAIT_TIME
             self.log.debug('waiting_for_pygtail_to_fully_ack', wait_time=t)
             time.sleep(t)
+        time.sleep(self.LOG_FILE_POLL_INTERVAL)
 
     def _get_msgs_from_queue(self, msgs, timeout):
         msgs_pending = []

diff --git a/logagg/forwarders.py b/logagg/forwarders.py
@@ -62,7 +62,6 @@ def _ensure_connection(self):
 
     def _parse_msg_for_mongodb(self, msgs):
         msgs_list = []
-        #TODO: We need to do this by using iteration object.
         for msg in msgs:
             msg['_id'] = msg.pop('id')
             msgs_list.append(msg)
@@ -74,18 +73,20 @@ def _insert_1by1(self, records):
                 self.collection.update({'_id': r['_id']}, r, upsert=True)
             except pymongo.errors.OperationFailure as opfail:
                 self.log.exception('failed_to_insert_record_in_mongodb',
-                                    record=r, tb=opfail.details)
+                                    record=r, tb=opfail.details,
+                                    num_records=1, type='metric')
 
     def handle_logs(self, msgs):
         msgs_list = self._parse_msg_for_mongodb(msgs)
         try:
             self.log.debug('inserting_msgs_mongodb')
             self.collection.insert_many(msgs_list, ordered=False)
-            self.log.info('logs_inserted_into_mongodb', num_msgs=len(msgs), type='metric')
+            self.log.info('logs_inserted_into_mongodb',
+                            num_records=len(msgs), type='metric')
         except pymongo.errors.AutoReconnect(message='connection_to_mongodb_failed'):
             self._ensure_connection()
         except pymongo.errors.BulkWriteError as bwe:
-            self.log.exception('bulk_write_to_mongodb_failed', tb=bwe.details)
+            self.log.exception('bulk_write_to_mongodb_failed')
             self._insert_1by1(msgs_list)
 
 
@@ -167,9 +168,10 @@ def handle_logs(self, msgs):
             self.log.debug('inserting_the_metrics_into_influxdb')
             self.influxdb_client.write_points(records)
             self.log.info('metrics_inserted_into_influxdb',
-                           length=len(records),
+                           num_records=len(records),
                            type='metric')
         except (InfluxDBClientError, InfluxDBServerError) as e:
             self.log.exception('failed_to_insert metric',
                                 record=records,
-                                length=len(records))
+                                num_records=len(records),
+                                type='metric')
diff --git a/setup.py b/setup.py
@@ -3,7 +3,7 @@
 
 setup(
     name="logagg",
-    version="0.2.9",
+    version="0.3.0",
     description="logs aggregation framework",
     keywords="logagg",
     author="Deep Compute, LLC",
@@ -14,7 +14,7 @@
         "https://github.com/deep-compute/pygtail/tarball/master/#egg=pygtail-0.6.1",
     ],
     install_requires=[
-        "basescript==0.1.13",
+        "basescript==0.2.0",
         "pymongo==3.6.0",
         "nsq-py==0.1.10",
         "influxdb==4.1.1",