[MySQL] master_host= master_port=3306 master_user=root master_password= slave_host= [DingTalk] #生产 prod_webhook_url = https://oapi.dingtalk.com/robot/send?access_token= prod_secret= #测试 dev_webhook_url= https://oapi.dingtalk.com/robot/send?access_token= dev_secret= mobile_number= [Redis] redis_master_host= redis_master_port= 6379 redis_master_password = zk123 redis_slave_host= [Code] mobile_phone_number = signature_name = template_code = master_monitor_address= slave_monitor_address= [sms] phonenumbers = signname = templatecode =
[root@acs-hk-ctos7-prod-01 scripts]# cat monit_vip_redis.py #!/usr/bin/env python # -*- coding: utf-8 -*- # @Time : 2023/6/28 17:09 # @Author : zk_liunx # @File : keep_vip.py # @Software: PyCharm import os import time import re import redis import logging import configparser import hashlib import base64 import urllib import hmac import requests import datetime import socket logging.basicConfig(level=logging.INFO, filename='/server/scripts/log/keep_vip_drift_mon.log', filemode='a', format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s' ) def get_digest(secret, timestamp): secret_enc = secret.encode('utf-8') string_to_sign = '{}\n{}'.format(timestamp, secret) string_to_sign_enc = string_to_sign.encode('utf-8') hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return f"×tamp={timestamp}&sign={sign}" class DingTalkConfig: def __init__(self, config_file): self.config_file = config_file def get_config(self): config = configparser.ConfigParser() config.read(self.config_file) return config['DingTalk'] def get_keepalived_vip(): res = os.popen("ip add|grep 172.31.59.12") ip_address = re.search(r'\b(?:\d{1,3}\.){3}\d{1,3}\b', res.read()) if ip_address: keep_vip = ip_address.group() return keep_vip class RedisConfig: def __init__(self, config_file): self.config_file = config_file def get_config(self): config = configparser.ConfigParser() config.read(self.config_file) return config['Redis'] def send_dingtalk_notification(webhook_url, secret, mobile_number, title, msg): host_name = socket.gethostname() ip_address = socket.gethostbyname(host_name) timestamp = str(round(time.time() * 1000)) msg = "**告警主题**:" + "\n" + title + "\n\n" \ ">当前时间:" + "\n" + datetime.datetime.now().strftime( '%Y-%m-%d %H:%M:%S') + "\n\n" \ " >当前主机名:" + "\n" + host_name + "\n\n" \ " >当前服务器IP:" + "\n" + ip_address + "\n\n" \ "状态消息:" + "\n" + msg + "\n\n" data = { "msgtype": "markdown", "markdown": { "title": title, "text": msg, }, "at": { "atMobiles": [ mobile_number ], "isAtAll": False } } rec = requests.post(webhook_url + get_digest(secret, timestamp), json=data) class RedisReplicationChecker: def __init__(self, master_host, master_port, master_password, redis_slave_host): self.master_host = master_host self.master_port = master_port self.master_password = master_password self.slave_host = redis_slave_host def check_master_replication(self): # 获取主redis状态 master_connect_redis = redis.Redis(host=self.master_host, port=self.master_port, password=self.master_password) master_role_status = master_connect_redis.info()['role'] # logging.info("当前redis主从状态:{}".format(master_role_status)) return master_role_status def check_slave_replication(self): slave_connect_redis = redis.Redis(host=self.slave_host, port=self.master_port, password=self.master_password) slave_role_status = slave_connect_redis.info()['role'] # logging.info("当前redis从状态:{}".format(slave_role_status)) return slave_role_status class MonitorRedisVipMasterStatus(): ALERT_FILE = 'alert_status.txt' def get_master_slave_vip_status(self): redis_config = RedisConfig('config.ini').get_config() checker = RedisReplicationChecker( redis_config['redis_master_host'], int(redis_config['redis_master_port']), redis_config['redis_master_password'], redis_config['redis_slave_host'] ) master_status = checker.check_master_replication() slave_status = checker.check_slave_replication() res = get_keepalived_vip() return master_status, slave_status, res def webhookurl(self): dingtalk_config = DingTalkConfig('config.ini').get_config() webhook_url = dingtalk_config['prod_webhook_url'] secret = dingtalk_config['prod_secret'] mobile_number = dingtalk_config['mobile_number'] return webhook_url, secret, mobile_number def delete_first_line(self, file_name): with open(file_name, 'r') as file: logging.info("VIP switching notification status deletion") lines = file.readlines() if lines: with open(file_name, 'w') as writ_file: writ_file.write(lines.pop(0)) logging.info("VIP switch notification status has been deleted {}".format(lines)) def get_status(self): master_slave_vip_status = res.get_master_slave_vip_status() hook_secret_mobile = res.webhookurl() with open(MonitorRedisVipMasterStatus.ALERT_FILE, 'r') as file: status = file.read().strip() logging.info(master_slave_vip_status[2]) if not master_slave_vip_status[2]: logging.error("Master The current machine VIP does not exist {}.".format(master_slave_vip_status[2])) time.sleep(5) if master_slave_vip_status[0] == "slave" and master_slave_vip_status[1] == "master": logging.info("VIP switch successful") print(status) if status != "send": time.sleep(5) send_dingtalk_notification(hook_secret_mobile[0], hook_secret_mobile[1], hook_secret_mobile[2], "Redis VIP切换通知", '切换后从变更master当前状态:{},切换后从变更主,当前状态:{}'.format( master_slave_vip_status[1], [ master_slave_vip_status[0]])) with open(MonitorRedisVipMasterStatus.ALERT_FILE, 'w') as wr: wr.write('send' + "\n") else: logging.info("VIP switch notification has already been sent and is not being sent for the second time") elif master_slave_vip_status[0] == "master" and master_slave_vip_status[1] == "master": send_dingtalk_notification(hook_secret_mobile[0], hook_secret_mobile[1], hook_secret_mobile[2], "主从异常-Redis存在双主", '当前主Redis状态:{},从Redis状态{}'.format(master_slave_vip_status[0], [ master_slave_vip_status[1]])) logging.error("Redis synchronization exception, current status dual master") elif master_slave_vip_status[0] == "slave" and master_slave_vip_status[1] == "slave": send_dingtalk_notification(hook_secret_mobile[0], hook_secret_mobile[1], hook_secret_mobile[2], "主从异常-Redis存在双从", '当前主Redis状态:{},从Redis状态:{}'.format(master_slave_vip_status[0], [ master_slave_vip_status[1]])) logging.error("Redis synchronization is abnormal, and the current status is from Chinese New Year to Chinese New Year") else: logging.info("The current VIP has not been switched and remains in the check state") res.delete_first_line(MonitorRedisVipMasterStatus.ALERT_FILE) if __name__ == '__main__': res = MonitorRedisVipMasterStatus() res.get_status()
TRANSLATE with