fix 100% cpu usage, remove log_level to get all logs, reduce verbosity of SHIPPED messages

Author: Felix Barbalet

README.md

@@ -21,16 +21,14 @@ sudo -u pyjdbq pip install google-api-python-client pytz tzlocal -t /opt/pyjdbq/
 ```
 
 Download pyjdbq and move it into /opt/pyjdbq
+Put your JSON Google Big Query service account credentials.json file into the same directory
 
 ```bash
 
 curl -LOk https://github.com/xlfe/pyjdbq/archive/master.zip
 unzip master.zip
 sudo cp pyjdbq-master/* /opt/pyjdbq
-
-# Put your JSON Google Big Query service account credentials file into the same directory
 sudo mv credentials.json /opt/pyjdbq
-
 sudo chown -R pyjdbq: /opt/pyjdbq
 ```
 
@@ -59,7 +57,7 @@ optional arguments:
   --debug            Extra verbose
 ```
 
-Install the systemd unit file and check your journald logs to make sure there's no errors!
+Install the systemd unit file and check your journald logs to make sure there's no errors
 
 ```bash
 

journal_reader.py

@@ -13,6 +13,7 @@
 import json
 import pytz
 from tzlocal import get_localzone
+import time
 
 # get local timezone
 local_tz = get_localzone()
@@ -61,8 +62,8 @@ def __init__(self, writer, log, args):
 
         self.writer = writer
         j = journal.Reader()
-        j.log_level(journal.LOG_DEBUG)
         self.log = log
+        self.total_shipped = 0
         self.CURSOR_FILE = args.cursor
         self.COUNT_THRESHOLD = args.count
         self.SECOND_THRESHOLD = args.timeout
@@ -102,7 +103,7 @@ def check_bucket(self):
         Do we need to ship the log entries yet?
         """
 
-        if not self.bucket:
+        if not self.bucket or len(self.bucket) == 1:
             self.log.debug('Nothing in bucket')
             return False
 
@@ -119,14 +120,13 @@ def check_bucket(self):
 
 
     def run(self):
-        """
-
-        """
 
         while True:
+            time.sleep(1)
 
             #get all entries currently available
             for entry in self.journal:
+                #print transform_entry(entry)
                 self.bucket.append(transform_entry(entry))
                 self.cursor = entry['__CURSOR']
 
@@ -142,11 +142,6 @@ def run(self):
             else:
                 self.poll.poll(max_delay*1000)
 
-            if self.journal.process() != journal.APPEND:
-                #Ignore NOP and INVALIDATE entries
-                self.log.debug('NOP or INVALIDATE')
-                continue
-
 
     def ship_logs(self):
         """
@@ -156,7 +151,11 @@ def ship_logs(self):
         self.writer.put(self.bucket)
         self.last_ship = datetime.datetime.now()
         self.save_cursor()
-        self.log.info('SHIPPED count={}'.format(count))
+        self.log.debug('SHIPPED count={}'.format(count))
+        self.total_shipped += count
+        if self.total_shipped > 500:
+            self.log.info('SHIPPED count={}'.format(self.total_shipped))
+            self.total_shipped = 0
         self.bucket = []
 
 

schema.txt

@@ -1,5 +1,7 @@
-priority:integer,
+__realtime_timestamp:timestamp,
+_source_realtime_timestamp:timestamp,
 message,
+priority:integer,
 errno:integer,
 syslog_facility:integer,
 syslog_identifier,
@@ -15,7 +17,7 @@ _cap_effective,
 _audit_session:integer,
 _audit_loginuid:integer,
 _systemd_cgroup,
-_systemd_session,
+_systemd_session:integer,
 _systemd_unit,
 _systemd_user_unit,
 _systemd_owner_uid:integer,
@@ -33,6 +35,4 @@ _udev_devlink,
 code_file,
 code_line:integer,
 code_function,
-__realtime_timestamp:timestamp,
-_source_realtime_timestamp:timestamp,
 extra:string