大家好，我是云计算磊哥，从业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使用率、内存使用率、磁盘使用率、系统进程、网卡流量，是服务器日常运维最核心的巡检内容。
3.1 完整巡检脚本
import psutilimport timefrom 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, statusdef 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, statusdef 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, statusdef 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_listdef 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, recvdef 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 pymysqlfrom datetime import datetime# ====================== MySQL数据库配置 ======================MYSQL_HOST = "127.0.0.1"MYSQL_PORT = 3306MYSQL_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, Falsedef 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, statusdef 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 dockerimport requestsimport redisfrom kubernetes import client, configfrom datetime import datetime# ====================== 服务配置 ======================REDIS_HOST = "127.0.0.1"REDIS_PORT = 6379NGINX_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 Falsedef 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 Falsedef 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 Falsedef 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 Falsedef 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 日志持久化
修改脚本，将每次巡检结果写入日志文件，实现运维数据留存，方便故障回溯、趋势分析。
10个月宝宝每天需要喝多少奶粉？