qreplay.py

import os, sys, base64, select, socket, struct, datetime, time, traceback

host_all = '127.0.0.1'
port_sql = 0
port_api = 0
packet_max_len = 16384
verbose = 0
debug = 0

banner_ql = '--- crashed SphinxQL request dump ---'
banner_api = '--- crashed SphinxAPI request dump ---'
banner_end = '--- request dump end ---'

def printHelp ():
	print "\nUsage: python qreplay.py SEARCHD.LOG [OPTIONS]\n"
	print "Options are:"
	print "-pQL port\tSphinxQL port"
	print "-pAPI port\tSphinxAPI port"
	print "-pmax\t\tmax packet length to look for (default %d bytes)" % packet_max_len
	print "-h\t\thost name"
	print "-v\t\tshow query on send"
	print "-sleep seconds\tdelay after each query ( could be 0.3 )"
	print "--print\t\tonly print query"
	print "-? \t\tthis help screen"		
	sys.exit(0)

def readPacket ( fd, maxlen ):
	res = ''
	lines = 0
	for enc in fd:
		lines += 1
		if enc.startswith ( banner_end ):
			break
		res += enc
		if len ( res )>=maxlen:
			print "packed cut at %d bytes" % maxlen
			break
			
	return ( res, lines )
	
def sendQueryQL ( query, qlen, portQL ):
	if portQL<=0:
		return

	try:
		# setup
		db = MySQLdb.connect ( host=host_all, port=portQL )
		cur = db.cursor()
		
		# query
		cur.execute ( query )
		
	except Exception, e:
		print ( '\tERROR: SphinxQL query to %s:%d failed (error=%s)' % ( host_all, portQL, e ) )
		
def sendQueryAPI ( query, qlen, portAPI ):
	if portAPI<=0:
		return

	try:	
		# setup
		af = socket.AF_INET
		addr = ( host_all, portAPI )
		sock = socket.socket ( af, socket.SOCK_STREAM )
		sock.connect ( addr )
	except socket.error, msg:
		if sock:
			sock.close()
		print ( '\tERROR: SphinxAPI query to %s:%d failed (%s)' % ( host_all, portAPI, msg ) )
		return
	
	ver = struct.unpack('>L', sock.recv(4))
	# all ok, send my version
	sock.send(struct.pack('>L', 1))

	# query
	sent = sock.send ( query )
	if qlen!=sent:
		print ('\t ERROR on send: len=%d, sent=%d' % (qlen, sent) )
	time.sleep ( 0.001 )
#	got = sock.recv( 4 )
	sock.close()


def checkQL ( portQL ):
	if portQL<=0:
		return 0
		
	cur = None
	try:
		db = MySQLdb.connect ( host=host_all, port=portQL )
		cur = db.cursor()
		cur.close()
		return portQL
		
	except Exception, e:
		if cur:
			cur.close()
		print ( "ERROR: checkQL (host='%s', port=%d) failed (error='%s')" % ( host_all, portQL, e ) )
		return 0
		
def checkAPI ( portAPI ):
	if portAPI<=0:
		return 0
	
	sock = None
	try:
		af = socket.AF_INET
		addr = ( host_all, portAPI )
		sock = socket.socket ( af, socket.SOCK_STREAM )
		sock.connect ( addr )

	except socket.error, msg:		
		if sock:
			sock.close()

		print ( "ERROR: checkAPI (host='%s', port=%d) failed (error='%s')" % ( host_all, portAPI, msg ) )
		return 0
	
	sock.close()
	return portAPI
	
def printQL ( decoded ):
	if verbose==0:
		return
		
	print ("\t'%s'" % decoded.strip() )

