7月自动化运维脚本精选：Shell与Python在生产环境中最具价值的10个效率工具集
·
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_NOTREADY=0
ALERT_THRESHOLD_CPU=80
ALERT_THRESHOLD_MEM=80
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 type=Warning \
--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 (NR==1) {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" --tail=100 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(
level=logging.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(parents=True, exist_ok=True)
def _load_config(self, config_file: str) -> dict:
"""加载YAML配置文件"""
try:
with open(config_file, 'r', encoding='utf-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 / f"mysql_{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-purged=OFF', # 防止GTID冲突
database
]
gzip_cmd = ['gzip', '-c']
with open(backup_file, 'wb') as f_out:
dump_proc = subprocess.Popen(
mysqldump_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
gzip_proc = subprocess.Popen(
gzip_cmd, stdin=dump_proc.stdout, stdout=f_out, stderr=subprocess.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', errors='replace')
logger.error(f"mysqldump失败({database}): {error_msg}")
backup_file.unlink(missing_ok=True)
raise subprocess.CalledProcessError(
dump_proc.returncode, mysqldump_cmd, stderr=dump_err
)
if gzip_proc.returncode != 0:
error_msg = gzip_err.decode('utf-8', errors='replace')
logger.error(f"gzip压缩失败: {error_msg}")
backup_file.unlink(missing_ok=True)
raise subprocess.CalledProcessError(
gzip_proc.returncode, gzip_cmd, stderr=gzip_err
)
file_size_mb = backup_file.stat().st_size / (1024 * 1024)
logger.info(f"MySQL备份完成: {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_url=self.config.get('s3_endpoint'),
aws_access_key_id=self.config.get('s3_access_key'),
aws_secret_access_key=self.config.get('s3_secret_key')
)
file_name = os.path.basename(file_path)
s3_key = f"db_backups/{datetime.now():%Y/%m}/{file_name}"
s3_client.upload_file(
file_path, self.s3_bucket, s3_key,
ExtraArgs={'StorageClass': 'STANDARD_IA'} # 非频繁访问层，节省成本
)
logger.info(f"S3上传成功: s3://{self.s3_bucket}/{s3_key}")
return True
except ClientError as e:
logger.error(f"S3上传失败: {e}")
return False
except Exception as e:
logger.error(f"S3上传异常: {e}")
return False
def cleanup_old_backups(self):
"""清理超过保留期的本地备份文件"""
cutoff_date = datetime.now() - timedelta(days=self.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_info=True)
continue
# 清理过期备份
self.cleanup_old_backups()
logger.info("所有备份任务完成")
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='多数据库自动备份工具')
parser.add_argument(
'--config', '-c',
required=True,
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(
level=logging.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), timeout=timeout) as sock:
with context.wrap_socket(sock, server_hostname=domain) 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', encoding='utf-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(description='SSL证书到期监控')
parser.add_argument('--domains', '-d', required=True, help='域名列表文件')
parser.add_argument('--alert-days', '-a', type=int, default=30,
help='提前多少天告警(默认30)')
parser.add_argument('--port', '-p', type=int, default=443,
help='SSL端口(默认443)')
parser.add_argument('--timeout', '-t', type=int, default=10,
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, f"SSL错误:{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 资料来源索引，并在发布前将具体来源贴到对应断言之后。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献22条内容
所有评论(0)