大家好，我是云计算磊哥，从业20年的IT老鸟。IT架构师培训15年，总结了一套从入门到精通的全运维开发宝典手册。准备用300天时间写一套博文，手把手从安装软件讲起，从linux系统管理，shell脚本编程，mysql运维架构备份核心技术，Apache/nginx/tomcatWEB服务器管理，ansible自动化运维，redis集群哨兵，LVM/HAproxy/keepalived集群架构，rabbitMQ消息队列，docker&K8S集群资源管理，K8S自愈，K8S自动扩容，PYTHON编程基础，PYTHON自动化运维，从行业到产品，从过去到未来，从理论到操作，从视频到文档工具，100+篇系列文章一站式发布。从零基础入门到30k运维开发工程师岗位诸多就业问题。多方位全方面的给你讲清楚云计算这个行业该如何做。关注我。后续AI大模型开发课程更精彩。

---

Python自动化运维实战指南

前言

学完 Python 基础语法（变量、循环、条件、函数、模块、文件操作、异常处理）后，自动化运维是最实战、最落地的核心场景。

传统运维依靠人工敲命令、肉眼看日志、手动查服务器状态，效率低、易出错、漏告警。利用 Python 可以编写轻量化巡检脚本，实现服务器、数据库、各类中间件、容器服务的全自动监控、数据采集、异常判定，替代 80% 重复性运维工作。

本文档聚焦企业真实运维场景，包含：云服务器全指标巡检、MySQL数据库巡检、Docker/K8s/Redis/Nginx服务巡检

一、自动化运维核心优势

• 低门槛落地：无需高深框架，基础Python语法即可实现

• 标准化巡检：规避人工巡检漏检、错检、标准不统一问题

• 全天候值守：配合定时任务，实现7×24小时自动监控

• 异常可视化：自动判定资源超标、服务宕机、数据库异常

• 轻量化无依赖：脚本体积小、运行快，适配所有Linux云服务器

二、前置环境准备

2.1 依赖库安装

本次所有巡检脚本依赖通用运维Python库，执行一键安装：

pip install psutil pymysql docker kubernetes requests

• psutil：系统资源、进程、网卡流量监控核心库

• pymysql：MySQL数据库连接、数据查询

• docker：操作Docker容器状态

• kubernetes：简易获取K8s集群资源状态

2.2 通用规则说明

• 所有脚本内置资源阈值告警（CPU/内存/磁盘超阈值自动告警）

• 统一输出结构化巡检报告，区分【正常】【异常】

• 兼容 CentOS / Ubuntu 主流云服务器系统

• 所有代码可直接运行，无需二次修改核心逻辑

三、场景一：云服务器全维度自动化巡检

覆盖运维核心指标：CPU使用率、内存使用率、磁盘使用率、系统进程、网卡流量，是服务器日常运维最核心的巡检内容。

3.1 完整巡检脚本

import psutil
import time
from datetime import datetime

# ====================== 巡检阈值配置 ======================
CPU_THRESHOLD = 80    # CPU使用率阈值
MEM_THRESHOLD = 85    # 内存使用率阈值
DISK_THRESHOLD = 90   # 磁盘使用率阈值
# ========================================================================

def check_cpu():
    """CPU使用率巡检"""
    # 采集1秒内CPU平均使用率
    cpu_percent = psutil.cpu_percent(interval=1)
    status = "异常" if cpu_percent > CPU_THRESHOLD else "正常"
    return cpu_percent, status

def check_memory():
    """内存使用率巡检"""
    mem = psutil.virtual_memory()
    mem_percent = mem.percent
    used = f"{mem.used / 1024 / 1024 / 1024:.2f}GB"
    total = f"{mem.total / 1024 / 1024 / 1024:.2f}GB"
    status = "异常" if mem_percent > MEM_THRESHOLD else "正常"
    return total, used, mem_percent, status

def check_disk():
    """磁盘使用率巡检（根目录）"""
    disk = psutil.disk_usage("/")
    disk_percent = disk.percent
    used = f"{disk.used / 1024 / 1024 / 1024:.2f}GB"
    total = f"{disk.total / 1024 / 1024 / 1024:.2f}GB"
    status = "异常" if disk_percent > DISK_THRESHOLD else "正常"
    return total, used, disk_percent, status

def check_process(target_process):
    """指定进程巡检（判断进程是否存活）"""
    process_list = []
    for p in psutil.process_iter(["name"]):
        try:
            if target_process in p.info["name"]:
                process_list.append(p.pid)
        except (psutil.NoSuchProcess, psutil.AccessDenied):
            continue
    return len(process_list) > 0, process_list

def check_network_flow():
    """网卡流量巡检（实时收发总量）"""
    net = psutil.net_io_counters()
    send = f"{net.bytes_sent / 1024 / 1024:.2f}MB"
    recv = f"{net.bytes_recv / 1024 / 1024:.2f}MB"
    return send, recv

