qreplay.py

# Modified to (only) handle SEARCHD_COMMAND_EXCERPT = 1, VER_COMMAND_EXCERPT = 0x104
# as currently defined in
# https://github.com/sphinxsearch/sphinx/blob/master/api/sphinxapi.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 ):
			print "hit end at %d lines" % lines
			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

	(decoded, (cmd, ver, length)) = unpack('>2HL', decoded)

	if debug != 0:
		print "cmd %d, ver 0x%x, length %d" % (cmd, ver, length)
	
	if cmd == 1 and ver == 0x104:
		(decoded, flags) = unpack('>2L', decoded)
		for key in ["index", "words", "before_match", "after_match", "chunk_separator"]:
			(decoded, value) = unpack_string('>L', decoded)
			print "%s = %s" % (key, value)
		for key in ["limit", "around", "limit_passages", "limit_words", "start_passage_id"]:
			(decoded, (value,)) = unpack('>L', decoded)
			print "%s = %s" % (key, value)
		for key in ["html_strip_mode", "passage_boundary"]:
			(decoded, value) = unpack_string('>L', decoded)
			print "%s = %s" % (key, value)
		(decoded, (number_of_docs,)) = unpack('>L', decoded)
		print "number_of_docs = %d" % number_of_docs
		for doc in xrange(1, number_of_docs - 1):
			(decoded, value) = unpack_string('>L', decoded)
			print "doc = %s" % value

def unpack(format, bytes_array):
	values = struct.unpack_from(format, bytes_array)
	after = struct.calcsize(format)
	return (bytes_array[after:], values)

def unpack_string(length_format, bytes_array):
	(bytes_array, (string_length,)) = unpack(length_format, bytes_array)
	string_format = '>' + str(string_length) + 's'
	(bytes_array, (string,)) = unpack(string_format, bytes_array)
	return (bytes_array, string)
		
##########################################################################

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 ) )