Shell/Python 运维自动化：配置管理与批量部署的工程化实践
·
Shell/Python 运维自动化：配置管理与批量部署的工程化实践
一、运维的"重复劳动"：100 台服务器，手动配置要多久？
运维团队最常见的场景：新上线一个服务，需要在 100 台服务器上安装依赖、修改配置、重启进程。手动 SSH 到每台服务器执行命令，即使每台只需 2 分钟，100 台也要 3 个多小时。更糟糕的是，手动操作的一致性无法保证——第 47 台服务器的配置文件少了一个逗号，排查这个 typo 又花了 2 小时。
配置管理和批量部署是运维自动化的基石。Ansible、SaltStack 等工具提供了声明式的配置管理能力，但在特定场景下（如无法安装 Agent 的受限环境、需要精细控制执行逻辑的复杂部署），Shell 脚本和 Python 脚本仍然是最灵活的选择。关键在于如何将"灵活"与"可靠"结合——脚本既要有足够的控制力，又要有完善的错误处理和幂等性保证。
二、配置管理的架构与幂等性设计
配置管理的核心原则是"幂等性"——同一份配置执行多次，结果始终一致。非幂等操作（如 echo "config" >> file）在重复执行时会产生重复内容，导致配置错误。幂等操作（如 lineinfile 或先检查再写入）则确保最终状态与期望一致。
flowchart TD
A[配置清单 YAML] --> B[配置渲染引擎<br/>Jinja2 模板]
B --> C[目标配置文件]
C --> D[差异比较<br/>当前 vs 目标]
D --> E{是否有差异?}
E -->|否| F[跳过，无需变更]
E -->|是| G[应用变更]
G --> H[验证变更结果]
H --> I{验证通过?}
I -->|是| J[记录变更日志]
I -->|否| K[自动回滚]
K --> L[告警通知]
subgraph "幂等性保证"
M[检查当前状态]
N[仅在状态不一致时变更]
O[变更后验证]
end
D --> M
G --> N
H --> O
批量部署的四个阶段：
- 预检：检查目标主机的连通性、磁盘空间、依赖版本
- 分发：将配置文件和二进制包推送到目标主机
- 执行：按序执行部署步骤（停止服务 → 更新文件 → 启动服务）
- 验证：检查服务健康状态，确认部署成功
三、工程化运维脚本的完整实现
配置管理框架
# config_manager.py — 声明式配置管理框架
# 设计意图：以 YAML 声明期望状态，自动计算差异并应用变更，
# 确保幂等性和可回滚性
import yaml
import hashlib
import shutil
import logging
from pathlib import Path
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ConfigChange:
"""配置变更记录"""
path: str
action: str # create / modify / delete
old_hash: Optional[str]
new_hash: str
timestamp: datetime
class ConfigManager:
"""声明式配置管理器"""
def __init__(self, state_dir: str = "/var/lib/config-manager"):
self.state_dir = Path(state_dir)
self.state_dir.mkdir(parents=True, exist_ok=True)
self.changes: List[ConfigChange] = []
def apply(self, manifest_path: str) -> dict:
"""应用配置清单"""
with open(manifest_path) as f:
manifest = yaml.safe_load(f)
results = {
"total": 0,
"changed": 0,
"unchanged": 0,
"failed": 0,
"details": [],
}
for item in manifest.get("files", []):
results["total"] += 1
try:
changed = self._apply_file(item)
if changed:
results["changed"] += 1
else:
results["unchanged"] += 1
results["details"].append({
"path": item["path"],
"status": "changed" if changed else "unchanged",
})
except Exception as e:
results["failed"] += 1
results["details"].append({
"path": item["path"],
"status": "failed",
"error": str(e),
})
for item in manifest.get("packages", []):
results["total"] += 1
try:
changed = self._apply_package(item)
if changed:
results["changed"] += 1
else:
results["unchanged"] += 1
results["details"].append({
"package": item["name"],
"status": "changed" if changed else "unchanged",
})
except Exception as e:
results["failed"] += 1
results["details"].append({
"package": item["name"],
"status": "failed",
"error": str(e),
})
return results
def _apply_file(self, item: dict) -> bool:
"""应用文件配置（幂等）"""
path = Path(item["path"])
content = item["content"]
owner = item.get("owner", "root")
group = item.get("group", "root")
mode = item.get("mode", "0644")
# 计算目标内容的哈希
target_hash = hashlib.sha256(content.encode()).hexdigest()
# 检查当前状态
if path.exists():
current_content = path.read_text()
current_hash = hashlib.sha256(current_content.encode()).hexdigest()
if current_hash == target_hash:
# 内容一致，无需变更
return False
# 内容不一致：备份后更新
self._backup(path)
path.write_text(content)
old_hash = current_hash
action = "modify"
else:
# 文件不存在：创建
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
old_hash = None
action = "create"
# 设置权限和属主
import os
os.chmod(path, int(mode, 8))
os.chown(path, self._get_uid(owner), self._get_gid(group))
# 记录变更
self.changes.append(ConfigChange(
path=str(path),
action=action,
old_hash=old_hash,
new_hash=target_hash,
timestamp=datetime.now(),
))
logger.info(f"{action}: {path}")
return True
def _apply_package(self, item: dict) -> bool:
"""应用包配置（幂等）"""
import subprocess
name = item["name"]
state = item.get("state", "present")
# 检查包是否已安装
check_cmd = f"dpkg -l {name} 2>/dev/null | grep -q '^ii'"
is_installed = subprocess.call(
check_cmd, shell=True, stdout=subprocess.DEVNULL
) == 0
if state == "present" and not is_installed:
subprocess.check_call(
["apt-get", "install", "-y", name],
stdout=subprocess.DEVNULL
)
logger.info(f"installed: {name}")
return True
elif state == "absent" and is_installed:
subprocess.check_call(
["apt-get", "remove", "-y", name],
stdout=subprocess.DEVNULL
)
logger.info(f"removed: {name}")
return True
return False
def rollback(self) -> int:
"""回滚所有变更"""
rolled_back = 0
for change in reversed(self.changes):
backup_path = self._get_backup_path(change.path)
if backup_path.exists():
shutil.copy2(str(backup_path), change.path)
rolled_back += 1
logger.info(f"rolled back: {change.path}")
elif change.action == "create":
# 创建的文件：删除
Path(change.path).unlink(missing_ok=True)
rolled_back += 1
logger.info(f"removed created file: {change.path}")
self.changes.clear()
return rolled_back
def _backup(self, path: Path):
"""备份当前文件"""
backup_path = self._get_backup_path(str(path))
backup_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(str(path), str(backup_path))
def _get_backup_path(self, original_path: str) -> Path:
"""获取备份文件路径"""
safe_name = original_path.replace("/", "_")
return self.state_dir / f"{safe_name}.bak"
def _get_uid(self, username: str) -> int:
"""获取用户 UID"""
import pwd
try:
return pwd.getpwnam(username).pw_uid
except KeyError:
return 0
def _get_gid(self, groupname: str) -> int:
"""获取组 GID"""
import grp
try:
return grp.getgrnam(groupname).gr_gid
except KeyError:
return 0
批量部署脚本
#!/bin/bash
# batch_deploy.sh — 批量部署脚本
# 设计意图：在多台目标主机上并行执行部署，包含预检、
# 分发、执行和验证四个阶段，支持滚动部署和失败回滚
set -euo pipefail
# ---- 配置 ----
HOSTS_FILE="${HOSTS_FILE:-hosts.txt}"
DEPLOY_DIR="/opt/app"
MAX_PARALLEL=10
ROLLING_BATCH=5 # 滚动部署：每批 5 台
ROLLBACK_ON_FAIL=1 # 失败时自动回滚
# ---- 日志 ----
LOG_DIR="/var/log/deploy"
mkdir -p "$LOG_DIR"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
LOG_FILE="$LOG_DIR/deploy_${TIMESTAMP}.log"
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" | tee -a "$LOG_FILE"
}
# ---- 预检 ----
preflight_check() {
local host=$1
log "[PREFLIGHT] Checking $host..."
# 检查 SSH 连通性
if ! ssh -o ConnectTimeout=5 -o BatchMode=yes "$host" "echo ok" &>/dev/null; then
log "[PREFLIGHT] FAILED: Cannot connect to $host"
return 1
fi
# 检查磁盘空间（至少 1GB）
local disk_avail
disk_avail=$(ssh "$host" "df -BG $DEPLOY_DIR | awk 'NR==2{print \$4}' | tr -d 'G'")
if [[ "$disk_avail" -lt 1 ]]; then
log "[PREFLIGHT] FAILED: $host has less than 1GB disk space"
return 1
fi
# 检查当前服务状态
local service_status
service_status=$(ssh "$host" "systemctl is-active app-service 2>/dev/null || echo unknown")
log "[PREFLIGHT] $host: disk=${disk_avail}G, service=$service_status"
return 0
}
# ---- 部署单台主机 ----
deploy_host() {
local host=$1
local log_prefix="[DEPLOY] $host"
log "$log_prefix: Starting deployment..."
# Step 1: 停止服务（优雅停止）
if ! ssh "$host" "systemctl stop app-service" 2>/dev/null; then
log "$log_prefix: WARN - Service stop failed, may not be running"
fi
# Step 2: 备份当前版本
ssh "$host" "cp -r $DEPLOY_DIR/current $DEPLOY_DIR/previous 2>/dev/null || true"
# Step 3: 分发新版本
if ! scp -r ./dist/* "$host:$DEPLOY_DIR/current/"; then
log "$log_prefix: FAILED - File distribution error"
return 1
fi
# Step 4: 更新配置
if ! ssh "$host" "cd $DEPLOY_DIR/current && python3 config_manager.py apply manifest.yml"; then
log "$log_prefix: FAILED - Configuration error"
# 回滚：恢复上一版本
if [[ "$ROLLBACK_ON_FAIL" -eq 1 ]]; then
ssh "$host" "rm -rf $DEPLOY_DIR/current && mv $DEPLOY_DIR/previous $DEPLOY_DIR/current"
fi
return 1
fi
# Step 5: 启动服务
if ! ssh "$host" "systemctl start app-service"; then
log "$log_prefix: FAILED - Service start error"
return 1
fi
# Step 6: 健康检查（等待最多 30 秒）
local retries=6
local wait_sec=5
local healthy=false
for i in $(seq 1 $retries); do
if ssh "$host" "curl -sf http://localhost:3000/health" &>/dev/null; then
healthy=true
break
fi
sleep $wait_sec
done
if [[ "$healthy" != "true" ]]; then
log "$log_prefix: FAILED - Health check timeout"
# 回滚
if [[ "$ROLLBACK_ON_FAIL" -eq 1 ]]; then
ssh "$host" "systemctl stop app-service"
ssh "$host" "rm -rf $DEPLOY_DIR/current && mv $DEPLOY_DIR/previous $DEPLOY_DIR/current"
ssh "$host" "systemctl start app-service"
fi
return 1
fi
log "$log_prefix: Deployment successful"
return 0
}
# ---- 滚动部署 ----
rolling_deploy() {
local hosts=("$@")
local total=${#hosts[@]}
local failed=0
local succeeded=0
log "Starting rolling deployment: $total hosts, batch size $ROLLING_BATCH"
for ((i=0; i<total; i+=ROLLING_BATCH)); do
local batch=("${hosts[@]:i:ROLLING_BATCH}")
local batch_num=$((i / ROLLING_BATCH + 1))
local batch_total=$(( (total + ROLLING_BATCH - 1) / ROLLING_BATCH ))
log "=== Batch $batch_num/$batch_total ==="
# 并行部署当前批次
local pids=()
for host in "${batch[@]}"; do
deploy_host "$host" &
pids+=($!)
done
# 等待当前批次完成
local batch_failed=0
for pid in "${pids[@]}"; do
if ! wait "$pid"; then
batch_failed=$((batch_failed + 1))
fi
done
if [[ "$batch_failed" -gt 0 ]]; then
log "Batch $batch_num: $batch_failed hosts failed"
failed=$((failed + batch_failed))
# 如果批次失败率超过 50%，停止部署
if [[ "$batch_failed" -gt "$((ROLLING_BATCH / 2))" ]]; then
log "Too many failures in batch $batch_num, stopping deployment"
break
fi
else
succeeded=$((succeeded + ${#batch[@]}))
log "Batch $batch_num: All hosts deployed successfully"
fi
done
log "=== Deployment Summary ==="
log "Total: $total, Succeeded: $succeeded, Failed: $failed"
return $((failed > 0 ? 1 : 0))
}
# ---- 主流程 ----
main() {
# 读取主机列表
if [[ ! -f "$HOSTS_FILE" ]]; then
log "ERROR: Hosts file not found: $HOSTS_FILE"
exit 1
fi
mapfile -t hosts < "$HOSTS_FILE"
log "Loaded ${#hosts[@]} hosts from $HOSTS_FILE"
# 预检
local valid_hosts=()
for host in "${hosts[@]}"; do
if preflight_check "$host"; then
valid_hosts+=("$host")
fi
done
if [[ ${#valid_hosts[@]} -eq 0 ]]; then
log "ERROR: No valid hosts after preflight check"
exit 1
fi
log "Preflight passed: ${#valid_hosts[@]}/${#hosts[@]} hosts"
# 执行滚动部署
rolling_deploy "${valid_hosts[@]}"
}
main "$@"
四、运维自动化的 Trade-offs
脚本 vs 配置管理工具：Shell/Python 脚本灵活但缺乏标准化，Ansible 等工具标准化但灵活性受限。脚本适合一次性任务和特殊场景，工具适合标准化运维。建议核心基础设施（服务器初始化、安全加固）使用 Ansible，业务部署和特殊操作使用脚本。
并行度与风险控制：并行部署可以大幅缩短时间，但增加了"同时出错"的风险。滚动部署（分批执行）在速度和风险之间取得平衡——每批部署后验证，失败则停止。建议生产环境使用 5-10 台/批的滚动部署，测试环境可以全量并行。
幂等性的实现成本：确保每个操作幂等需要额外的检查逻辑（如先判断文件是否存在再写入），增加了脚本复杂度。但非幂等操作在重复执行时的调试成本远高于幂等性的实现成本。建议对所有配置文件操作实现幂等性，对不可逆操作（如数据库迁移）实现显式的确认机制。
回滚的完整性：文件回滚相对简单（恢复备份），但数据库变更和服务状态回滚更复杂。一个已经执行的 ALTER TABLE 无法简单回滚，一个已经发送的通知无法撤回。需要在部署计划中为每个步骤定义回滚操作，并在自动化脚本中实现。
五、总结
Shell/Python 运维自动化通过声明式配置管理和滚动批量部署，将运维操作从"手动逐台执行"推向"自动化批量处理"。核心原则是幂等性（同一操作多次执行结果一致）和滚动部署（分批执行降低风险）。但脚本与工具的选择、并行度与风险控制的平衡、幂等性的实现成本和回滚的完整性是需要权衡的因素。在实际落地中，建议核心基础设施使用 Ansible 等工具标准化管理，业务部署使用脚本实现精细控制，所有操作实现幂等性保证，生产环境使用滚动部署策略。运维自动化的目标不是"消除人工"，而是"让机器做重复的事，让人做决策的事"。
更多推荐
6
0- 0
已为社区贡献8条内容
所有评论(0)