python如何监控PostgreSQL代码运行,,如何监控Postgr


如何监控PostgreSQL存储过程/函数代码运行?本文介绍用python+微信/邮件的方式进行报警、监控。

首先要有一张表、用于存放PostgreSQL存储过程/函数代码运行异常的信息。

技术图片

处理原则:若出现异常;把“发生时间+所在的程序+原因”通过微信/邮件发给对应人员。当然发送一次即可;起到通知的效果。

一、媒介

通过什么方式进行发送内容;下面介绍微信/邮件两种方式

1、python发送微信

py_wechar.py的内容

企业微信号;大家可以到企业微信上配置

#!/usr/bin/python3#coding=utf-8import jsonimport timeimport urllib.request as urllib2options = {? ? ‘WeiXin‘: {? ? ? ? ? ? ‘corp_id‘: ‘*‘,? #微信企业号ID? ? ? ? ? ? ‘agent_id‘: ‘*‘, #微信企业号应用ID? ? ? ? ? ? ‘agent_secret‘: ‘*‘,? #微信企业号密钥? ? ? ? ? ? ‘to_user‘: ‘@all‘? #发送给谁? ? },}class WeiXinSendMsg:? ? def __init__(self, wx_conf):? ? ? ? self.corp_id = wx_conf.get(‘corp_id‘)? ? ? ? self.agent_secret = wx_conf.get(‘agent_secret‘)? ? ? ? self.agent_id = wx_conf.get(‘agent_id‘)? ? ? ? self.to_user = wx_conf.get(‘to_user‘)? ? ? ? self.token = self.get_token()?? ? ? ? self.token_update_time = int(time.time())? ? ? ??? ? def get_token(self):? ? ? ? get_token_url = ‘https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=‘ + self.corp_id + ‘&corpsecret=‘ + self.agent_secret? ? ? ? token = json.loads(urllib2.urlopen(get_token_url).read().decode(‘utf-8‘))[‘access_token‘]? ? ? ? if token:? ? ? ? ? ? return token? ? # 微信发送端的token每1800秒会更新一次? ? def update_token(self):? ? ? ? if int(time.time()) - self.token_update_time >= 1800:? ? ? ? ? ? self.token = self.get_token()? ? ? ? ? ? self.token_update_time = int(time.time())? ? def send_message(self, msg):? ? ? ? try:? ? ? ? ? ? self.update_token()? ? ? ? ? ? send_url = ‘https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=‘ + self.token? ? ? ? ? ? send_val = {"touser":self.to_user, "toparty":"", "msgtype":"text", "agentid":self.agent_id, "text":{"content":msg}, "safe":"0"}? ? ? ? ? ? send_data = json.dumps(send_val, ensure_ascii=True).encode("utf-8")? ? ? ? ? ? send_request = urllib2.Request(send_url, send_data)? ? ? ? ? ? response = json.loads(urllib2.urlopen(send_request).read())? ? ? ? except Exception as e:? ? ? ? ? ? print(‘Exception WeiXin send_message:‘, e)if __name__ == ‘__main__‘:? ? WeiXin = WeiXinSendMsg(options.get(‘WeiXin‘))? ? WeiXin.send_message(‘hello world / 测试‘)

2、python发送邮件

py_email.py的内容

