Parsing Solr log files - version 2

Question

This post is in reference to: Parsing Solr log files

I re-wrote most of the code and split it up into a couple of classes. Currently, the class functionality is pretty limited, but I can see that I would need to reuse that functionality elsewhere and I plan to build them up a bit over time.

Let me know what you think.

Also, quick question:

send_rest in SolrServer just sends the rest of the buffer, if there is any data in there. I am indexing a dummy document with an id of commit via the regular send_dict_to_solr in same class.

Originally I was invoking the function like this:

    self.send_dict_to_solr({'id':'commit'},1)

But it wasn't working, so I had to do this weirdness with

    t = {}
    t['id'] = 'commit'
    self.send_dict_to_solr(t,1)

Any idea why?

logparse2.py

import re
import json
import requests
import argparse
import os
import sys
from solrlogparser import SolrLogParser
import datetime
from SolrServer import SolrServer
import gzip 
import time

parser = argparse.ArgumentParser(description='')
parser.add_argument('-solr', type=str, default='http://192.168.137.128:8983/solr/', nargs='+', help='Address of Solr server (ex: http://192.168.137.128:8983/solr/)')
parser.add_argument('-collection',  type=str, nargs='+', help='Name of Collection: (ex: collection1)')
parser.add_argument('-sendinc', default=1000, type=int, nargs=1, help='Number of documents after which to send and commit (ex: 1000)')
parser.add_argument('-commit', default=10000, type=int, nargs=1, help='Number of documents after which to send and commit (ex: 1000)')
parser.add_argument('-logs',  type=str, nargs='+', help='Directory of Log Files (ex: /opt/sw/solr/logs/)')
parser.add_argument('-workdir', default='./',type=str, nargs='+', help='Working Directory (ex: /opt/sw/solr/logs/)')
parser.add_argument('-archive',action='store_true', default=False, help='Use this to process log files in an archive, they have to be gzipped')
parser.add_argument('-tail',action='store_true', default=False, help='Use this to process log files in an archive, they have to be gzipped')
args = parser.parse_args()

maindata={}
maindata['controlfile']=args.workdir[0]+'parsercontrolfile2.txt'
control = {}

print(args)
def log_out(s):
    print("{} - {}".format(datetime.datetime.now(),s))



def main():
    if args.archive:
        log_out("Running in archive Processing Mode")
        for log in args.logs:
            if os.path.isfile(log):
                #It's a file
                if re.search('.log.gz$',log):
                    maindata['controlfile'] = os.path.dirname(os.path.realpath(log)) + os.sep + 'parsercontrolfile-archive.txt'
                    stat = archive_file_proc(log)
                    if stat == False:
                        log_out("Something Went Wrong with Processing " + log)
                else:
                    log_out(log + " Doesn't meet the criteria, files need to be compressed, ending in .log.gz")

            elif os.path.isdir(log):
                #It's a directory
                log_out("Going to Process {} in archive mode".format(log))
                files = doDir(log)
                if len(files) > 0:
                    maindata['controlfile']=log+'parsercontrolfile.txt'
                    for file in files:
                        log_out("Found "+ file)
                    for file in files:
                        stat = archive_file_proc(file)
                        if stat == False:
                            log_out("Something Went Wrong with Processing " + log)
            else:
                log_out("Supplied Input is not a valid directory or a compressed log file")
    elif args.tail:
        log_out("Running in -tail mode")
        if not os.path.isfile(args.logs[0]) and not os.path.isdir(args.logs[0]):
            log_out("Supply -logs to specify currently active log file or it's directory. Put in nohup and send to background ('&') in Linux")    
            sys.exit()
        else:
            maindata['controlfile'] = os.path.dirname(os.path.realpath(args.logs[0])) + os.sep + 'parsercontrolfile-tail.txt'
            if os.path.isfile(args.logs[0]):
                tail_file(args.logs[0])
            elif os.path.isdir(args.logs[0]):
                find_active_log(args.logs[0])
            else:
                log_out("Not a valid file specified")
    else:
        log_out("No Mode Specified, pick -archive for gzipped files, or -tail for current log files")

