Mysql巡检脚本

Mysql巡检脚本

环境说明

mysql8.0+

python 2.7

yum install python-pip -y
yum install python-devel
yum install MySQL-python -y

#!/usr/bin/env python
# -*- coding: utf-8 -*-#
import sys
import time
from datetime import datetime
import MySQLdb

class DBUtil:
    def __init__(self, user=None, passwd=None, host=None, port=None, db=None):
        self.user = "monitor"
        self.passwd = "passwordxxxxxxxxxx"
        self.host = "127.0.0.1"
        self.port = 3306
        self.db = "information_schema"
        self._conn = None
        self._cursor = None

    def __enter__(self):
        self._conn = MySQLdb.connect(host=self.host, port=self.port, user=self.user, passwd=self.passwd, db=self.db)
        self._cursor = self._conn.cursor()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._conn.close()
        self._conn = None

    def check_table_size(self):
        try:
            sql = "select table_schema,table_name,concat(round((data_length+index_length)/1024/1024,2),'M') FROM 
            information_schema.tables where (DATA_LENGTH+INDEX_LENGTH) > 10*1024*1024*1024  and table_schema not in 
('information_schema','mysql','performance_schema','sys')"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查超过10G大小的表')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s size: %s") % (row[0], row[1], row[2]))
        except Exception as e:
            raise (e)

    def check_table_index(self):
        try:
            sql = "select t1.name,t2.num from information_schema.INNODB_TABLES t1, (select table_id,count(*) as num from 
information_schema.INNODB_INDEXES group by table_id having count(*) >=6) t2 where t1.table_id =t2.table_id"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查索引超过6个的表')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s index_num: %s") % (row[0].split('/')[0], row[0].split('/')[1], row[1]))
        except Exception as e:
            raise (e)

    def check_table_fragment_pct(self):
        try:
            sql = "SELECT TABLE_SCHEMA as `db`, TABLE_NAME as `tbl`,   
1-(TABLE_ROWS*AVG_ROW_LENGTH)/(DATA_LENGTH + INDEX_LENGTH + DATA_FREE) AS `fragment_pct`   
FROM information_schema.TABLES WHERE  TABLE_SCHEMA not in ('information_schema','mysql','performance_schema','sys') 
and (1-(TABLE_ROWS*AVG_ROW_LENGTH)/(DATA_LENGTH + INDEX_LENGTH + DATA_FREE)) > 0.5 and (DATA_LENGTH + INDEX_LENGTH + DATA_FREE) > 1024*1024*1024 ;"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查碎片率超过50%的表')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s fragment_pct: %s") % (row[0], row[1], row[2]))
        except Exception as e:
            raise (e)

    def check_table_rows(self):
        try:
            sql = "select table_schema,table_name,table_rows from 
information_schema.TABLES where table_schema not in ('information_schema','mysql','performance_schema','sys') 
and table_rows > 10000000 order by table_rows desc;"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查行数超过1000万行的表')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s rows: %s") % (row[0], row[1], row[2]))
        except Exception as e:
            raise (e)

    def check_table_chaset(self):
        try:
            self._cursor.execute("show variables like 'character_set_server';")
            default_charset = str(self._cursor.fetchone()[1])
            default_charset = default_charset + "_general_ci"
            sql = "select table_schema,table_name,table_collation from information_schema.tables where table_schema not 
in ('information_schema','mysql','performance_schema','sys') and table_collation !='" + default_charset + "';"
            result = self._cursor.fetchall()
            print('检查非默认字符集的表')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s fragment_pct: %s") % (row[0], row[1], row[2]))
        except Exception as e:
            raise (e)

    def check_table_big_columns(self):
        try:
            sql = "select table_schema,table_name,column_name,data_type from information_schema.columns where data_type in 
('blob','clob','text','medium text','long text') and table_schema not in 
('information_schema','performance_schema','mysql','sys')"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查含大字段的表(blob,clob,text,medium text,long text)')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s column_name: %s data_type: %s") % (row[0], row[1], row[2], row[3]))
        except Exception as e:
            raise (e)

    def check_table_long_varchar(self):
        try:
            sql = "select table_schema,table_name,column_name,data_type,CHARACTER_MAXIMUM_LENGTH from information_schema.columns 
where DATA_TYPE='varchar' and CHARACTER_MAXIMUM_LENGTH > 500 and table_schema not in 
('information_schema','performance_schema','mysql','sys');"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查字段长度过大的表(>500)')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s column_name: %s data_type: %s(%s)") % (
                    row[0], row[1], row[2], row[3], row[4]))
        except Exception as e:
            raise (e)

    def check_table_no_index(self):
        try:
            sql = "SELECT t.table_schema,t.table_name FROM information_schema.tables AS t LEFT JOIN 
(SELECT DISTINCT table_schema, table_name FROM information_schema.`KEY_COLUMN_USAGE` ) AS kt ON 
kt.table_schema=t.table_schema AND kt.table_name = t.table_name WHERE t.table_schema NOT IN 
('mysql', 'information_schema', 'performance_schema', 'sys') AND kt.table_name IS NULL;"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查无主键/索引的表')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s") % (row[0], row[1]))
        except Exception as e:
            raise (e)

    def check_index_redundant(self):
        try:
            sql = "select table_schema,table_name,redundant_index_name,redundant_index_columns  from 
            sys.schema_redundant_indexes group by table_schema,table_name,redundant_index_name,redundant_index_columns;"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查重复索引')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s redundant_index_name：%s redundant_index_columns:%s ") % (
                    row[0], row[1], row[2], row[3]))
        except Exception as e:
            raise (e)

    def check_index_columns(self):
        try:
            sql = "select s.table_schema,s.table_name,s.index_name,s.column_name from information_schema.STATISTICS s,
