python3监控批量服务器网络通畅的工具

在环境中对服务器的网络进行持续监控，判断网络是否通畅。常用的方方法是通过ping检测，ping不通时认为网络不通。如果服务器禁用了imcp报文这样的防ping操作，本工具不适用。

# -*- coding:utf-8 -*-
# @Time      :2020/05/09
# 通过ping检测判断指定Ip的网络是否通畅
# 根据IP的数量，启用携程池，持续检测每个IP
# 依赖ping3 模块 通过pip install ping3 安装
# 依赖日志模块loguru 通过pip install loguru 安装

import ping3
from loguru import logger
import time
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED

logger.add("net_check.log")


def ping_func(ip):
    while 1:
        # 调用ping3模块 ping超时1秒，超时时rsp_time未None
        rsp_time = ping3.ping(ip, timeout=1)
        if rsp_time is None:
            logger.error(f"HOST:{ip},ping detection failed")
        time.sleep(1)


def run_by_pool(ip_list):
    # 根据ip数启用线程池
    pool = ThreadPoolExecutor(max_workers=len(ip_list))
    task_list = []
    for ip in ip_list:
        task_list.append(pool.submit(ping_func, ip))
    # 等待所有线程完成
    wait(task_list, return_when=ALL_COMPLETED)


if __name__ == "__main__":
    ip_list = [
        '192.168.52.205',
        '192.168.52.206',
        '192.168.52.207',
        '192.168.52.211',
        '192.168.52.212',
        '192.168.52.213',
        '192.168.52.214',
        '192.168.52.215',
        '192.168.52.216',
        '192.168.52.217',
        '192.168.52.218',
        '192.168.52.219',
        '192.168.54.122',
    ]
    run_by_pool(ip_list)

检测到服务器网络不通时，会打印日志如下：