D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
proc
/
self
/
root
/
opt
/
dedrads
/
mysql
/
Filename :
slowqueryparser.py
back
Copy
#!/usr/lib/rads/venv/bin/python3 import argparse import configparser from pathlib import Path import sys import re from pymysql.optionfile import Parser as PyMySQLParser USER_RE = re.compile(r'^# User@Host:\s+([a-z0-9]+)') STATS_RE = re.compile( r'^# Query_time:\s+([0-9\.]+)\s+Lock_time:\s+([0-9\.]+)\s+' r'Rows_sent:\s+(\d+)\s+Rows_examined:\s+(\d+)' ) def parse_args(): parser = argparse.ArgumentParser() # fmt: off parser.add_argument( "-q", "--quiet", dest="quiet", action='store_true', help="Suppress stderr output", ) parser.add_argument( '-H', '--no-header', dest='no_header', action='store_true', help='Suppress column headers', ) parser.add_argument( "-o", "--output", metavar="FILE", help="Write output to FILE (default: stdout)", ) parser.add_argument( "-u", "--user", metavar="USER", default=None, help="Output USER's queries instead of tallys", ) parser.add_argument( "-a", "--average", action="store_true", help="Print averages per query instead of totals", ) parser.add_argument('logpath', nargs='?', help='Path to slow query log') # fmt: on return parser.parse_args() class MySQLUser: """Holds a user name and tracks numbers of queries""" def __init__(self, username: str): self.username = username self.num_queries = 0 self.query_time = 0.0 self.lock_time = 0.0 self.rows_sent = 0 self.rows_examined = 0 def add_query(self, stats_match: re.Match): query_time, lock_time, rows_sent, rows_examined = stats_match.groups() self.num_queries += 1 self.query_time += float(query_time) self.lock_time += float(lock_time) self.rows_sent += int(rows_sent) self.rows_examined += int(rows_examined) @classmethod def row_header(cls, file=sys.stdout): cls.header( ['QUERIES', 'TIME', 'LOCKTIME', 'ROWSSENT', 'ROWSRECVD'], file=file ) @classmethod def avg_header(cls, file=sys.stdout): cls.header( ['QUERIES', 'TIME/Q', 'LOCKTIME/Q', 'ROWSSENT/Q', 'ROWSRECV/Q'], file=file, ) @staticmethod def header(cols: list[str], file=sys.stdout): print('USER'.rjust(16), end=' ', file=file) print(*map(lambda x: x.rjust(10), cols), file=file) def row_print(self, file=sys.stdout): print( self.username.rjust(16), str(self.num_queries).rjust(10), str(int(self.query_time)).rjust(10), str(int(self.lock_time)).rjust(10), str(self.rows_sent).rjust(10), str(self.rows_examined).rjust(10), file=file, ) def avg_print(self, file=sys.stdout): print( self.username.rjust(16), str(self.num_queries).rjust(10), str(int(self.query_time / self.num_queries)).rjust(10), str(int(self.lock_time / self.num_queries)).rjust(10), str(int(self.rows_sent / self.num_queries)).rjust(10), str(int(self.rows_examined / self.num_queries)).rjust(10), file=file, ) def default_log_path(): try: parser = PyMySQLParser(strict=False) if not parser.read('/etc/my.cnf'): return None path = Path(parser.get('mysqld', 'slow_query_log_file')).resolve() if path == Path('/dev/null'): print("MySQL log points to /dev/null currently", file=sys.stderr) return None return str(path) except configparser.Error: return None def open_log(args): # if nothing piped to stdin and no path supplied if not args.logpath and sys.stdin.isatty(): query_log = default_log_path() if not query_log: print( "Failed to get slow query log path from /etc/my.cnf", file=sys.stderr, ) sys.exit(1) if not args.quiet: print( f"Reading from the default log file, `{query_log}'", file=sys.stderr, ) return open(query_log, encoding='utf-8', errors='replace') # if something piped to stdin with no path supplied, or explicitly sent - if not args.logpath or args.logpath == '-': query_log = sys.stdin if not args.quiet: print( "MySQL slow query log parser reading from stdin/pipe...", file=sys.stderr, ) return sys.stdin return open(args.logpath, encoding='utf-8', errors='replace') def iter_log(args): try: with open_log(args) as query_log: while line := query_log.readline(): yield line except OSError as exc: sys.exit(exc) def main(): args = parse_args() if args.output: # if we've specified an output file out_file = open(args.output, "w", encoding='utf-8') else: out_file = sys.stdout with out_file: # init user and id dictionaries user_table: dict[str, MySQLUser] = {} this_user = "NO_SUCH_USER" for line in iter_log(args): if user_match := USER_RE.match(line): this_user = user_match.group(1) if args.user and this_user == args.user: print(line, end=' ', file=out_file) elif stats_match := STATS_RE.match(line): if this_user not in user_table: user_table[this_user] = MySQLUser(this_user) user_table[this_user].add_query(stats_match) if args.user and this_user == args.user: print(line, end=' ', file=out_file) elif args.user and this_user == args.user: try: print(line, end='', file=out_file) except Exception: sys.exit(0) if args.user: return if not args.no_header: if args.average: MySQLUser.avg_header(out_file) else: MySQLUser.row_header(out_file) for data in sorted(user_table.values(), key=lambda x: x.num_queries): if args.average: data.avg_print(out_file) else: data.row_print(out_file) if __name__ == '__main__': main()