def find_active_log(dir):
    if not os.path.isdir(dir):
        log_out("Not a valid directory supplied")
    else:
        parser = SolrLogParser()
        log_out("Checking {} for Active Log Files".format(dir))
        dirfilelist = [os.path.join(dir,f) for f in os.listdir(dir) if os.path.isfile(os.path.join(dir,f))]
        files = {}
        for file in dirfilelist:
            if parser.logtype_uncomp(file) == 'solrcore':
                files[file] = {'filename':os.path.basename(file),'size':os.path.getsize(file),'mtime':os.path.getmtime(file),'active':0}
                log_out("Found " +file)
        time.sleep(15)
        for file in files:
            if os.path.getsize(file) > files[file]['size'] and os.path.getmtime(file) > files[file]['mtime']:
                tail_file(file)
        time.sleep(45)
        for file in files:
            if os.path.getsize(file) > files[file]['size'] and os.path.getmtime(file) > files[file]['mtime']:
                tail_file(file)

def tail_file(file):
    parser = SolrLogParser()
    solr = SolrServer(args.solr+args.collection[0],args.sendinc)

    if check_if_processed(file) == True:
        log_out("Already Processing " + file)
        return True
    else:
        if parser.logtype_uncomp(file) == 'solrcore':
            #mark_as_inprogress(file)
            log_out("Starting to Process " + file)
        else:
            log_out("File not in the correct name format " + file)
            return False


    startsize = os.path.getsize(file)

    with open(file,'r') as fh:
        print(control)
        filename = os.path.basename(file)
        if filename in control and (control[filename] != "0" or control[filename] != "1"):
            fh.seek(int(control[filename]))
            log_out("Resuming at " +control[filename])
        line = fh.readline()
        count = 0
        #left off here, need to set it up so it updates file byte offset in control file
        while line:
            data = {}
            data = parser.parseSolrCoreLine(str(line))
            if 'id' in data:
                count += 1

                if count % args.sendinc == 0:
                    #print("Processed %s lines - Sending Data to Solr" % (str(count)))
                    solr.send_dict_to_solr(data,1)
                    mark_as_inprogress(file, fh.tell())
                else: 
                    solr.send_dict_to_solr(data,0)
            line = fh.readline()

            if not line:
                log_out("Caught up on the File. Pending Changes")
                time.sleep(30)
                if startsize < os.path.getsize(file):
                    startsize = os.path.getsize(file)
                    #File grew while we were processing it, so it is still active, need to periodically scan it. 
                    line = fh.readline

        mark_as_inprogress(file, fh.tell())
        fh.close()
        solr.send_rest()
        log_out("Finished with "+file)
    find_active_log(os.path.dirname(file))

def doDir(directory):
    files = [os.path.join(directory,f) for f in os.listdir(directory) if os.path.isfile(os.path.join(directory,f)) and re.search('core.*.log.gz$',f)]
    if files:
        return files
    else:
        print("No files in the directory that end with .log.gz")
        return files

def archive_file_proc(file):
    parser = SolrLogParser()
    solr = SolrServer(args.solr+args.collection[0],args.sendinc)

    if check_if_processed(file) == True:
        log_out("Already Processed " + file)
        return True
    else:
        if parser.logtype_comp(file) == 'solrcore':
            mark_as_inprogress(file)
            log_out("Starting to Process " + file)
        else:
            return False

    with gzip.open(file,'r') as fh:
        count = 1
        for line in fh.readlines():
            data = {}

            data = parser.parseSolrCoreLine(str(line))
            if 'id' in data:
                count += 1
                if count % args.sendinc == 0:
                    print("Processed %s lines - Sending Data to Solr" % (str(count)))
                    solr.send_dict_to_solr(data,1)
                    #write_json(file,data)
                else: 
                    solr.send_dict_to_solr(data,0)
                    #write_json(file, data)
        fh.close()
        solr.send_rest()
        mark_as_processed(file)
        return True

def write_json(file, data):
    directory = os.path.dirname(os.path.realpath(file))
    pass


def check_if_processed(file):
    global control
    file = os.path.basename(file)
    control = read_control()
    #print(control)
    log_out("Checking {}".format(file))

    try:
        if file in control and (file[control] == "1" or file[control] == "0"):
            return True
    except:
        return False

def mark_as_processed(file):
    global control
    control = read_control()
    file = os.path.basename(file)
    control[file] = "1"
    log_out("Marking {} as Processed".format(file))
    write_control()


def mark_as_inprogress(file,*stat):
    global control

    if stat:
        stat = str(stat[0])
    else:
        stat = "0"
    control = read_control()
    file = os.path.basename(file)
    control[file] = stat
    log_out("Marking {} as In-Progress".format(file))
    write_control()

def write_control():
    if os.path.isfile(maindata['controlfile']):
        try:
            with open(maindata['controlfile'],'w') as maindata['pcf']:
                data = ''
                for file in control:
                    data += "{}\t{}\n".format(file,control[file])
                maindata['pcf'].write(data)
                maindata['pcf'].flush()
                maindata['pcf'].close()
        except:
            print("Couldn't Open Control File for Writing: " + maindata['controlfile'])
            sys.exit()

