~jan-kneschke/mysql-proxy/packet-tracking-assertions

--[[ $%BEGINLICENSE%$
 Copyright (c) 2007, 2008, Oracle and/or its affiliates. All rights reserved.

 This program is free software; you can redistribute it and/or
 modify it under the terms of the GNU General Public License as
 published by the Free Software Foundation; version 2 of the
 License.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 GNU General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with this program; if not, write to the Free Software
 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
 02110-1301  USA

 $%ENDLICENSE%$ --]]

local commands     = require("proxy.commands")
local tokenizer    = require("proxy.tokenizer")
local auto_config  = require("proxy.auto-config")

---
-- configuration
--
-- SET GLOBAL analyze_query.auto_filter = 0
if not proxy.global.config.analyze_query then
	proxy.global.config.analyze_query = {
		analyze_queries  = true,   -- track all queries
		auto_explain     = false,  -- execute a EXPLAIN on SELECT queries after we logged
		auto_processlist = false,  -- execute a SHOW FULL PROCESSLIST after we logged
		min_queries      = 1000,   -- start logging after evaluating 1000 queries
		min_query_time   = 1000,   -- don't log queries less then 1ms
		query_cutoff     = 160,    -- only show the first 160 chars of the query
		log_slower_queries = false -- log the query if it is slower than the queries of the same kind
	}
end

--- dump the result-set to stdout
--
-- @param inj "packet.injection"
function resultset_to_string( res )
	local s = ""
	local fields = res.fields

	if not fields then return "" end

	local field_count = #fields

	local row_count = 0
	for row in res.rows do
		local o
		local cols = {}

		for i = 1, field_count do
			if not o then
				o = ""
			else 
				o = o .. ", "
			end

			if not row[i] then
				o = o .. "NULL"
			-- TODO check if the indexes are in range 
			elseif fields[i].type == proxy.MYSQL_TYPE_STRING or  
				fields[i].type == proxy.MYSQL_TYPE_VAR_STRING or 
				fields[i].type == proxy.MYSQL_TYPE_LONG_BLOB or 
				fields[i].type == proxy.MYSQL_TYPE_MEDIUM_BLOB then
				o = o .. string.format("%q", row[i])
			else
				-- print("  [".. i .."] field-type: " .. fields[i].type)
				o = o .. row[i]
			end
		end
		s = s .. ("  ["..row_count.."]{ " .. o .. " }\n")
		row_count = row_count + 1
	end

	return s
end


function math.rolling_avg_init()
	return { count = 0, value = 0, sum_value = 0 }
end

function math.rolling_stddev_init()
	return { count = 0, value = 0, sum_x = 0, sum_x_sqr = 0 }
end

function math.rolling_avg(val, tbl) 
	tbl.count = tbl.count + 1
	tbl.sum_value = tbl.sum_value + val

	tbl.value = tbl.sum_value / tbl.count

	return tbl.value
end


function math.rolling_stddev(val, tbl)
	tbl.sum_x     = tbl.sum_x + val
	tbl.sum_x_sqr = tbl.sum_x_sqr + (val * val)
	tbl.count     = tbl.count + 1

	tbl.value = math.sqrt((tbl.count * tbl.sum_x_sqr - (tbl.sum_x * tbl.sum_x)) / (tbl.count * (tbl.count - 1)))

	return tbl.value
end

---
-- init query counters
-- 
-- the normalized queries are 
if not proxy.global.queries then
	proxy.global.queries = { }
end

if not proxy.global.baseline then
	proxy.global.baseline = {
		avg = math.rolling_avg_init(),
		stddev = math.rolling_stddev_init()
	}
end