(select table_name,index_name,count(*) from information_schema.STATISTICS where table_schema not in 
('information_schema','performance_schema','mysql','sys') group by table_name,index_name having count(*)>5)t where 
s.table_name=t.table_name and s.index_name=t.index_name;"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查超过5个字段的联合索引')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s index_name：%s column_name:%s ") % (row[0], row[1], row[2], row[3]))
        except Exception as e:
            raise (e)

    def check_index_unused(self):
        try:
            sql = "select * from sys.schema_unused_indexes where object_schema not in ('information_schema','performance_schema','mysql','sys') ;"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查无用的索引')
            if not result:
                print("结果不存在")
            for row in result:
                print(("schema: %s tablename: %s indexname：%s") % (row[0], row[1], row[2]))
        except Exception as e:
            raise (e)

    def check_important_variables(self):
        print('检查重要参数')
        variables_list = ['version', 'innodb_buffer_pool_size', 'innodb_flush_log_at_trx_commit','innodb_flush_method',
                          'innodb_log_file_size', 'innodb_log_files_in_group', 'innodb_file_per_table','innodb_open_files','innodb_data_home_dir',
                          'innodb_max_dirty_pages_pct', 'sync_binlog', 'max_connections', 'sort_buffer_size','read_buffer_size','max_allowed_packet',
                          'table_open_cache', 'table_definition_cache','thread_cache_size','key_buffer_size','time_zone']

        for variable in variables_list:
            try:
                sql = ("show global variables like '%s'" % variable)
                self._cursor.execute(sql)
                result = self._cursor.fetchone()[1]
                print(('%s : %s') % (variable, result))
            except Exception as e:
                raise (e)

    def check_important_status(self):
        print('检查重要状态')
        status_list = ['Uptime', 'Opened_files', 'Opened_table_definitions', 'Opened_tables', 'Max_used_connections',
                       'Threads_created', 'Threads_connected', 'Aborted_connects', 'Aborted_clients',
                       'Table_locks_waited', 'Innodb_buffer_pool_wait_free', 'Innodb_log_waits',
                       'Innodb_row_lock_waits', 'Innodb_row_lock_time_avg', 'Binlog_cache_disk_use', 'Created_tmp_disk_tables']
        for status in status_list:
            try:
                sql = ("show global status like '%s'" % status)
                self._cursor.execute(sql)
                result = self._cursor.fetchone()[1]
                print(('%s : %s') % (status, result))
            except Exception as e:
                raise (e)
        self._cursor.execute("show engine innodb status")
        innodb_status = self._cursor.fetchall()
        innodb_status_format = str(innodb_status).split('\n')
        for item in innodb_status_format:
            if "Log sequence number" in item:
                logsequencenumber = item.split(' ')[3]
                print(('%s : %s') % ('Log sequence number', logsequencenumber))
            if "Log flushed up to" in item:
                logflushnumber = item.split(' ')[6]
                print(('%s : %s') % ('Log flushed up to', logflushnumber))
            if "Last checkpoint at" in item:
                checkpoint = item.split(' ')[4]
                print(('%s : %s') % ('Last checkpoint at', checkpoint))
            if "History list length" in item:
                historylength = item.split(' ')[3]
                print(('%s : %s') % ('historylength', historylength))


    def check_user_nopass(self):
        try:
            sql = "select user,host from mysql.user where authentication_string='';"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查无密码用户')
            if not result:
                print("结果不存在")
            for row in result:
                print(("user: %s host: %s") % (row[0], row[1]))
        except Exception as e:
            raise (e)

    def check_user_nowhere(self):
        try:
            sql = "select user,host from mysql.user where host='%';"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            print('检查%用户')
            if not result:
                print("结果不存在")
            for row in result:
                print(("user: %s host: %s") % (row[0], row[1]))
        except Exception as e:
            raise (e)

    def check_user_privileges(self):
        try:
            sql = "select user,host from mysql.user where user not in ('mysql.session','mysql.sys');"
            self._cursor.execute(sql)
            result = self._cursor.fetchall()
            user_list = []
            for row in result:
                user_list.append("'" + row[0] + "'" + "@" + "'" + row[1] + "'")
            print('检查用户权限')
            for user in user_list:
                sql = "show grants for %s;" % user
                # print(sql)
                self._cursor.execute(sql)
                result = self._cursor.fetchall()
                for row in result:
                    print(row[0])
        except Exception as e:
            raise (e)

if __name__ == '__main__':
    with DBUtil('user', 'password', 'hostip', 3306, 'information_schema') as client:
        client.check_table_size()
        client.check_table_index()
        client.check_table_fragment_pct()
        client.check_table_rows()
        client.check_table_chaset()
        client.check_table_big_columns()
        client.check_table_long_varchar()
        client.check_table_no_index()
        client.check_index_redundant()
        client.check_index_columns()
        client.check_index_unused()
        client.check_important_variables()
        client.check_important_status()
        client.check_user_nopass()
        client.check_user_nowhere()
        client.check_user_privileges()