def read_control():
    global control
    control = {}
    if os.path.isfile(maindata['controlfile']):
        try:
            with open(maindata['controlfile'],'r+') as maindata['pcf']:
                for line in maindata['pcf'].readlines():
                    if line[:-1] == '\n':
                        a = line[:-1].split('\t')
                    else: 
                        a = line.split('\t')
                    control[a[0]] = a[1][0:-1]
                    maindata['pcf'].close()
            return control
        except:
            log_out("Couldn't Open Control File for Reading: " + maindata['controlfile'])
            sys.exit()
    else:
        try:
            maindata['pcf'] = open(maindata['controlfile'],'w+')
            maindata['pcf'].flush()
            maindata['pcf'].close()
            return control
        except:
            log_out("Couldn't Create Control File: " + maindata['controlfile'])
            sys.exit()
main()

solrparser.py:

import re
import json
import requests
import argparse
import os

class SolrLogParser:
    def __init__(self):
        pass

    def log_out(s):
        print("{} - {}".format(date.datetime.now(),s))

    def parseSolrCoreLine(self,l):
        out = {}
        final = {}
        temp = l.split(' ')
        if len(temp)>1:
            out['date'] = temp[0]
            out['time'] = temp[1]
            out['event_timestamp'] = "%sT%sZ" % (temp[0], temp[1])

            if len(temp)>7 and temp[3] == 'core.SolrCore':

                if temp[2] == "INFO" and temp[7] == 'path=/select':
                    if 'ids' in temp[8] or 'group.topgroups.gsin' in temp[8]:
                        return out
                    out['type'] = 'select'
                    out['collection'] = self.getCollection(temp[5])
                    out['hits'] = temp[9].replace('hits=','')
                    out['status'] = temp[10].replace('status=','')
                    out['qtime'] = temp[11].replace('QTime=','')
                    params = self.parseParams(temp[8])

                    if 'q' in params:
                        out.update(params)
                        out['id'] = out['event_timestamp'] + '_' + out['q']
                    elif 'fq' in out:
                        out.update(params)
                        out['id'] = out['event_timestamp'] + '_' + out['fq']

                    if 'id' in out:
                        out['id'] = out['id'].replace(':','_')
        return out    

    def parseParams(self, d):
        d = d.replace('{','')
        d = d.replace('}','')
        d = d.replace('"','')
        d = d.replace('params=','')
        out = {}
        t= d.split('&')
        count = 0
        for l in t:
            la = l.split('=')
            if la[0] == 'fq':
                out['fq'] = ''
                lb = la[1].split(':')
                out['fq_'+ lb[0]] = lb[1]
                out['fq'] += lb[0] + ' '
            else:
                if ':' in la[1]:
                    la[1] = la[1].replace(':','=')
                    out[la[0]] = la[1]
                else:
                    out[la[0]] = la[1]
        return out

    def getCollection(self,d):
        d = d.replace('[','')
        d = d.replace(']','')
        if re.search('_',d):
            a = re.match('^(.+)_.*_.*',d)
            d = a.group(1)
        return d

    def printarray(self, a):
        for i in range (0,len(a)):
            print( "%r - %r " % (i, a[i]) )
        print("\n\n\n--------\n\n\n")

    def logtype_comp(self, filename):
        type = ''
        if re.search(r'core.\d\d\d\d_\d\d_\d\d.log.gz$',filename):
            type='solrcore'
        return type

    def logtype_uncomp(self, filename):
        type = ''
        if re.search(r'core.\d\d\d\d_\d\d_\d\d.log$',filename):
            type='solrcore'
        return type        

SolrServer.py

import re
import json
import requests
import argparse
import os
import datetime
import atexit