function read_query(packet) 
	local cmd = commands.parse(packet)

	local r = auto_config.handle(cmd)
	if r then return r end

	-- analyzing queries is disabled, just pass them on
	if not proxy.global.config.analyze_query.analyze_queries then
		return
	end

	-- we only handle normal queries
	if cmd.type ~= proxy.COM_QUERY then
		return
	end

	tokens = tokenizer.tokenize(cmd.query)
	norm_query = tokenizer.normalize(tokens)

	-- create a id for this query
	query_id   = ("%s.%s.%s"):format(
		proxy.connection.backend_ndx, 
		proxy.connection.client.default_db ~= "" and 
			proxy.connection.client.default_db or 
			"(null)", 
		norm_query)

	-- handle the internal data
	if norm_query == "SELECT * FROM `histogram` . `queries` " then
		proxy.response = {
			type = proxy.MYSQLD_PACKET_OK,
			resultset = {
				fields = { 
					{ type = proxy.MYSQL_TYPE_STRING,
					  name = "query" },
					{ type = proxy.MYSQL_TYPE_LONG,
					  name = "COUNT(*)" },
					{ type = proxy.MYSQL_TYPE_DOUBLE,
					  name = "AVG(qtime)" },
					{ type = proxy.MYSQL_TYPE_DOUBLE,
					  name = "STDDEV(qtime)" },
				}
			}
		}

		local rows = {}
		if proxy.global.queries then
			local cutoff = proxy.global.config.analyze_query.query_cutoff

			for k, v in pairs(proxy.global.queries) do
				local q = v.query

				if cutoff and cutoff < #k then
					q = k:sub(1, cutoff) .. "..."
				end

				rows[#rows + 1] = { 
					q,
					v.avg.count,
					string.format("%.2f", v.avg.value),
					v.stddev.count > 1 and string.format("%.2f", v.stddev.value) or nil
				}
			end
		end
		
		proxy.response.resultset.rows = rows

		return proxy.PROXY_SEND_RESULT
	elseif norm_query == "DELETE FROM `histogram` . `queries` " then
		proxy.response = {
			type = proxy.MYSQLD_PACKET_OK,
		}
		proxy.global.queries = {}
		return proxy.PROXY_SEND_RESULT
	end

	-- we are connection-global
	baseline = nil
	log_query = false

	---
	-- do we want to analyse this query ?
	--
	--
	local queries = proxy.global.queries

	if not queries[query_id] then
		queries[query_id] = {
			db     = proxy.connection.client.default_db,
			query  = norm_query,
			avg    = math.rolling_avg_init(),
			stddev = math.rolling_stddev_init()
		}
	end


	if queries[query_id].do_analyze then
		-- wrap the query in SHOW SESSION STATUS
		--
		-- note: not used yet
		proxy.queries:append(2, string.char(proxy.COM_QUERY) .. "SHOW SESSION STATUS", { resultset_is_needed = true })
		proxy.queries:append(1, packet, { resultset_is_needed = true }) -- FIXME: this should work without it being set (inj.resultset.query_status)
		proxy.queries:append(3, string.char(proxy.COM_QUERY) .. "SHOW SESSION STATUS", { resultset_is_needed = true })
	else
		proxy.queries:append(1, packet, { resultset_is_needed = true }) -- FIXME: this one too
	end

	return proxy.PROXY_SEND_QUERY
end

---
-- normalize a timestamp into a string
--
-- @param ts time in microseconds
-- @return a string with a useful suffix
function normalize_time(ts)
	local suffix = "us"
	if ts > 10000 then
		ts = ts / 1000
		suffix = "ms"
	end
	
	if ts > 10000 then
		ts = ts / 1000
		suffix = "s"
	end

	return string.format("%.2f", ts) .. " " .. suffix
end

