运维开发宝典042-Python自动化运维实战6
大家好，我是云计算磊哥，从业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库，执行一键安装：
pipinstallpsutil pymysqldockerkubernetes requestspsutil：系统资源、进程、网卡流量监控核心库
pymysql：MySQL数据库连接、数据查询
docker：操作Docker容器状态
kubernetes：简易获取K8s集群资源状态
2.2 通用规则说明
所有脚本内置资源阈值告警（CPU/内存/磁盘超阈值自动告警）
统一输出结构化巡检报告，区分【正常】【异常】
兼容 CentOS / Ubuntu 主流云服务器系统
所有代码可直接运行，无需二次修改核心逻辑
三、场景一：云服务器全维度自动化巡检
覆盖运维核心指标：CPU使用率、内存使用率、磁盘使用率、系统进程、网卡流量，是服务器日常运维最核心的巡检内容。
3.1 完整巡检脚本
importpsutilimporttimefromdatetimeimportdatetime# ====================== 巡检阈值配置 ======================CPU_THRESHOLD=80# CPU使用率阈值MEM_THRESHOLD=85# 内存使用率阈值DISK_THRESHOLD=90# 磁盘使用率阈值# ========================================================================defcheck_cpu():"""CPU使用率巡检"""# 采集1秒内CPU平均使用率cpu_percent=psutil.cpu_percent(interval=1)status="异常"ifcpu_percent>CPU_THRESHOLDelse"正常"returncpu_percent,statusdefcheck_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="异常"ifmem_percent>MEM_THRESHOLDelse"正常"returntotal,used,mem_percent,statusdefcheck_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="异常"ifdisk_percent>DISK_THRESHOLDelse"正常"returntotal,used,disk_percent,statusdefcheck_process(target_process):"""指定进程巡检（判断进程是否存活）"""process_list=[]forpinpsutil.process_iter(["name"]):try:iftarget_processinp.info["name"]:process_list.append(p.pid)except(psutil.NoSuchProcess,psutil.AccessDenied):continuereturnlen(process_list)>0,process_listdefcheck_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"returnsend,recvdefserver_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[核心进程巡检]")forprocinproc_list:is_run,pid_list=check_process(proc)ifis_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 完整巡检脚本
importpymysqlfromdatetimeimportdatetime# ====================== MySQL数据库配置 ======================MYSQL_HOST="127.0.0.1"MYSQL_PORT=3306MYSQL_USER="root"MYSQL_PWD="你的数据库密码"# 连接数告警阈值CONN_THRESHOLD=200# =======================================================================defget_mysql_conn():"""获取MySQL连接，检测连通性"""try:conn=pymysql.connect(host=MYSQL_HOST,port=MYSQL_PORT,user=MYSQL_USER,password=MYSQL_PWD,charset="utf8mb4")returnconn,TrueexceptExceptionase:print(f"MySQL连接失败：{str(e)}")returnNone,Falsedefcheck_mysql_connection():"""1. 检测数据库连通性 + 当前连接数"""conn,is_ok=get_mysql_conn()ifnotis_ok:returnFalse,0cursor=conn.cursor()# 查询当前活跃连接数cursor.execute("show global status like 'Threads_connected';")res=cursor.fetchone()conn_count=int(res[1])status="异常 连接数过高"ifconn_count>CONN_THRESHOLDelse"正常"cursor.close()conn.close()returnTrue,conn_count,statusdefcheck_slow_query():"""2. 检测慢查询状态 + 慢查询总数"""conn,is_ok=get_mysql_conn()ifnotis_ok:return0,Falsecursor=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()returnslow_num,slow_status=="ON"defmysql_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()ifconn_ok:print(f"\n[数据库连通性] 连接正常")print(f"[当前活跃连接数]{conn_count}状态：{conn_status}")else:print("\n[数据库连通性] 连接失败")# 2. 慢查询巡检slow_num,slow_open=check_slow_query()print(f"\n[慢查询功能] 已开启：{'是'ifslow_openelse'否'}")print(f"[累计慢查询数量]{slow_num}条")ifslow_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 综合服务巡检脚本
importdockerimportrequestsimportredisfromkubernetesimportclient,configfromdatetimeimportdatetime# ====================== 服务配置 ======================REDIS_HOST="127.0.0.1"REDIS_PORT=6379NGINX_STATUS_URL="http://127.0.0.1/nginx-status"# =====================================================defcheck_docker():"""Docker服务巡检：运行中容器数量、异常容器"""try:client=docker.from_env()containers=client.containers.list(all=True)run_count=len([cforcincontainersifc.status=="running"])stop_count=len(containers)-run_countprint(f"[Docker巡检] 总容器数：{len(containers)}运行中：{run_count}已停止：{stop_count}")returnTrueexceptExceptionase:print(f"[Docker巡检] 服务异常❌：{str(e)}")returnFalsedefcheck_k8s():"""K8s简易巡检：节点状态、运行Pod数量"""try:# 加载K8s集群配置config.load_kube_config()v1=client.CoreV1Api()# 统计正常节点nodes=v1.list_node()ready_nodes=0fornodeinnodes.items:forcondinnode.status.conditions:ifcond.type=="Ready"andcond.status=="True":ready_nodes+=1break# 统计运行Podpods=v1.list_pod_for_all_namespaces()run_pod=len([pforpinpods.itemsifp.status.phase=="Running"])print(f"[K8s巡检] 集群节点正常数：{ready_nodes}/{len(nodes.items)}运行中Pod：{run_pod}")returnTrueexceptExceptionase:print(f"[K8s巡检] 未配置集群/服务异常❌：{str(e)}")returnFalsedefcheck_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']}")returnTrueexceptExceptionase:print(f"[Redis巡检] 连接失败❌：{str(e)}")returnFalsedefcheck_nginx():"""Nginx运行状态巡检"""try:res=requests.get(NGINX_STATUS_URL,timeout=3)ifres.status_code==200:print("[Nginx巡检] 服务运行正常✅")returnTrueelse:print(f"[Nginx巡检] 服务异常❌ 状态码：{res.status_code}")returnFalseexceptExceptionase:print(f"[Nginx巡检] 访问失败❌：{str(e)}")returnFalsedefservice_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.py6.2 告警升级
可基于现有脚本二次开发，实现异常自动推送：钉钉机器人、企业微信、邮件告警，实现故障秒级发现。
6.3 日志持久化
修改脚本，将每次巡检结果写入日志文件，实现运维数据留存，方便故障回溯、趋势分析。