Merge pull request #66 from supriyopaul/custom_handlers

README.md update one how to create and use custom handlers

Author: supriyopaul

.travis.yml

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

README.md

@@ -167,7 +167,7 @@ Collects all the logs from the server and parses it for making a common schema f
     ##### or
 - Docker run
     ```bash
-    $ sudo docker run --name collector --volume /var/log:/var/log deepcompute/logagg logagg --log-level INFO collect --file file=/var/log/serverstats.log:formatter=logagg.formatters.basescript --nsqtopic logagg --nsqd-http-address <nsq-server-ip-or-DNS>:4151
+    $ sudo docker run --name collector --volume /var/log:/var/log deepcompute/logagg logagg collect --file file=/var/log/serverstats.log:formatter=logagg.formatters.basescript --nsqtopic logagg --nsqd-http-address <nsq-server-ip-or-DNS>:4151
     ```
     - **Note**: Replace **<nsq-server-ip-or-DNS>** with the ip of `nsq` server eg.: **192.168.0.211**
     - **Note**: **--volume** argument is to mount local directory of log file into `Docker` `container`
@@ -322,7 +322,7 @@ $ python
 - You can see the basic format of record like below:
 ```json
 {
-  "_id" : ObjectId("some_id"),
+  "_id" : "20180301T065838_f7e042841d1d11e8bcf1000c2925b24d",
   "level" : "info",
   "timestamp" : "isoformat_time. Ex: 2017-08-01T07:32:24.183981Z",
   "data" : {},
@@ -332,7 +332,6 @@ $ python
   "event" : "default_event",
   "file" : "/path/to/log/file",
   "type" : "log | metric"
-  "id" : "20180301T065838_f7e042841d1d11e8bcf1000c2925b24d"
 }
 ```
 
@@ -390,8 +389,84 @@ time                request_time
 |MongoDBForwarder|`--target forwarder=logagg.forwarders.MongoDBForwarder:host=<mongoDB-server-ip>:port=<mongod-port-number>:user=<user-name>:password=<passwd>:db=<db-name>:collection=<collection name>`|
 |InfluxDBForwarder|`--target forwarder=logagg.forwarders.InfluxDBForwarder:host=<influxDB-server-ip>:port=<influxd-port-number>:user=<user-name>:password=<passwd>:db=<db-name>:collection=nothing`|
 
