Merge pull request #3 from tarazou9/logparse

Add garnet log file into logparse

Author: xgerman

garnet_log_events.py

@@ -0,0 +1,162 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+
+# Utility methods to deal with log events in DB.
+
+import json
+import os
+import shutil
+import sqlite3
+import statsd
+import time
+import traceback
+
+from contextlib import closing
+
+# set up statsd
+stats = statsd.StatsClient('localhost', 8125)
+account_name = os.getenv("MONITORING_MDM_ACCOUNT");
+namespace = os.getenv("MONITORING_MDM_NAMESPACE");
+tenant = os.getenv("MONITORING_TENANT");
+role = os.getenv("MONITORING_ROLE");
+role_instance = os.getenv("MONITORING_ROLE_INSTANCE");
+
+metric_identifier = {
+    "Account": account_name,
+    "Namespace": namespace,
+}
+
+def create_connection(database_file):
+    return sqlite3.connect(database_file)
+
+def create_table(connection):
+    with closing(connection.cursor()) as cursor:
+        query = """CREATE TABLE IF NOT EXISTS LogEvent (
+            event_product text,
+            event_category text,
+            event_type text,
+            event_date text,
+            attributes text,
+            PRIMARY KEY(event_product, event_category, event_type)
+        ); """
+        cursor.execute(query)
+
+def upsert_events(connection, events_map):
+    '''
+    Values of the events_map are upserted into the DB as part of a single transaction,
+    and committed at the end, to minimize disk I/O.
+    '''
+    if len(events_map) == 0:
+        return
+        
+    success = False
+    retry_attempt = 0
+    max_retry_attempts = 5
+    while not success and retry_attempt < max_retry_attempts:
+        with closing(connection.cursor()) as cursor:
+            try:
+                for event in events_map.values():
+                    attributes = get_attributes(event)
+                    upsert(connection, cursor, event["event_product"], event["event_category"], event["event_type"], event["date"], attributes)       
+                
+                connection.commit()
+                success = True
+
+            except sqlite3.Error as e:
+                print("Error occurred while comitting log events {0}. Retrying: {1}".format(str(events_map), str(e)))
+                connection.rollback()
+                success = False
+                retry_attempt += 1
+                if retry_attempt == max_retry_attempts:
+                    print("Emitting metrics for commit error")
+                    emit_commit_error_metrics()
+                else:
+                    time.sleep(1)
+
+def get_attributes(event):
+    attributes = {}
+
+    try:
+        if event["event_category"] == 'startup':
+            if event["event_type"] == 'ip_address_conflict':
+                attributes = {"endpoint": event["endpoint"]}
+
+        if event["event_category"] == 'tombstone':
+            if event["event_type"] == 'warning_threshold_exceeded':
+                attributes = {"tombstoned_cells": event["tombstoned_cells"], "keyspace": event["keyspace"], "table": event["table"]}
+
+            if event["event_type"] == 'error_threshold_exceeded':
+                attributes = {"live_cells": event['live_cells'], "tombstoned_cells": event['tombstoned_cells'], "keyspace": event['keyspace'], "table": event['table'], "key": event['key'], "requested_columns": event['requested_columns'], slice_start: event['slice_start'], "slice_end": event['slice_end'], "deletion_info": event['deletion_info']}
+    except:
+        ex = traceback.format_exc()
+        print("get_attributes: exception encountered - " + ex)
+        
+    return json.dumps(attributes)
+
+def upsert(connection, cursor, event_product, event_category, event_type, event_date, attributes):
+    query = """INSERT INTO LogEvent(event_product, event_category, event_type, event_date)
+    VALUES (?, ?, ?, ?) ON CONFLICT(event_product, event_category, event_type) DO UPDATE set event_date=?, attributes=?"""
+    values = (event_product, event_category, event_type, event_date, event_date, attributes)
+ 
+    cursor.execute(query, values)
+
+def emit_commit_error_metrics():
+    dims = metric_identifier.copy()
+    dims['Metric'] = "CommitLogEventsError"
+    dims['Dims'] = {
+        'Tenant': tenant,
+        "Role": role,
+        "RoleInstance": role_instance,
+        "Service": "cassandra",
+    }
+
+    emit_metrics(dims)
+
+    emit_metrics_to_file(dims, '/var/log/logevents_commit_error_metrics_new.json', '/var/log/logevents_commit_error_metrics.json')
+
+def emit_metrics(dims):
+    jsonDims = None
+    try:
+        jsonDims = json.dumps(dims)
+        stats.gauge(jsonDims, 1)
+    except Exception as e:
+        print("Error emitting metrics " + jsonDims + ": " + str(e))
+
+def emit_metrics_to_file(dims, src, dst):
+    # Dump metrics into local filesystem
+    try:
+        # Make a copy of destination file if it already exists
+        if os.path.exists(dst):
+            shutil.copyfile(dst, src)
+        
+        # Append to it and rename it as the destination file
+        with open(src, 'a+') as f:
+            f.write(json.dumps(dims))
+            f.close()
+            os.rename(f.name, dst)
+
+    except Exception as e:
+        print("Error writing out metrics to file "+ dst +": " + str(e))
+
+def init():
+    database_file = r"/var/lib/garnet/logevents.db"
+
+    connection = create_connection(database_file)
+    connection.execute('pragma journal_mode=wal')
+
+    create_table(connection)
+
+    return connection
+
+def main():
+    with closing(init()) as conn:
+        print("Log events stored in Nova DB:")
+
+        query = '''SELECT * FROM LogEvent'''
+        with closing(conn.cursor()) as cursor:
+            cursor.execute(query)
+            output = cursor.fetchall()
+            for row in output:
+                print(row)
+
+if __name__ == '__main__':
+    main()