class SolrServer:
    databuffer = ''
    buffercount = 0
    sendinc = 1000

    def __init__(self,solrserver,sendinc):
        self.solr = solrserver

        if solrserver[-1] == '/':
            self.solr = solrserver
        else:
            self.solr = solrserver + '/'

        self.sendinc = sendinc

    def set_param(p,v):
        self[p] = v

    def log_out(self,s):
        print("{} - {}".format(datetime.datetime.now(),s)) 

    def send_dict_to_solr(self,data,commit):
        data = json.dumps(data)
        if self.buffercount == 0:
            self.databuffer = '[\n'
        self.buffercount += 1
        self.databuffer += data + ',\n'

        if self.buffercount >= self.sendinc:
            if commit == 1 or commit == True:
                solr = self.solr+'update?commit=true'
                self.log_out("Sending Commit with next batch")
            else:
                solr = self.solr+'update'

            self.databuffer = self.databuffer[:-2]
            self.databuffer += ']\n'

            r = requests.post(solr,data=self.databuffer,headers = {'content-type': 'application/json'})
            if r.status_code != 200:
                self.log_out("ERROR - Couldn't Index Data Into Solr")
                self.log_out(r.status_code)
                self.log_out(r.raw)
                self.log_out(r.url)
                with open('error.log','w+') as er:
                    er.write(data)
                    er.close()
            else:
                self.log_out("Send Successful")
            r.close()
            self.databuffer = ''
            self.buffercount = 0
        return True

    def send_rest(self):
        self.buffercount = self.sendinc
        t = {}
        t['id'] = 'commit'
        self.send_dict_to_solr(t,1)

Answer 1

Here is my 2c. Please note this is constructive criticism, and goal is to help you get better. Please take any useful advice below, and disregard anything you don't agree with. I hope this helps. Some of this is extremely anal, but you will meet a lot of anal python developers who are religiously big on idioms. :-)

I would sort your imports above. Python idioms dictate a method along the lines of:

import (core libs first alphabetically)
import (third party libs last)

i.e.

import argparse
import datetime
import gzip
import json
import os
import re
import sys
import time

import requests

from solrlogparser import SolrLogParser
from SolrServer import SolrServer

I would add doc strings to your classes and methods. Since the goal is re-use, documentation is key. I typically use the following format for methods:

def some_method(*args):
    """Return some value that is the product of something I am multiplying. 

    Arguments -
    args: A list of numbers I am multiplying.

    """ 

I would space out statements like this for readability:

len(temp) > 1

This might be better done by by defining a list of of items you want to replace, then returning a list comprehension:

    d = d.replace('{','')
    d = d.replace('}','')
    d = d.replace('"','')
    d = d.replace('params=','')

Most books recommend the above method due to a supposed performance gain, however you should run your own measurements to see if the extra keystrokes really add value.

Using the following string:

badstuff = '{ ""this is a paramx"" params={test} "another test{}" ""testme{}"" }'

Your method above takes longer to complete then if I used a list and a for loop to perform the same task.

0.006 ms

List and for loop:

0.005 ms

The below method is not pep8 compliant.

def parseSolrCoreLine

I would recommend using CamelCase for your classes, and snake_case for your methods. Main goal is to make sure your method naming convention is consistent. Looks like you have CamelCase, and snake_case for your methods above.

Instead of creating a seperate log_out function, I would recommend looking into the python logging module.

import logging
logging.basicConfig(level=logging.INFO)
logging.info('Let me send something to stdout.')

It is customizable, and a much better solution for you in the long run.

Also, while single letter variables are accepted. I would recommend using them as a last resort. Per pep8, you are out of compliance on one of your variables 'l'. Here is the link in case you want to review:

PEP8 Names To Avoid

I would recommend going through the entire PEP8 doc.

Also this is generally not required:

commit == True:

You can just use:

commit:

If you are not initializing anything, maybe you may want to remove this extraneous method:

def __init__(self):
    pass

This line would be considered to not be pep8 compliant:

files[file] = {'filename':os.path.basename(file),'size':os.path.getsize(file),'mtime':os.path.getmtime(file),'active':0}

Consider using (note space between keys, and values for clarity):

files[file] = {
    'filename': os.path.basename(file),
    'size': os.path.getsize(file),
    'mtime': os.path.getmtime(file),
    'active': 0
}

In your method "parseSolrCoreLine" there are a lot of "if" statements checking for number of values based on your split. Look at "itertools.izip_longest". You could potentially eliminate a lot of your if statements with this module, and developer a more elegant solution.

--- 2nd pass

I would add a spaces in your concatenation.

self.solr + 'update?commit=true'

Based on your print statements, looks like you are coding for python3. If so you can rewrite this:

if self.buffercount == 0:

to

if not self.buffercount:

In python 3, 1 is true. I think you can probably rewrite this:

if commit == 1 or commit == True:

to

if commit:

No need to close your file when using a context mgr:

with open('error.log','w+') as er:
                er.write(data)
                er.close() # Remove this line.

Per the python documentation:

It is good practice to use the with keyword when dealing with file objects. This has the advantage that the file is properly closed after its suite finishes, even if an exception is raised on the way.

K.. Hopefully this is enough to get you goin. :-)