#!/usr/bin/python3#coding=utf-8import smtplibfrom email.header import Headerfrom email.mime.text import MIMETextfrom email.utils import parseaddr, formataddroptions = {? ? ‘Email‘: {? ? ? ? ‘smtp_server‘: ‘smtp.exmail.qq.com‘,? #邮箱服务器地址? ? ? ? ‘from_addr‘: ‘monitor@qq.com‘,? #发送人账号? ? ? ? ‘password‘: ‘123456‘, #发送人密码? ? ? ? ‘to_addr‘: [‘hanbo@126.com‘, ‘hanbo@163.com‘], #发送给谁? ? }}class EmailSendMsg:? ? def __init__(self, email_conf):? ? ? ? self.smtp_server = email_conf.get(‘smtp_server‘)? ? ? ? self.from_addr = email_conf.get(‘from_addr‘)? ? ? ? self.password = email_conf.get(‘password‘)? ? ? ? self.to_addr = email_conf.get(‘to_addr‘)? ? # def __del__(self):? ? #? ? ?self.server.quit()? ? def format_addr(self, str):? ? ? ? name, addr = parseaddr(str)? ? ? ? return formataddr(( ? ? ? ? ? ? Header(name, ‘utf-8‘).encode(), ? ? ? ? ? ? addr.encode(‘utf-8‘) if isinstance(addr, unicode) else addr))? ??? ? def send_msg(self, text):? ? ? ? try:? ? ? ? ? ? self.server = smtplib.SMTP(self.smtp_server, 25)? ? ? ? ? ? self.server.set_debuglevel(1)? ? ? ? ? ? self.server.login(self.from_addr, self.password)? ? ? ? ? ? msg = MIMEText(text, ‘plain‘, ‘utf-8‘)? ? ? ? ? ? msg[‘From‘] = self.format_addr(u‘监控 <%s>‘ % self.from_addr)? ? ? ? ? ? for i in range(len(self.to_addr)):? ? ? ? ? ? ? ? msg[‘To‘] = self.format_addr(u‘<%s>‘ % self.to_addr[i])? ? ? ? ? ? msg[‘Subject‘] = Header(u‘异常报警…‘, ‘utf-8‘).encode()? ? ? ? ? ? self.server.sendmail(self.from_addr, self.to_addr, msg.as_string())? ? ? ? ? ? self.server.quit()? ? ? ? except Exception as e:? ? ? ? ? ? print ‘Exception Email send_message:‘, eif __name__ == ‘__main__‘:Email = EmailSendMsg(options.get(‘Email‘))Email.send_msg(‘hello world!‘)

二、python连接数据库

PostgreSQL Python

看这个链接可以研究下python如何连接PostgreSQL数据库

三、python报警

上面我们知道如何通过python发送微信内容、以及python连接PostgreSQL数据库。现在我们要如何获取报警时机;报警内容。

python_alert.py

#!/usr/bin/python3?import psycopg2from config import configfrom py_wechar import WeiXinSendMsg,optionsdef get_errors():? ? """ query data from the vendors table """? ? conn = None? ? try:? ? ? ? params = config()? ? ? ? WeiXin = WeiXinSendMsg(options.get(‘WeiXin‘))? ? ? ? conn = psycopg2.connect(**params)? ? ? ? cur = conn.cursor()? ? ? ? cur.execute("select error_time, error_desc, proc_name from adsas.tbl_error_log where deal_status = 0 order by id")? ? ? ? rows = cur.fetchall()? ? ? ? if cur.rowcount > 0 :? ? ? ? ? ? WeiXin.send_message("The number of parts: {}".format(cur.rowcount))? ? ? ? ? ? for row in rows:? ? ? ? ? ?# WeiXin.send_message(‘-‘*60)? ? ? ? ? ?# WeiXin.send_message(‘发生时间:{}‘.format(row[0]))? ? ? ? ? ?# WeiXin.send_message(‘错误原因:{}‘.format(row[1]))? ? ? ? ? ?# WeiXin.send_message(‘报警代码:{}‘.format(row[2]))? ? ? ? ? ? ? ? str_error=‘发生时间:{}\n错误原因:{}\n报警代码:{}‘.format(row[0],row[1],row[2])? ? ? ? ? ? ? ? WeiXin.send_message(str_error)? ? ? ? ? ? cur.execute("update adsas.tbl_error_log set deal_status = 1 where deal_status = 0 ")? ? ? ? conn.commit()? ? ? ? cur.close()? ? except (Exception, psycopg2.DatabaseError) as error:? ? ? ? print(error)? ? finally:? ? ? ? if conn is not None:? ? ? ? ? ? conn.close()if __name__ == ‘__main__‘:? ? get_errors()

四、部署

可以通过cron/或者开源的定时任务系统进行报警;

报警信息:
技术图片

python如何监控PostgreSQL代码运行

评论关闭