+**Note:** For using multiple forwarders use the format ``--target <forwarder1> <forwarder2>`` and not ``--target <forwarder1> --target <forwarder2>``
+
+### How to create and use custom formatters for log files
+#### Step 1: make a directory and append it's path to evironment variable $PYTHONPATH
+```bash
+$ echo $PYTHONPATH
+
+$ mkdir customformatters
+$ #Now append the path to $PYTHONPATH
+$ export PYTHONPATH=$PYTHONPATH:/home/path/to/customformatters/
+
+$ echo $PYTHONPATH
+:/home/path/to/customformatters
+```
+#### Step 2: Create a another directory and put your formatter file(s) inside it.
+
+```bash
+$ cd customformatters/
+$ mkdir myformatters
+$ cd myformatters/
+$ touch formatters.py
+$ touch __init__.py
+$ echo 'import formatters' >> __init__.py
+$ #Now write your formatter functions inside the formatters.py file
+```
+#### Step 3: Write your formatter functions inside the formatters.py file
+
+**Important:** 
+1. Only **python standard modules** can be imported in formatters.py file
+2. A formatter function should return a **dict()** `datatype`
+3. The 'dict()' should only contain keys which are mentioned in the above [log structure](https://github.com/deep-compute/logagg#features).
+4. Sample formatter functions:
+    ```python
+    import json 
+    import re
+
+    sample_log_line = '2018-02-07T06:37:00.297610Z [Some_event] [Info] [Hello_there]'
+
+    def sample_formatter(log_line):
+        log = re.sub('[\[+\]]', '',log_line).split(' ')
+        timestamp = log[0]
+        event = log[1]
+        level = log[2]
+        data = dict({'message': log[3]})
+
+        return dict(timestamp = timestamp,
+                     event = event,
+                     level = level,
+                     data = data,
+                    )
+     ```
+  To see more examples, look [here](https://github.com/deep-compute/logagg/blob/master/logagg/formatters.py) 
+
+5. Check if the custom handler works in `python interpreter` like for logagg.
+    ```python
+    >>> import myformatters
+    >>> sample_log_line = '2018-02-07T06:37:00.297610Z [Some_event] [Info] [Hello_there]'
+    >>> output = myformatters.formatters.sample_formatter(sample_log_line)
+    >>> from pprint import pprint
+    >>> pprint(output)
+    {'data': {'message': 'Hello_there'},
+     'event': 'Some_event',
+     'level': 'Info',
+     'timestamp': '2018-02-07T06:37:00.297610Z'}
+    ```
+6. Pseudo logagg collect commands:
+    ```
+    $ sudo logagg collect --file file=logfile.log:myformatters.formatters.sample_formatter --nsqtopic logagg --nsqd-http-address localhost:4151
+    ```
+    **or**
+    docker run 
+    ```
+    $ sudo docker run --name collector --env PYTHONPATH=$PYTHONPATH --volume /var/log:/var/log deepcompute/logagg logagg collect --file file=logfile.log:myformatters.formatters.sample_formatter --nsqtopic logagg --nsqd-http-address <nsq-server-ip-or-DNS>:4151
+    ```
+
 ---
-## Build on it
+
+## Build on logagg
 
 You're more than welcome to hack on this:-)
 

logagg/formatters.py

@@ -9,7 +9,7 @@ def docker_log_file_driver(line):
     log = json.loads(json.loads(line)['msg'])
     if 'formatter' in log.get('extra'):
         return RawLog(dict(formatter=log.get('extra').get('formatter'),
-                            raw=str(log.get('message')),
+                            raw=log.get('message'),
                             host=log.get('host'),
                             timestamp=log.get('timestamp'),
                             )

logagg/forwarders.py

@@ -53,14 +53,14 @@ def _ensure_connection(self):
                                          self.host,
                                          self.port)
         client = MongoClient(url, serverSelectionTimeoutMS=self.SERVER_SELECTION_TIMEOUT)
-        self.log.info('MongoDB_server_connection_established', host=self.host)
+        self.log.info('mongodb_server_connection_established', host=self.host)
         self.database = client[self.db_name]
-        self.log.info('MongoDB_database_created', db=self.db_name)
+        self.log.info('mongodb_database_created', db=self.db_name)
         self.collection = self.database[self.coll]
-        self.log.info('MongoDB_collection_created' ,
+        self.log.info('mongodb_collection_created' ,
                        collection=self.collection, db=self.db_name)
 
-    def _parse_msg_for_mongoDB(self, msgs):
+    def _parse_msg_for_mongodb(self, msgs):
         msgs_list = []
         #TODO: We need to do this by using iteration object.
         for msg in msgs:
@@ -75,19 +75,19 @@ def _insert_1by1(self, records):
             try:
                 self.collection.insert_one(r, ordered=False)
             except pymongo.errors.OperationFailure as opfail:
-                self.log.exception('failed_to_insert_record_in_mongoDB',
+                self.log.exception('failed_to_insert_record_in_mongodb',
                                     record=msg, tb=opfail.details)
 
     def handle_logs(self, msgs):
-        msgs_list = self._parse_msg_for_mongoDB(msgs)
+        msgs_list = self._parse_msg_for_mongodb(msgs)
         try:
             self.log.debug('inserting_msgs_mongodb')
             self.collection.insert_many([msg for msg in msgs_list], ordered=False)
             self.log.info('logs_inserted_into_mongodb', num_msgs=len(msgs), type='metric')
-        except pymongo.errors.AutoReconnect(message='connection_to_mongoDB_failed'):
+        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', tb=bwe.details)
             self._insert_1by1(msgs_list)
 
 
@@ -98,9 +98,9 @@ def handle_logs(self, msgs):
 from logagg.util import flatten_dict, is_number
 
 class InfluxDBForwarder(BaseForwarder):
-    EXCLUDE_TAGS = ["raw", "timestamp", "type", "event"]
+    EXCLUDE_TAGS = ["id","raw", "timestamp", "type", "event"]
 
-    influxDB_records = []
+    influxdb_records = []
 
     def __init__(self,
                  host, port,
@@ -116,12 +116,12 @@ def __init__(self,
         self._ensure_connection()
 
     def _ensure_connection(self):
-        # Establish connection to influxdb to store metrics
+        # Establish connection to influxDB to store metrics
         self.influxdb_client = InfluxDBClient(self.host, self.port, self.user,
                     self.passwd, self.db_name)
-        self.log.info('InfluxDB_server_connection_established', host=self.host)
+        self.log.info('influxdb_server_connection_established', host=self.host)
         self.influxdb_database = self.influxdb_client.create_database(self.db_name)
-        self.log.info('InfluxDB_database_created', dbname=self.db_name)
+        self.log.info('influxdb_database_created', dbname=self.db_name)
 
     def _tag_and_field_maker(self,event):
         t = dict()
@@ -134,7 +134,7 @@ def _tag_and_field_maker(self,event):
                     t[key] = event[key]
         return t, f
 
-    def parse_msg_for_influxDB(self, msgs):
+    def parse_msg_for_influxdb(self, msgs):
         #TODO: We need to do this by using iteration object.
         series = []
         for msg in msgs:
@@ -161,19 +161,18 @@ def handle_logs(self, msgs):
             msgs_list.append(msg_body)
 
         self.log.debug('parsing_of_metrics_started')
-        records = self.parse_msg_for_influxDB(msgs_list)
-        self.influxDB_records.extend(records)
+        records = self.parse_msg_for_influxdb(msgs_list)
+        self.influxdb_records.extend(records)
         self.log.debug('parsing_of_metrics_completed')
 
-        self.influxDB_records = [record for record in self.influxDB_records if record]
         try:
             self.log.debug('inserting_the_metrics_into_influxdb')
-            self.influxdb_client.write_points(self.influxDB_records)
+            self.influxdb_client.write_points(self.influxdb_records)
             self.log.info('metrics_inserted_into_influxdb',
-                           length=len(self.influxDB_records),
+                           length=len(self.influxdb_records),
                            type='metric')
-            self.influxDB_records = []
+            self.influxdb_records = []
         except (InfluxDBClientError, InfluxDBServerError) as e:
             self.log.exception('failed_to_insert metric',
-                                record=self.influxDB_records,
-                                length=len(self.influxDB_records))
+                                record=self.influxdb_records,
+                                length=len(self.influxdb_records))

setup.py

@@ -3,7 +3,7 @@
 
 setup(
     name="logagg",
-    version="0.2.6",
+    version="0.2.7",
     description="logs aggregation framework",
     keywords="logagg",
     author="Deep Compute, LLC",