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)