def server_inspect():
    """服务器综合巡检入口"""
    print("=" * 70)
    print(f"【云服务器全自动巡检报告】 巡检时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 70)

    # CPU巡检
    cpu_pct, cpu_status = check_cpu()
    print(f"\n[CPU资源] 使用率：{cpu_pct}%  状态：{cpu_status}")

    # 内存巡检
    mem_total, mem_used, mem_pct, mem_status = check_memory()
    print(f"[内存资源] 总量：{mem_total}  已用：{mem_used}  使用率：{mem_pct}%  状态：{mem_status}")

    # 磁盘巡检
    disk_total, disk_used, disk_pct, disk_status = check_disk()
    print(f"[磁盘资源] 总量：{disk_total}  已用：{disk_used}  使用率：{disk_pct}%  状态：{disk_status}")

    # 核心进程巡检（可自定义进程名：nginx/docker/mysql）
    proc_list = ["nginx", "docker", "mysqld"]
    print("\n[核心进程巡检]")
    for proc in proc_list:
        is_run, pid_list = check_process(proc)
        if is_run:
            print(f"  {proc} 运行中，进程PID：{pid_list}")
        else:
            print(f"  {proc} 未运行 ❌")

    # 网卡流量巡检
    send_flow, recv_flow = check_network_flow()
    print(f"\n[网卡流量] 累计发送：{send_flow}  累计接收：{recv_flow}")
    print("\n" + "=" * 70 + "\n巡检完成\n" + "=" * 70)

if __name__ == "__main__":
    server_inspect()

3.2 脚本功能说明

• CPU巡检：采集1秒瞬时使用率，超80%触发异常告警

• 内存巡检：统计总内存、已用内存、使用率，精准展示占用情况

• 磁盘巡检：检测服务器根目录磁盘占用，预防磁盘爆满宕机

• 进程巡检：自动检测Nginx、Docker、MySQL核心进程是否存活

• 网卡流量：统计开机至今全网卡收发流量，排查流量异常

四、场景二：MySQL数据库自动化巡检

数据库连通性检测、当前连接数统计、慢查询日志查询，覆盖日常DBA基础巡检场景。

4.1 完整巡检脚本

import pymysql
from datetime import datetime

# ====================== MySQL数据库配置 ======================
MYSQL_HOST = "127.0.0.1"
MYSQL_PORT = 3306
MYSQL_USER = "root"
MYSQL_PWD = "你的数据库密码"
# 连接数告警阈值
CONN_THRESHOLD = 200
# =======================================================================

def get_mysql_conn():
    """获取MySQL连接，检测连通性"""
    try:
        conn = pymysql.connect(
            host=MYSQL_HOST,
            port=MYSQL_PORT,
            user=MYSQL_USER,
            password=MYSQL_PWD,
            charset="utf8mb4"
        )
        return conn, True
    except Exception as e:
        print(f"MySQL连接失败：{str(e)}")
        return None, False

def check_mysql_connection():
    """1. 检测数据库连通性 + 当前连接数"""
    conn, is_ok = get_mysql_conn()
    if not is_ok:
        return False, 0
    
    cursor = conn.cursor()
    # 查询当前活跃连接数
    cursor.execute("show global status like 'Threads_connected';")
    res = cursor.fetchone()
    conn_count = int(res[1])

    status = "异常 连接数过高" if conn_count > CONN_THRESHOLD else "正常"
    cursor.close()
    conn.close()
    return True, conn_count, status

def check_slow_query():
    """2. 检测慢查询状态 + 慢查询总数"""
    conn, is_ok = get_mysql_conn()
    if not is_ok:
        return 0, False
    
    cursor = conn.cursor()
    # 查看慢查询开启状态
    cursor.execute("show variables like 'slow_query_log';")
    slow_status = cursor.fetchone()[1]
    # 查看慢查询总数
    cursor.execute("show global status like 'Slow_queries';")
    slow_num = int(cursor.fetchone()[1])

    cursor.close()
    conn.close()
    return slow_num, slow_status == "ON"

def mysql_inspect():
    """MySQL综合巡检入口"""
    print("=" * 70)
    print(f"【MySQL数据库巡检报告】 巡检时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 70)

    # 1. 连通性+连接数巡检
    conn_ok, conn_count, conn_status = check_mysql_connection()
    if conn_ok:
        print(f"\n[数据库连通性] 连接正常")
        print(f"[当前活跃连接数] {conn_count}  状态：{conn_status}")
    else:
        print("\n[数据库连通性] 连接失败")

    # 2. 慢查询巡检
    slow_num, slow_open = check_slow_query()
    print(f"\n[慢查询功能] 已开启：{'是' if slow_open else '否'}")
    print(f"[累计慢查询数量] {slow_num} 条")

    if slow_num > 10:
        print("  警告：存在大量慢查询，需优化SQL语句！")

    print("\n" + "=" * 70 + "\n数据库巡检完成\n" + "=" * 70)

if __name__ == "__main__":
    mysql_inspect()

4.2 核心巡检能力说明

• 连通性检测：快速判断数据库是否宕机、账号密码是否失效

• 连接数监控：防止连接数打满导致数据库拒绝新业务请求

• 慢查询统计：定位低效SQL，解决业务卡顿、接口超时问题

五、场景三：Docker / K8s / Redis / Nginx 服务巡检

针对运维高频中间件、容器服务，实现自动化状态检测，无需手动输入 docker ps、kubectl、redis-cli、nginx status 命令。

5.1 综合服务巡检脚本

import docker
import requests
import redis
from kubernetes import client, config
from datetime import datetime

# ====================== 服务配置 ======================
REDIS_HOST = "127.0.0.1"
REDIS_PORT = 6379
NGINX_STATUS_URL = "http://127.0.0.1/nginx-status"
# =====================================================

def check_docker():
    """Docker服务巡检：运行中容器数量、异常容器"""
    try:
        client = docker.from_env()
        containers = client.containers.list(all=True)
        run_count = len([c for c in containers if c.status == "running"])
        stop_count = len(containers) - run_count
        print(f"[Docker巡检] 总容器数：{len(containers)}  运行中：{run_count}  已停止：{stop_count}")
        return True
    except Exception as e:
        print(f"[Docker巡检] 服务异常❌：{str(e)}")
        return False

def check_k8s():
    """K8s简易巡检：节点状态、运行Pod数量"""
    try:
        # 加载K8s集群配置
        config.load_kube_config()
        v1 = client.CoreV1Api()
        # 统计正常节点
        nodes = v1.list_node()
        ready_nodes = 0
        for node in nodes.items:
            for cond in node.status.conditions:
                if cond.type == "Ready" and cond.status == "True":
                    ready_nodes += 1
                    break
        # 统计运行Pod
        pods = v1.list_pod_for_all_namespaces()
        run_pod = len([p for p in pods.items if p.status.phase == "Running"])
        print(f"[K8s巡检] 集群节点正常数：{ready_nodes}/{len(nodes.items)}  运行中Pod：{run_pod}")
        return True
    except Exception as e:
        print(f"[K8s巡检] 未配置集群/服务异常❌：{str(e)}")
        return False

def check_redis():
    """Redis连通性+状态巡检"""
    try:
        r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0, socket_timeout=3)
        r.ping()
        info = r.info()
        print(f"[Redis巡检] 连接正常✅  在线客户端：{info['connected_clients']}  内存使用：{info['used_memory_human']}")
        return True
    except Exception as e:
        print(f"[Redis巡检] 连接失败❌：{str(e)}")
        return False

def check_nginx():
    """Nginx运行状态巡检"""
    try:
        res = requests.get(NGINX_STATUS_URL, timeout=3)
        if res.status_code == 200:
            print("[Nginx巡检] 服务运行正常✅")
            return True
        else:
            print(f"[Nginx巡检] 服务异常❌ 状态码：{res.status_code}")
            return False
    except Exception as e:
        print(f"[Nginx巡检] 访问失败❌：{str(e)}")
        return False

def service_inspect():
    """综合服务巡检入口"""
    print("=" * 70)
    print(f"【中间件/容器服务巡检报告】 时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 70)

    check_docker()
    check_k8s()
    check_redis()
    check_nginx()

    print("\n" + "=" * 70 + "\n服务巡检完成\n" + "=" * 70)

if __name__ == "__main__":
    service_inspect()

5.2 各服务巡检核心说明

1. Docker 巡检

自动统计全部容器、运行中容器、停止容器数量，快速排查容器意外退出、服务挂掉问题，替代手动 docker ps -a 命令。

2. K8s 集群巡检

自动检测集群正常节点数、运行中Pod数量，快速判断集群健康状态，适合小型集群日常巡检。无K8s环境不报错，仅提示异常。

3. Redis 巡检

检测Redis连通性、在线客户端连接数、内存占用，排查Redis宕机、连接爆满、内存溢出问题。

4. Nginx 巡检

通过Nginx状态页检测服务可用性，判断反向代理、网关服务是否正常运行，及时发现Nginx挂死、端口监听异常问题。

六、自动化运维进阶拓展方案

6.1 定时自动巡检

配合Linux crontab定时任务，实现每日/每小时自动巡检，无需人工执行脚本：

# 每小时执行一次服务器巡检
0 * * * * /usr/bin/python3 /root/server_inspect.py

6.2 告警升级

可基于现有脚本二次开发，实现异常自动推送：钉钉机器人、企业微信、邮件告警，实现故障秒级发现。

6.3 日志持久化

修改脚本，将每次巡检结果写入日志文件，实现运维数据留存，方便故障回溯、趋势分析。