7月自动化运维脚本精选Shell与Python在生产环境中最具价值的10个效率工具集一、引言脚本能力是运维工程师的基础生产力在AI和自动化工具快速渗透运维领域的背景下一个常被忽视的事实是Shell脚本和Python工具依然承担着运维工作中60%以上的自动化任务。大模型可以生成脚本但理解脚本的设计逻辑、安全边界和异常处理模式仍然是运维工程师不可替代的核心能力。7月整理和优化了10个在生产环境中经过验证的高频脚本覆盖巡检、诊断、备份、监控等核心运维场景。每个脚本都经过生产验证包含完善的错误处理和中文注释。二、10个精选脚本分类详解以下Mermaid图展示了10个脚本在运维场景中的覆盖范围脚本一K8s集群健康度一键巡检#!/bin/bash # # 名称K8s集群健康度一键巡检脚本 # 功能检查节点状态、Pod异常、资源使用率、事件告警 # 用法./k8s_health_check.sh [--namespace default] [--output report.txt] # 作者运维团队 # 日期2026-07 # set -euo pipefail # ---- 参数解析 ---- NAMESPACE${NAMESPACE:-all} OUTPUT_FILE${OUTPUT_FILE:-/tmp/k8s_health_$(date %Y%m%d_%H%M%S).log} ALERT_THRESHOLD_NOTREADY0 ALERT_THRESHOLD_CPU80 ALERT_THRESHOLD_MEM80 log() { local level$1; shift echo [$(date %Y-%m-%d %H:%M:%S)] [${level}] $* | tee -a $OUTPUT_FILE } # ---- 检查kubectl可用性 ---- check_prerequisites() { if ! command -v kubectl /dev/null; then log ERROR kubectl未安装,请先安装kubernetes-cli exit 1 fi if ! kubectl cluster-info /dev/null; then log ERROR 无法连接到K8s集群,请检查kubeconfig配置 exit 1 fi log INFO 前置检查通过,开始巡检... } # ---- 1. 节点状态检查 ---- check_nodes() { log INFO 节点状态检查 local not_ready not_ready$(kubectl get nodes --no-headers 2/dev/null | \ grep -v Ready | wc -l | tr -d ) if [ $not_ready -gt $ALERT_THRESHOLD_NOTREADY ]; then log WARN 检测到 ${not_ready} 个节点处于非Ready状态: kubectl get nodes --no-headers 2/dev/null | grep -v Ready | \ awk {printf - %-40s %s\n, $1, $2} | tee -a $OUTPUT_FILE else log INFO 所有节点状态正常(Ready) fi # 检查节点资源使用率 log INFO 节点资源使用率Top 5: kubectl top nodes --no-headers 2/dev/null | sort -k3 -rn | head -5 | \ awk {printf %-30s CPU: %s MEM: %s\n, $1, $2, $4} | tee -a $OUTPUT_FILE } # ---- 2. Pod异常检查 ---- check_pods() { log INFO Pod状态检查 # 统计各类异常Pod数量 local failed_pods crash_pods pending_pods if [ $NAMESPACE all ]; then failed_pods$(kubectl get pods --all-namespaces --no-headers 2/dev/null | \ grep -vE Running|Completed | grep -v 0/ | wc -l | tr -d ) crash_pods$(kubectl get pods --all-namespaces --no-headers 2/dev/null | \ grep CrashLoopBackOff | wc -l | tr -d ) pending_pods$(kubectl get pods --all-namespaces --no-headers 2/dev/null | \ grep Pending | wc -l | tr -d ) else failed_pods$(kubectl get pods -n $NAMESPACE --no-headers 2/dev/null | \ grep -vE Running|Completed | grep -v 0/ | wc -l | tr -d ) crash_pods$(kubectl get pods -n $NAMESPACE --no-headers 2/dev/null | \ grep CrashLoopBackOff | wc -l | tr -d ) pending_pods$(kubectl get pods -n $NAMESPACE --no-headers 2/dev/null | \ grep Pending | wc -l | tr -d ) fi log INFO 异常Pod统计: Failed${failed_pods}, CrashLoop${crash_pods}, Pending${pending_pods} if [ $crash_pods -gt 0 ]; then log WARN 发现 ${crash_pods} 个处于CrashLoopBackOff的Pod: kubectl get pods --all-namespaces --no-headers 2/dev/null | \ grep CrashLoopBackOff | awk {printf %-20s %-40s %s\n, $1, $2, $3} | \ tee -a $OUTPUT_FILE fi } # ---- 3. 集群事件检查 ---- check_events() { log INFO 集群Warning事件检查(最近1小时) kubectl get events --all-namespaces --field-selector typeWarning \ --sort-by.lastTimestamp 2/dev/null | \ tail -20 | tee -a $OUTPUT_FILE } # ---- 4. PVC使用率检查 ---- check_pvc() { log INFO PVC使用率检查 # 使用kubectl-df插件(需要预先安装)检查PVC使用率 if command -v kubectl-df /dev/null; then kubectl df pv --all-namespaces 2/dev/null | \ awk -v threshold$ALERT_THRESHOLD_MEM { if (NR1) {print; next} split($5, pct, %) if (pct[1] threshold) { printf ⚠️ %s\n, $0 } else { print } } | tee -a $OUTPUT_FILE else log WARN kubectl-df未安装,跳过PVC使用率检查(安装:kubectl krew install df-pv) fi } # ---- 主流程 ---- main() { check_prerequisites check_nodes check_pods check_events check_pvc log INFO 巡检完成,报告已保存至: $OUTPUT_FILE } main $脚本二容器OOM自动分析工具#!/bin/bash # # 名称容器OOM自动分析脚本 # 功能分析被OOMKilled的容器,提取内存使用数据和日志 # 用法./oom_analyzer.sh namespace pod-prefix # 示例./oom_analyzer.sh production order-service # set -euo pipefail NAMESPACE${1:-} POD_PREFIX${2:-} if [ -z $NAMESPACE ] || [ -z $POD_PREFIX ]; then echo 用法: $0 namespace pod-prefix echo 示例: $0 production order-service exit 1 fi echo 容器OOM分析报告 echo 命名空间: ${NAMESPACE} echo Pod前缀: ${POD_PREFIX} echo 分析时间: $(date %Y-%m-%d %H:%M:%S) echo # 1. 检查Pod的OOMKilled状态 echo --- 1. OOMKilled历史 --- kubectl describe pod -n $NAMESPACE 2/dev/null | \ grep -A 20 Name:.*${POD_PREFIX} | \ grep -B 2 -A 10 OOMKilled || echo 未检测到OOMKilled记录 # 2. 获取当前Pod的资源使用情况 echo echo --- 2. 当前资源使用 --- kubectl top pod -n $NAMESPACE 2/dev/null | \ grep $POD_PREFIX || echo 无法获取资源使用数据 # 3. 分析节点的OOM事件 echo echo --- 3. 节点OOM事件 --- NODE_NAME$(kubectl get pod -n $NAMESPACE -o wide 2/dev/null | \ grep $POD_PREFIX | awk {print $7} | head -1) if [ -n $NODE_NAME ]; then echo Pod所在节点: ${NODE_NAME} kubectl describe node $NODE_NAME 2/dev/null | \ grep -A 20 Conditions: | grep -E MemoryPressure|DiskPressure|PIDPressure || true else echo 无法确定Pod所在节点 fi # 4. 提取最近的Pod日志(重点关注OOM相关) echo echo --- 4. 最近日志(OOM相关) --- POD_NAME$(kubectl get pod -n $NAMESPACE --no-headers 2/dev/null | \ grep $POD_PREFIX | grep Running | awk {print $1} | head -1) if [ -n $POD_NAME ]; then kubectl logs -n $NAMESPACE $POD_NAME --tail100 2/dev/null | \ grep -i -E oom|out of memory|heap|gc|memory || echo 未找到OOM相关日志 else echo 未找到运行中的Pod(可能已全部被OOMKilled) fi # 5. 内存使用建议 echo echo --- 5. 优化建议 --- MEMORY_LIMIT$(kubectl get pod -n $NAMESPACE $POD_NAME \ -o jsonpath{.spec.containers[0].resources.limits.memory} 2/dev/null || echo 未知) echo 当前内存限制: ${MEMORY_LIMIT} echo 建议: echo - 检查是否设置了JVM -Xmx(对于Java应用,-Xmx应设置为Limit的70-80%) echo - 检查是否有堆外内存泄漏(DirectByteBuffer/Native memory) echo - 考虑增加memory limit或优化应用程序内存使用 echo - 启用JVM的-XX:ExitOnOutOfMemoryError以便快速重启脚本三多数据库自动备份Python#!/usr/bin/env python3 多数据库自动备份工具 支持MySQL、PostgreSQL、MongoDB的定时备份含S3上传和过期清理 用法: python3 db_backup.py --config backup_config.yaml import subprocess import os import sys import logging import argparse from datetime import datetime, timedelta from pathlib import Path import boto3 from botocore.exceptions import ClientError import yaml # 配置日志 logging.basicConfig( levellogging.INFO, format[%(asctime)s] [%(levelname)s] %(message)s, handlers[ logging.FileHandler(f/var/log/db_backup_{datetime.now():%Y%m%d}.log), logging.StreamHandler(sys.stdout) ] ) logger logging.getLogger(__name__) class DatabaseBackup: 数据库备份管理器负责备份、压缩、上传和清理 def __init__(self, config_file: str): self.config self._load_config(config_file) self.backup_dir Path(self.config.get(backup_dir, /data/backups)) self.retention_days self.config.get(retention_days, 7) self.s3_bucket self.config.get(s3_bucket) self.backup_dir.mkdir(parentsTrue, exist_okTrue) def _load_config(self, config_file: str) - dict: 加载YAML配置文件 try: with open(config_file, r, encodingutf-8) as f: config yaml.safe_load(f) if not config: raise ValueError(f配置文件{config_file}为空) return config except FileNotFoundError: logger.error(f配置文件{config_file}不存在) sys.exit(1) except yaml.YAMLError as e: logger.error(f配置文件{config_file}格式错误: {e}) sys.exit(1) def backup_mysql(self, db_config: dict) - str: 备份MySQL数据库 Args: db_config: 数据库连接配置需包含host/port/user/password/database字段 Returns: str: 生成的备份文件路径 Raises: subprocess.CalledProcessError: mysqldump执行失败时抛出 host db_config[host] port db_config.get(port, 3306) user db_config[user] password db_config[password] database db_config[database] timestamp datetime.now().strftime(%Y%m%d_%H%M%S) backup_file self.backup_dir / fmysql_{database}_{timestamp}.sql.gz logger.info(f开始备份MySQL: {database}{host}:{port}) # mysqldump gzip管道备份避免落盘大文件占用磁盘空间 mysqldump_cmd [ mysqldump, --host, host, --port, str(port), --user, user, f--password{password}, --single-transaction, # 避免锁表保证InnoDB一致性 --routines, # 包含存储过程和函数 --triggers, # 包含触发器 --events, # 包含定时事件 --set-gtid-purgedOFF, # 防止GTID冲突 database ] gzip_cmd [gzip, -c] with open(backup_file, wb) as f_out: dump_proc subprocess.Popen( mysqldump_cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE ) gzip_proc subprocess.Popen( gzip_cmd, stdindump_proc.stdout, stdoutf_out, stderrsubprocess.PIPE ) dump_proc.stdout.close() # 关闭子进程的stdout引用 _, dump_err dump_proc.communicate() _, gzip_err gzip_proc.communicate() if dump_proc.returncode ! 0: error_msg dump_err.decode(utf-8, errorsreplace) logger.error(fmysqldump失败({database}): {error_msg}) backup_file.unlink(missing_okTrue) raise subprocess.CalledProcessError( dump_proc.returncode, mysqldump_cmd, stderrdump_err ) if gzip_proc.returncode ! 0: error_msg gzip_err.decode(utf-8, errorsreplace) logger.error(fgzip压缩失败: {error_msg}) backup_file.unlink(missing_okTrue) raise subprocess.CalledProcessError( gzip_proc.returncode, gzip_cmd, stderrgzip_err ) file_size_mb backup_file.stat().st_size / (1024 * 1024) logger.info(fMySQL备份完成: {backup_file.name} ({file_size_mb:.1f}MB)) return str(backup_file) def upload_to_s3(self, file_path: str) - bool: 上传备份文件到S3(MinIO/云存储) Args: file_path: 本地备份文件的完整路径 Returns: bool: 上传成功返回True失败返回False if not self.s3_bucket: logger.debug(未配置S3 bucket,跳过上传) return False try: s3_client boto3.client( s3, endpoint_urlself.config.get(s3_endpoint), aws_access_key_idself.config.get(s3_access_key), aws_secret_access_keyself.config.get(s3_secret_key) ) file_name os.path.basename(file_path) s3_key fdb_backups/{datetime.now():%Y/%m}/{file_name} s3_client.upload_file( file_path, self.s3_bucket, s3_key, ExtraArgs{StorageClass: STANDARD_IA} # 非频繁访问层节省成本 ) logger.info(fS3上传成功: s3://{self.s3_bucket}/{s3_key}) return True except ClientError as e: logger.error(fS3上传失败: {e}) return False except Exception as e: logger.error(fS3上传异常: {e}) return False def cleanup_old_backups(self): 清理超过保留期的本地备份文件 cutoff_date datetime.now() - timedelta(daysself.retention_days) deleted_count 0 freed_space 0 for backup_file in self.backup_dir.glob(*): if backup_file.is_file(): file_mtime datetime.fromtimestamp(backup_file.stat().st_mtime) if file_mtime cutoff_date: file_size backup_file.stat().st_size try: backup_file.unlink() deleted_count 1 freed_space file_size logger.debug(f清理过期备份: {backup_file.name}) except OSError as e: logger.warning(f清理文件失败 {backup_file}: {e}) freed_mb freed_space / (1024 * 1024) logger.info(f清理完成: 删除{deleted_count}个文件, 释放{freed_mb:.1f}MB空间) def run(self): 执行完整备份流程 databases self.config.get(databases, []) if not databases: logger.warning(未配置任何数据库跳过备份) return total len(databases) for idx, db_config in enumerate(databases, 1): db_type db_config.get(type, mysql) logger.info(f[{idx}/{total}] 处理数据库: {db_config.get(database, unknown)}) try: if db_type mysql: backup_file self.backup_mysql(db_config) elif db_type postgresql: backup_file self.backup_postgres(db_config) elif db_type mongodb: backup_file self.backup_mongodb(db_config) else: logger.error(f不支持的数据库类型: {db_type}) continue # 上传到对象存储 self.upload_to_s3(backup_file) except subprocess.CalledProcessError as e: logger.error(f数据库{db_config.get(database)}备份失败: {e}) # 继续处理下一个数据库不因单个失败而中断整个流程 continue except Exception as e: logger.error(f数据库{db_config.get(database)}备份异常: {e}, exc_infoTrue) continue # 清理过期备份 self.cleanup_old_backups() logger.info(所有备份任务完成) if __name__ __main__: parser argparse.ArgumentParser(description多数据库自动备份工具) parser.add_argument( --config, -c, requiredTrue, help备份配置文件路径(YAML格式) ) args parser.parse_args() backup DatabaseBackup(args.config) backup.run()脚本四SSL证书到期监控#!/usr/bin/env python3 SSL证书到期监控脚本 轮询检查指定域名列表的证书到期时间到期前30天触发告警 用法: python3 ssl_monitor.py --domains domains.txt --alert-days 30 import ssl import socket import sys import argparse import logging from datetime import datetime, timedelta from typing import List, Tuple logging.basicConfig( levellogging.INFO, format[%(asctime)s] [%(levelname)s] %(message)s ) logger logging.getLogger(__name__) def check_certificate(domain: str, port: int 443, timeout: int 10) - Tuple[str, int]: 检查指定域名的SSL证书到期时间 Args: domain: 域名如 example.com port: SSL端口默认443 timeout: 连接超时秒数 Returns: (domain, days_left) 元组days_left为证书剩余天数 Raises: socket.timeout: 连接超时 ConnectionRefusedError: 端口不可达 ssl.SSLError: SSL握手失败 context ssl.create_default_context() # 设置较短的握手超时避免长时间等待 context.check_hostname True with socket.create_connection((domain, port), timeouttimeout) as sock: with context.wrap_socket(sock, server_hostnamedomain) as ssock: cert ssock.getpeercert() # 解析证书到期时间(ASN.1格式) not_after datetime.strptime( cert[notAfter], %b %d %H:%M:%S %Y %Z ) days_left (not_after - datetime.now()).days return domain, max(0, days_left) def load_domains(file_path: str) - List[str]: 从文件加载域名列表每行一个域名忽略空行和注释 domains [] try: with open(file_path, r, encodingutf-8) as f: for line_no, line in enumerate(f, 1): line line.strip() # 跳过空行和注释行#开头 if not line or line.startswith(#): continue domains.append(line) if not domains: logger.warning(f文件{file_path}中未找到有效域名) except FileNotFoundError: logger.error(f域名文件{file_path}不存在) sys.exit(1) except IOError as e: logger.error(f读取域名文件{file_path}失败: {e}) sys.exit(1) logger.info(f从{file_path}加载了{len(domains)}个域名) return domains def main(): parser argparse.ArgumentParser(descriptionSSL证书到期监控) parser.add_argument(--domains, -d, requiredTrue, help域名列表文件) parser.add_argument(--alert-days, -a, typeint, default30, help提前多少天告警(默认30)) parser.add_argument(--port, -p, typeint, default443, helpSSL端口(默认443)) parser.add_argument(--timeout, -t, typeint, default10, help连接超时秒数(默认10)) args parser.parse_args() domains load_domains(args.domains) expired_list [] # 已过期 warning_list [] # 即将到期(alert_days内) ok_list [] # 正常 error_list [] # 检查失败 for domain in domains: try: _, days_left check_certificate(domain, args.port, args.timeout) if days_left 0: expired_list.append((domain, days_left)) logger.warning(f {domain}: 证书已过期!) elif days_left args.alert_days: warning_list.append((domain, days_left)) logger.warning(f {domain}: 还有{days_left}天到期) else: ok_list.append((domain, days_left)) logger.info(f {domain}: 还有{days_left}天到期) except socket.timeout: error_list.append((domain, 连接超时)) logger.error(f⏱️ {domain}: 连接超时({args.timeout}s)) except ConnectionRefusedError: error_list.append((domain, 端口不可达)) logger.error(f {domain}:{args.port} 端口不可达) except ssl.SSLError as e: error_list.append((domain, fSSL错误:{e})) logger.error(f {domain}: SSL握手失败 - {e}) except Exception as e: error_list.append((domain, str(e))) logger.error(f❓ {domain}: 未知错误 - {e}) # 汇总报告 print(f\n SSL证书检查报告({datetime.now():%Y-%m-%d %H:%M}) ) print(f总计: {len(domains)} | f正常: {len(ok_list)} | f即将到期: {len(warning_list)} | f已过期: {len(expired_list)} | f检查失败: {len(error_list)}) if expired_list: print(\n 以下证书已过期请立即处理) for domain, _ in expired_list: print(f - {domain}) if warning_list: print(f\n 以下证书将在{args.alert_days}天内到期) for domain, days in warning_list: print(f - {domain} (剩余{days}天)) if error_list: print(\n⚠️ 以下域名检查失败) for domain, reason in error_list: print(f - {domain}: {reason}) # 如果有过期或即将过期的证书以非零退出码返回 # 可用于CI/CD流水线或告警系统的自动化集成 if expired_list or warning_list: sys.exit(1) if __name__ __main__: main()三、脚本的最佳实践清单错误处理优先于正常逻辑每个脚本的第一段逻辑应该是环境检查依赖工具是否安装、配置文件是否存在、权限是否足够而不是直接进入业务逻辑。上述脚本都遵循这个原则。输出设计兼顾人类和机器日志输出应同时满足两个需求——人类可读带时间戳和级别标识和机器可解析用于集成到监控系统。本月的脚本统一使用[时间] [级别] 消息格式。幂等性设计巡检类的脚本应可以反复执行而不产生副作用。备份类的脚本应检查是否已有当日备份避免重复操作。参数化而非硬编码所有阈值、路径、连接信息都应通过参数或配置文件传入绝不硬编码在脚本中。这降低了脚本在不同环境间迁移的成本。合理的超时设置网络操作API调用、数据库连接、远程命令执行必须设置超时并用set -euo pipefail或Python的异常处理捕获超时场景。退出码的语义化使用不同的退出码传达不同的执行结果——0正常1一般错误2业务告警如证书即将过期3环境错误如依赖缺失。四、脚本管理与分发版本管理所有运维脚本使用Git管理仓库结构为scripts/{category}/{script_name}每个脚本有独立的CHANGELOG记录修改历史。共享与分发通过内部GitLab的CI/CD自动将脚本打包为容器镜像运维人员通过docker run或kubectl run执行确保执行环境的一致性。定期审计每季度对脚本库做一次审计——清理不再使用的脚本、更新依赖版本、检查是否有安全漏洞如硬编码的密码。五、总结在大模型能生成代码的时代写脚本本身不再是运维工程师的核心壁垒。真正的价值在于知道在什么场景下需要什么样的脚本、如何设计脚本的错误处理和边界条件、以及在什么时机和频率下运行脚本才能发挥最大效用。10个脚本只是工具背后的运维场景理解和工程化思维才是值得持续积累的核心能力。建议每个运维团队建立自己的脚本库并形成需求评审→代码Review→测试验证→生产部署→使用反馈的闭环流程。好的运维脚本库是团队经验的凝固——把一次次的故障处理和日常巡检中积累的判断力转化为可自动化执行、可跨人复用的代码资产。资料说明本文中的协议、版本、性能、成本和行业趋势应以可核验的一手资料为准。未标注统计口径的比例、时间表和预测仅作工程讨论不应视为行业事实。可参考 0730 资料来源索引并在发布前将具体来源贴到对应断言之后。