Python学习笔记之IP监控及告警

news2025/2/12 11:07:03

一、需求说明

作为一名运维工程师，监控系统必不可少。不过我们的监控系统往往都是部署在内网的，如果互联网出口故障，监控系统即使发现了问题，也会告警不出来，这个时候我们就需要补充监控措施，增加从外到内的监控，主要监控互联网出口IP地址的通断性。监控需求如下：
1、可以根据需要自定义监控IP地址；
2、监控可达秒级，比如间隔3秒，5次不通就告警；
3、可以通过邮件进行告警；
4、监控记录日志存档。

二、监控脚本实现思路

本次需求实现使用Python语言，Python的安装可以参考博文Ubuntu之Anaconda3安装，这里不再赘述。

1、确定模块依赖

日志记录和格式化输出需要使用logging库；系统日志目录获取等需要用到os模块；监控使用ping，需要用到ping模块；邮件告警需要用到smtplib。

2、定义监控参数和邮件参数

主要是用配置待监控IP地址，ping检查间隔，不可达次数，邮件参数等。

3、日志文件检查和定义日志格式

检查日志文件目录是否存在，不存在则创建。

3、定义邮件告警发生函数

使用Python中的smptlib库创建邮件告警函数，调用上述步骤参数中的信息。

4、定义监控函数

这个是最重要的函数，主要是创建一个true循环，通过ping记录响应时间，超时则认为不可达，连续超过5次则发送告警邮件。

初始化不可达次数为0
获取日期，创建日志文件
ping检查目标地址，获取响应时间
如果不可达则不可达次数+1
如果不可达次数超过参数定义次数则发生告警邮件
如果ping检测结果正常不可达次数归零
服务暂停监控间隔时间后继续

5、定义主函数

三、监控脚本内容

# -*- coding: utf-8 -*
import time
import logging
import os
from datetime import datetime
from ping3 import ping
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

# 监控及邮件参数
TARGET_IP = "36.158.39.200"  # 待监控IP
CHECK_INTERVAL = 3         # Ping监控间隔
UNREACHABLE_THRESHOLD = 5  # 不可达次数记录
LOG_DIR = "./checklog"  #日志存储目录
EMAIL_SENDER = "wuhongsheng@email.com"     # 发件邮箱
EMAIL_RECEIVER = "wuhongsheng@email.com" # 收件邮箱
SMTP_SERVER = "smtp.qiye.aliyun.com"        # SMTP邮件服务器地址
SMTP_PORT = 465                         # SMTP服务端口
EMAIL_USER = "wuhongsheng@email.com"       # 邮箱账户
EMAIL_PASSWORD = "Wuhs@20240603"             # 邮箱密码


# 日志检查和格式化
if not os.path.exists(LOG_DIR):
    os.makedirs(LOG_DIR)

logging.basicConfig(
    filename=os.path.join(LOG_DIR, f"{datetime.now().strftime('%Y-%m-%d')}.log"),
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)

# 定义邮件告警函数
def send_alert():
    msg = MIMEMultipart()
    msg['From'] = EMAIL_SENDER
    msg['To'] = EMAIL_RECEIVER
    msg['Subject'] = f"Ping Alert: {TARGET_IP} is Unreachable"
    body = f"Warning: {TARGET_IP} has been unreachable for {UNREACHABLE_THRESHOLD} consecutive attempts."
    msg.attach(MIMEText(body, 'plain'))
    
    with smtplib.SMTP_SSL(SMTP_SERVER, 465) as server:
        server.login(EMAIL_USER, EMAIL_PASSWORD)
        server.sendmail(EMAIL_SENDER, EMAIL_RECEIVER, msg.as_string())

# 定义监控函数
def monitor():
    unreachable_count = 0
    while True:
        #日志按天记录
        today = datetime.now().strftime("%Y-%m-%d")
        logging.root.handlers[0].baseFilename = os.path.join(LOG_DIR, f"{today}.log")
        
        # ping结果判断，记录连续中断次数
        response = ping(TARGET_IP, timeout=1)
        if response is None:
            unreachable_count += 1
            logging.warning(f"{TARGET_IP} is unreachable ({unreachable_count}/{UNREACHABLE_THRESHOLD})")
            if unreachable_count >= UNREACHABLE_THRESHOLD:
                send_alert()
                unreachable_count = 0  # 
        else:
            unreachable_count = 0
            latency = round(response * 1000, 2)  # 

            logging.info(f"{TARGET_IP} is reachable, latency: {latency} ms")
        
        time.sleep(CHECK_INTERVAL)

# 主函数
if __name__ == "__main__":
    monitor()