log_garnet_daemon.py

@@ -0,0 +1,63 @@
+#!/home/azureuser/garnet-logparse/.venv/bin/python3
+
+# Copyright (c) Microsoft Corporation. All rights reserved.
+
+# Processes garnet log, emits it to Geneva, and stores log events in durable storage.
+
+import datetime
+import garnet_log_events
+import systemlog
+import time
+
+from contextlib import closing
+from fluent import sender
+from os import path
+from os import environ
+from pygtail import Pygtail
+
+
+# Set up fluent
+logger = sender.FluentSender('nova', port=25234, nanosecond_precision=True)
+
+log_file = '/var/log/garnet/log/garnet.log'
+sleep_time = 5 # Sleep time in seconds
+address = environ.get('ADDRESS')
+
+try:
+    with closing(garnet_log_events.init()) as connection:
+        while True:
+            if not path.exists(log_file):
+                print("Log file {0} does not exist. Going to sleep for {1} seconds".format(log_file, sleep_time))
+                time.sleep(sleep_time)
+                continue
+
+            print("File found")
+            lines = Pygtail(log_file) # Fetch log lines
+
+            events = dict()
+            num_lines = 0
+
+            for parsed_line in systemlog.parse_log(lines): # Processes each log line, and outputs it as a map of fields
+                num_lines += 1
+                # Emit the parsed log to Geneva
+                timestamp = datetime.datetime.timestamp(parsed_line["date"])
+                parsed_line["date"] = str(parsed_line["date"]) # If not converted to string, fluentd throws a serialization error for datetime object
+                parsed_line["address"] = address
+                if logger.emit_with_time('cassandra', timestamp, parsed_line):
+                    print("log sent successfully")
+                else:
+                    print("Failed to send log")
+
+                # Add the parsed log to a map, which will be iterated over later, and stored persistently in the DB
+                if parsed_line['event_type'] != 'unknown':
+                    key = "{0}:{1}:{2}".format(parsed_line["event_product"], parsed_line["event_category"], parsed_line["event_type"])
+                    events[key] = parsed_line
+
+            if num_lines == 0:
+                # This is to keep Pygtail from consuming >99% CPU
+                time.sleep(1)
+                continue
+
+            garnet_log_events.upsert_events(connection, events)
+finally:
+    print("Log parsing stopped")