def printAPI ( decoded ):
	if verbose==0:
		return

	(cmd, ver) = struct.unpack_from('>HH', decoded, 0)

	if debug!=0:
		#print ( decoded )
		print ( "%d 0x%x" % ( cmd, ver ) )
	
	ifrom = 32
	if cmd==0:
		if ver>=0x104:
			ifrom = 24
		if ver>=0x118:
			ifrom = 36
		if ver>=0x11C:
			ifrom = 40
		
	sortbylen = struct.unpack_from('>L', decoded, ifrom)[0]
	ifrom += 4
	sortby = struct.unpack_from('>'+str(sortbylen)+'s', decoded, ifrom)[0]
	ifrom += sortbylen
	qlen = struct.unpack_from('>L', decoded, ifrom)[0]
	ifrom += 4
	q = struct.unpack_from('>'+str(qlen)+'s', decoded, ifrom)[0]
	ifrom += qlen
	num_weights = struct.unpack_from('>L', decoded, ifrom)[0]
	ifrom += 4 + 4*num_weights
	ilen = struct.unpack_from('>L', decoded, ifrom)[0]
	ifrom += 4
	index = struct.unpack_from('>'+str(ilen)+'s', decoded, ifrom)[0]
	ifrom += ilen

	if sortbylen>0:
		print "\tquery = %s\n\tsort = %s\n\tindex = %s" % (q, sortby, index)
	else:
		print "\tquery = %s\n\tindex = %s" % (q, index)
		
##########################################################################

if not sys.argv[1:]:
	printHelp()

i = 0
crashlog = ''
sleep = 0
while (i<len(sys.argv)):
	arg = sys.argv[i].lower()
	if arg=='-pql':
		i += 1
		port_sql = int(sys.argv[i])
		import MySQLdb
	elif arg=='-papi':
		i += 1
		port_api = int(sys.argv[i])
	elif arg=='-pmax':
		i += 1
		packet_max_len = int(sys.argv[i])
	elif arg=='-h':
		i += 1
		host_all = string(sys.argv[i])
	elif arg=='-?':
		printHelp()
	elif arg=='-v':
		verbose = 1
	elif arg=='-sleep':
		i += 1
		sleep = float ( sys.argv[i] )
	elif arg=='--print':
		verbose = 2
	elif arg=='--debug':
		debug = 1
	else:
		crashlog = arg
	i += 1
	
fd = open ( crashlog, 'r' )
if not fd:
	print ( "ERROR: failed to open %s..." % sys.argv[1] )
	sys.exit ( 1 )

if verbose!=2:	
	port_sql = checkQL ( port_sql )
	port_api = checkAPI ( port_api )

i = 0
qfound = 0
qsend = 0
qerr = 0
for line in fd:
	i += 1
	# entry setup
	fn_send = None
	fn_decode = None
	port = 0
	found_tuple = None
	fn_print = None
	if line.startswith ( banner_ql ):
		fn_send = sendQueryQL
		fn_decode = lambda enc: enc
		port = port_sql
		found_tuple = ( 'SphinxQL', i )
		fn_print = printQL
	elif line.startswith ( banner_api ):
		fn_send = sendQueryAPI
		fn_decode = lambda enc: base64.standard_b64decode ( enc )
		port = port_api
		found_tuple = ( 'SphinxAPI', i )
		fn_print = printAPI

	if fn_send:
		try:
			# info on found
			print ( "+++ found %s banner on line=%d..." % found_tuple )
			qfound += 1
			
			# read crash log
			(encoded, lines ) = readPacket ( fd, packet_max_len )
			i += lines
			if not encoded or encoded=='':
				continue

			# decode
			decoded = fn_decode ( encoded )
			if not decoded or decoded=='':
				continue			

			# info on send
			fn_print ( decoded )

			qlen0 = len ( encoded )
			qlen1 = len ( decoded )
			print ( '[' + datetime.datetime.now().ctime() + ']' + ' read=' + str(qlen0) + ', sent=' + str(qlen1) + ', lines ' + str(i-lines+1) + ' => ' + str (i-1) )
			
			# send
			qsend += 1
			fn_send ( decoded, qlen1, port )
			
			if sleep>0:
				time.sleep ( sleep )
				
		except Exception, e:
			qerr += 1
			print 'failed error: %s' % str ( e )
			if debug!=0:
				traceback.print_exc()
	
print ( "\nQUERY TOTAL:\n\tfound=%d, sent=%d, decode error=%d" % ( qfound, qsend, qerr ) )