function read_query_result(inj)
	local res = assert(inj.resultset)
	local config = proxy.global.config.analyze_query

	if inj.id == 1 then
		log_query = false
		if not res.query_status or res.query_status == proxy.MYSQLD_PACKET_ERR then
			-- the query failed, let's clean the queue
			--
			proxy.queries:reset()
		end

		-- get the statistics values for the query
		local bl = proxy.global.baseline
		local avg_query_time    = math.rolling_avg(   inj.query_time, bl.avg)
		local stddev_query_time = math.rolling_stddev(inj.query_time, bl.stddev)

		local st = proxy.global.queries[query_id]
		local q_avg_query_time    = math.rolling_avg(   inj.query_time, st.avg)
		local q_stddev_query_time = math.rolling_stddev(inj.query_time, st.stddev)

		o = "# " .. os.date("%Y-%m-%d %H:%M:%S") .. 
			" [".. proxy.connection.server.thread_id .. 
			"] user: " .. proxy.connection.client.username .. 
			", db: " .. proxy.connection.client.default_db .. "\n" ..
			"  Query:                     " .. string.format("%q", inj.query:sub(2)) .. "\n" ..
			"  Norm_Query:                " .. string.format("%q", norm_query) .. "\n" ..
			"  query_time:                " .. normalize_time(inj.query_time) .. "\n" ..
			"  global(avg_query_time):    " .. normalize_time(avg_query_time) .. "\n" ..
			"  global(stddev_query_time): " .. normalize_time(stddev_query_time) .. "\n" ..
			"  global(count):             " .. bl.avg.count .. "\n" ..
			"  query(avg_query_time):     " .. normalize_time(q_avg_query_time) .. "\n" ..
			"  query(stddev_query_time):  " .. normalize_time(q_stddev_query_time) .. "\n" ..
			"  query(count):              " .. st.avg.count .. "\n" 


		-- we need a min-query-time to filter out
		if inj.query_time >= config.min_query_time then
			-- this query is slower than 95% of the average
			if bl.avg.count > config.min_queries and
			   inj.query_time > avg_query_time + 5 * stddev_query_time then
				o = o .. "  (qtime > global-threshold)\n"
				log_query = true
			end
	
			-- this query was slower than 95% of its kind
			if config.log_slower_queries and
			   st.avg.count > config.min_queries and 
			   inj.query_time > q_avg_query_time + 5 * q_stddev_query_time then
				o = o .. "  (qtime > query-threshold)\n"
				log_query = true
			end
		end

		if log_query and config.auto_processlist then
			proxy.queries:append(4, string.char(proxy.COM_QUERY) .. "SHOW FULL PROCESSLIST",
				{ resultset_is_needed = true })
		end
	
		if log_query and config.auto_explain then
			if tokens[1] and tokens[1].token_name == "TK_SQL_SELECT" then
				proxy.queries:append(5, string.char(proxy.COM_QUERY) .. "EXPLAIN " .. inj.query:sub(2),
					{ resultset_is_needed = true })
			end
		end

	elseif inj.id == 2 then
		-- the first SHOW SESSION STATUS
		baseline = {}
		
		for row in res.rows do
			-- 1 is the key, 2 is the value
			baseline[row[1]] = row[2]
		end
	elseif inj.id == 3 then
		local delta_counters = { }
		
		for row in res.rows do
			if baseline[row[1]] then
				local num1 = tonumber(baseline[row[1]])
				local num2 = tonumber(row[2])

				if row[1] == "Com_show_status" then
					num2 = num2 - 1
				elseif row[1] == "Questions" then
					num2 = num2 - 1
				elseif row[1] == "Last_query_cost" then
					num1 = 0
				end
				
				if num1 and num2 and num2 - num1 > 0 then
					delta_counters[row[1]] = (num2 - num1)
				end
			end
		end
		baseline = nil
	
		--- filter data
		-- 
		-- we want to see queries which 
		-- - trigger a tmp-disk-tables
		-- - are slower than the average
		--

		if delta_counters["Created_tmp_disk_tables"] then
			log_query = true
		end

		if log_query then
			for k, v in pairs(delta_counters) do
				o = o .. ".. " .. row[1] .. " = " .. (num2 - num1) .. "\n"
			end
		end
	elseif inj.id == 4 then
		-- dump the full result-set to stdout
		o = o .. "SHOW FULL PROCESSLIST\n"
		o = o .. resultset_to_string(inj.resultset)
	elseif inj.id == 5 then
		-- dump the full result-set to stdout
		o = o .. "EXPLAIN <query>\n"
		o = o .. resultset_to_string(inj.resultset)
	end

	if log_query and proxy.queries:len() == 0 then
		print(o)
		o = nil
		log_query = false
	end

	-- drop all resultsets but the original one
	if inj.id ~= 1 then return proxy.PROXY_IGNORE_RESULT end
end