## ================================================================================== ## 让读书成为一种生活方式。就像吃喝拉撒每天必须要干的事， ## 终有一天你的举止、言谈、气质会不一样。 ## —- async ## ## Created Date: Sunday, 2021-03-25, 9:48:04 am ## copyright (c): WHHL Tech. LTD. ## Engineer: async ## Module Name: ## Revision: v0.01 ## Description: ## ## Revision History : ## Revision editor date Description ## v0.01 async 2021-03-25 File Created ## ================================================================================== #import click import mysql.connector import requests,datetime,json # # @click.command() # @click.option('-u', '--username', required=True) # @click.option('-h', '--host', required=True) # @click.option('--dry-run', is_flag=True) # @click.password_option('-p', '--password', confirmation_prompt=False) # @click.argument('statement', nargs=-1) def run(username, password, host): """ 定义连接数据库执行动作 """ cnx = mysql.connector.connect(user=username,password=password,host=host) cursor = cnx.cursor(buffered=True) cursor.execute("SELECT ID,STATE,COMMAND,lower(INFO) FROM INFORMATION_SCHEMA.PROCESSLIST WHERE STATE in ('executing','Sending data') and time>50;") kinfo=cursor.fetchall() for _id, row_state, command, info in kinfo: kill_cs = cnx.cursor() # kill_cs.execute("KILL {};".format(_id)) kill_cs.execute("CALL mysql.rds_kill({session_id});".format(session_id=_id)) kill_cs.close() cursor.close() cnx.close() # print(kinfo) return kinfo def get_host(x): """ 定义dblist """ db_list=[ # ('10.10.xxx,','user','xx'), ('mysql2', 'user1', 'pawd1'), ('myql3','user2','paswd2'), .... ##可以定义很多 ] db_ip = [i[0] for i in db_list] db_ipx=db_ip[x] db_user=[i[1] for i in db_list] db_userx=db_user[x] db_pwd= [i[2] for i in db_list] db_pwdx = db_pwd[x] return db_ipx,db_userx,db_pwdx def db_auth(q): """ 获取数据库ip,用户名，密码 """ a=get_host(q) host1=str(a[0].strip(',')) user1=str(a[1].strip(',')) passwd1=str(a[2].strip(',')) return host1,user1,passwd1 def get_message(title, content): """ 定义告警内容格式 """ message = "AWS故障: " + "<font color=#FF0000 size=3>aws rds存在慢sql </font>" + " >报警主题：" + "当前数据库为 : "+ title + " >报警时间：" + str(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')) + " >详细信息：" + "exe sql is : " + content return message def send_message(message): ## 钉钉组中创建机器人的时候给出的webhook webhook = "https://oapi.dingtalk.com/robot/send?access_token=xxxxx" ## 调用request.post发送json格式的参数 headers = {'Content-Type': 'application/json'} data = { "msgtype": "markdown", "markdown": { "title": "AWS告警" + "....", "text": message }, "at": { "atMobiles": [xxxxx], "isAtAll": False } } requests.post(url=webhook, data=json.dumps(data), headers=headers) def get_cont(): followquant = db_auth(0) copytrading = db_auth(1) ## copytrading push = db_auth(2) ##push ...... res_followquant = run(host=followquant[0], username=followquant[1], password=followquant[2]) res_copytrading = run(host=copytrading[0], username=copytrading[1], password=copytrading[2]) ...... followquant_tuple = [list(item) for item in res_followquant].__str__() copytrading_tuple = [list(item) for item in res_copytrading].__str__() ......
if len(followquant_tuple) !=2: send_message(message=get_message(title=followquant[0], content=followquant_tuple)) if len(copytrading_tuple) !=2: send_message(message=get_message(title=copytrading[0], content=copytrading_tuple)) ...... start_time = datetime.datetime.now() #if __name__ == "__main__": def lambda_handler(event, context): send_message(message=get_cont()) end_time= datetime.datetime.now() time_cost = end_time - start_time print("当前脚本运行耗时为: " + str(time_cost).split('.')[0])
然后再cloudwatch加到计划任务，一分钟执行一次。