第330篇：自动化运维综合实战案例

关键词

综合实战、全流程自动化、案例复盘、端到端自动化、企业实践、运维转型、最佳实践整合

一、案例背景

1.1 企业现状

某中型企业网络自动化转型案例：

企业基本信息：

行业：金融科技 网络规模：3 个数据中心 + 20 个分支机构 设备总量：260 台 厂商分布：华为 60% + Cisco 25% + H3C 15% 运维团队：8 人（含 1 名经理） 网络变更频率：每周 15-25 次

转型前的痛点：

❌ 配置备份靠人工 └─ 每月手动 SSH 备份一次 └─ 遇到设备故障才想起上次没备份 ❌ 变更靠"胆量" └─ 深夜变更，工程师逐台 SSH 配置 └─ 没有回滚方案，出问题手忙脚乱 └─ 发生过配置错误影响生产业务 ❌ 巡检形同虚设 └─ 每周抽查几台设备 └─ 无法发现趋势性问题 ❌ 运维知识靠个人 └─ 核心工程师离职可能导致运维瘫痪 └─ 新人上手需要 3-6 个月

1.2 转型目标

自动化转型目标：

第一阶段（3 个月）：

1. 配置备份 100% 自动化
2. 日常巡检自动化，覆盖所有设备
3. 建立 Git 配置仓库
4. 统计覆盖率：30%

第二阶段（6 个月）：

1. 配置模板化（Jinja2）
2. Ansible 批量配置部署
3. 合规检查自动化
4. CI/CD 基础流水线
5. 统计覆盖率：60%

第三阶段（12 个月）：

1. 自动化运维平台 MVP
2. 变更审批流程自动化
3. 自动回滚机制
4. 统一多厂商管理
5. 统计覆盖率：85%

二、第一阶段实现

2.1 配置备份系统

#!/usr/bin/env python3
# stage1_backup.py — 配置备份系统

from netmiko import ConnectHandler
from concurrent.futures import (
    ThreadPoolExecutor, as_completed
)
from datetime import datetime
import os
import json
import logging
import git  # GitPython

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)


class ConfigBackupSystem:
    """配置备份系统"""

    def __init__(
        self,
        inventory_file: str = "inventory.json",
        backup_dir: str = "backup/",
        git_repo: str = "config_repo/",
    ):
        self.inventory_file = inventory_file
        self.backup_dir = backup_dir
        self.git_repo = git_repo

        os.makedirs(backup_dir, exist_ok=True)
        self._init_git()

    def _init_git(self):
        """初始化 Git 仓库"""
        if os.path.exists(self.git_repo):
            self.repo = git.Repo(self.git_repo)
        else:
            self.repo = git.Repo.init(self.git_repo)
        logger.info(f"Git 仓库: {self.git_repo}")

    def load_inventory(self) -> list:
        """加载设备清单"""
        with open(self.inventory_file, "r") as f:
            return json.load(f)

    def backup_device(self, device: dict) -> dict:
        """备份单台设备"""
        start_time = datetime.now()
        hostname = device.get("hostname", "unknown")

        try:
            conn = ConnectHandler(**device)
            conn.enable()
            hostname = conn.find_prompt().rstrip("#>")

            config = conn.send_command(
                "display current-configuration",
                read_timeout=60,
            )
            conn.disconnect()

            # 保存到文件
            timestamp = start_time.strftime("%Y%m%d_%H%M%S")
            filename = f"{hostname}_{timestamp}.cfg"
            filepath = os.path.join(
                self.backup_dir, filename
            )
            with open(filepath, "w") as f:
                f.write(f"# {hostname}\n")
                f.write(f"# 备份时间: {start_time}\n")
                f.write(f"{'='*60}\n")
                f.write(config)

            # 也保存最新版本
            latest = os.path.join(
                self.backup_dir, f"{hostname}_latest.cfg"
            )
            with open(latest, "w") as f:
                f.write(config)

            duration = (
                datetime.now() - start_time
            ).total_seconds()

            logger.info(
                f"✅ {hostname}: 备份完成 ({duration:.1f}s)"
            )
            return {
                "hostname": hostname,
                "success": True,
                "file": filename,
                "duration": duration,
            }

        except Exception as e:
            logger.error(f"❌ {hostname}: 备份失败 - {e}")
            return {
                "hostname": hostname,
                "success": False,
                "error": str(e),
            }

    def backup_all(self) -> dict:
        """备份所有设备"""
        devices = self.load_inventory()
        logger.info(f"开始备份 {len(devices)} 台设备...")

        results = {"success": 0, "failed": 0, "details": []}

        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = {
                executor.submit(self.backup_device, dev): dev
                for dev in devices
            }

            for future in as_completed(futures):
                result = future.result()
                results["details"].append(result)
                if result["success"]:
                    results["success"] += 1
                else:
                    results["failed"] += 1

        # Git commit
        self._git_commit()
        return results

    def _git_commit(self):
        """提交到 Git"""
        try:
            self.repo.index.add("*")
            timestamp = datetime.now().strftime(
                "%Y-%m-%d %H:%M:%S"
            )
            self.repo.index.commit(
                f"配置备份 - {timestamp}"
            )
            logger.info("配置已提交到 Git 仓库")
        except Exception as e:
            logger.warning(f"Git 提交失败: {e}")

    def get_backup_status(self) -> dict:
        """获取备份状态"""
        devices = self.load_inventory()
        status = {}

        for dev in devices:
            hostname = dev.get("hostname", "unknown")
            latest = os.path.join(
                self.backup_dir, f"{hostname}_latest.cfg"
            )
            if os.path.exists(latest):
                mtime = os.path.getmtime(latest)
                status[hostname] = {
                    "backed_up": True,
                    "last_backup": datetime.fromtimestamp(
                        mtime
                    ).isoformat(),
                }
            else:
                status[hostname] = {
                    "backed_up": False,
                }

        return status

2.2 日常巡检系统

# stage1_patrol.py — 日常巡检系统（已集成到第 324 篇）
# 实际部署时复用第 324 篇的 PatrolEngine

# 定时任务配置（crontab）
# 每天早上 8:00 自动执行巡检
# 0 8 * * * cd /opt/network-ops && python patrol_runner.py

# patrol_runner.py — 巡检执行入口
from patrol_engine import PatrolEngine
from patrol_report import PatrolReportGenerator
import json

def main():
    engine = PatrolEngine(max_workers=10)

    with open("inventory.json", "r") as f:
        devices = json.load(f)

    results = engine.patrol_all(devices, "daily")

    report_file = f"reports/patrol_{datetime.now().strftime('%Y%m%d')}.html"
    PatrolReportGenerator.generate_html(results, report_file)

    # 检查严重告警
    critical_alerts = sum(
        len([a for a in r.alerts if a["level"] == "CRITICAL"])
        for r in results.values()
    )
    if critical_alerts > 0:
        send_alert(f"巡检发现 {critical_alerts} 个严重告警！")

    print(f"巡检完成，报告: {report_file}")

if __name__ == "__main__":
    from datetime import datetime
    main()

三、第二阶段实现

3.1 配置模板化

模板化配置生成流程：

数据层（YAML）： | device_vlans.yml ┌─────────────────────────────────────┐ └─────────────────────────────────────┘ 模板层（Jinja2）： ┌─────────────────────────────────────┐ └─────────────────────────────────────┘ 渲染： └─ YAML + Jinja2 → 完整的 CLI 配置 | vlans: - id: 10 name: MGMT subnet: 10.10.10.0/24 - id: 20 name: SERV_PROD subnet: 10.10.20.0/24 - id: 30 name: SERV_TEST subnet: 10.10.30.0/24 vlan_batch {{ vlans | map( attribute='id') | join(' ') }} {% for vlan in vlans %} vlan {{ vlan.id }} name {{ vlan.name }} {% endfor %} | | | --- | --- | --- |

3.2 Ansible 部署脚本

# deploy_vlan_playbook.yml — VLAN 部署 Playbook
---
- name: 批量部署 VLAN 配置
  hosts: all
  gather_facts: no
  vars:
    vlan_data: "{{ lookup('file', 'vlan_data.yml') | from_yaml }}"

  tasks:
    - name: 生成接口配置
      template:
        src: "interface_config.j2"
        dest: "/tmp/{{ inventory_hostname }}_config.cfg"
      delegate_to: localhost

    - name: 备份当前配置
      huawei_vrp_command:
        commands: "display current-configuration"
      register: pre_config

    - name: 部署配置
      huawei_vrp_config:
        src: "/tmp/{{ inventory_hostname }}_config.cfg"

    - name: 保存配置
      huawei_vrp_command:
        commands: save

    - name: 验证 BGP 邻居
      huawei_vrp_command:
        commands: display bgp peer
      register: bgp_status

    - name: 检查 BGP 状态
      fail:
        msg: "BGP 邻居异常！"
      when: "'Established' not in bgp_status.stdout"

3.3 CI/CD 流水线

# .gitlab-ci.yml — GitLab CI 流水线
stages:
  - validate
  - test
  - deploy
  - verify

validate:
  stage: validate
  script:
    - python -c "import yaml; yaml.safe_load(open('config/vlans.yml'))"
    - python -c "import jinja2; env = jinja2.Environment(); env.get_template('templates/*.j2')"
  only:
    - develop
    - master

test:
  stage: test
  script:
    - ansible-playbook test_playbook.yml --syntax-check
    - python tests/validate_config.py
  only:
    - develop

deploy:
  stage: deploy
  script:
    - ansible-playbook deploy_playbook.yml -i inventory/staging.yml
  environment:
    name: staging
  only:
    - develop

deploy-production:
  stage: deploy
  script:
    - ansible-playbook deploy_playbook.yml -i inventory/production.yml
  environment:
    name: production
  when: manual
  only:
    - master

verify:
  stage: verify
  script:
    - python tests/verify_deployment.py
  only:
    - master

四、第三阶段实现

4.1 自动化运维平台

# platform/app.py — 自动化运维平台核心后端
from flask import Flask, render_template, request, jsonify
from flask_login import LoginManager, login_required, current_user
from flask_sqlalchemy import SQLAlchemy
import json
from datetime import datetime

app = Flask(__name__)
app.config["SECRET_KEY"] = "your-secret-key"
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///network_ops.db"
db = SQLAlchemy(app)
login_manager = LoginManager(app)


# =============================================
# 数据模型
# =============================================

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    role = db.Column(db.String(20))  # admin/operator/auditor

class ChangeOrder(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200))
    description = db.Column(db.Text)
    status = db.Column(db.String(20))  # pending/approved/rejected/done
    requester = db.Column(db.String(80))
    reviewer = db.Column(db.String(80))
    devices = db.Column(db.Text)  # JSON
    commands = db.Column(db.Text)  # JSON
    rollback_commands = db.Column(db.Text)
    risk_level = db.Column(db.String(10))
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    executed_at = db.Column(db.DateTime)


# =============================================
# 变更管理 API
# =============================================

@app.route("/api/change/create", methods=["POST"])
@login_required
def create_change():
    data = request.get_json()
    change = ChangeOrder(
        title=data["title"],
        description=data.get("description", ""),
        status="pending",
        requester=current_user.username,
        devices=json.dumps(data.get("devices", [])),
        commands=json.dumps(data.get("commands", [])),
        rollback_commands=json.dumps(
            data.get("rollback_commands", [])
        ),
        risk_level=data.get("risk_level", "LOW"),
    )
    db.session.add(change)
    db.session.commit()

    # 发送审批通知
    notify_reviewer(change)
    return jsonify({"change_id": change.id})


@app.route("/api/change/<int:change_id>/approve", methods=["POST"])
@login_required
def approve_change(change_id):
    change = ChangeOrder.query.get_or_404(change_id)
    change.status = "approved"
    change.reviewer = current_user.username
    db.session.commit()

    # 自动执行变更
    execute_change(change)
    return jsonify({"status": "approved"})


@app.route("/api/change/<int:change_id>/rollback", methods=["POST"])
@login_required
def rollback_change(change_id):
    change = ChangeOrder.query.get_or_404(change_id)
    # 执行回滚
    rollback_result = execute_rollback(change)
    change.status = "rolled_back"
    db.session.commit()
    return jsonify(rollback_result)


# =============================================
# 仪表盘
# =============================================

@app.route("/api/dashboard/summary")
@login_required
def dashboard_summary():
    return jsonify({
        "total_devices": Device.query.count(),
        "backup_coverage": get_backup_coverage(),
        "compliance_score": get_compliance_score(),
        "recent_changes": get_recent_changes(10),
        "alerts_today": get_today_alerts(),
        "automation_rate": calculate_automation_rate(),
    })


@app.route("/")
@login_required
def index():
    return render_template("dashboard.html")

4.2 变更流程 UI

<!-- templates/change_create.html — 变更创建页面 -->
<!DOCTYPE html>
<html>
<head>
    <title>创建变更工单</title>
    <link href="https://cdn.bootcdn.net/ajax/libs/twitter-bootstrap/5.3.0/css/bootstrap.min.css" rel="stylesheet">
</head>
<body>
<div class="container mt-4">
    <h2>创建变更工单</h2>

    <form id="changeForm">
        <div class="mb-3">
            <label>变更标题</label>
            <input type="text" class="form-control" name="title" required>
        </div>

        <div class="mb-3">
            <label>变更描述</label>
            <textarea class="form-control" name="description" rows="3"></textarea>
        </div>

        <div class="mb-3">
            <label>目标设备</label>
            <select class="form-select" name="devices" multiple>
                {% for device in devices %}
                <option value="{{ device.id }}">{{ device.hostname }}</option>
                {% endfor %}
            </select>
        </div>

        <div class="mb-3">
            <label>配置命令</label>
            <textarea class="form-control" name="commands" rows="5"
                      placeholder="请输入配置命令，每行一条"></textarea>
        </div>

        <div class="mb-3">
            <label>回滚命令</label>
            <textarea class="form-control" name="rollback_commands" rows="5"
                      placeholder="请输入回滚命令，每行一条"></textarea>
        </div>

        <div class="mb-3">
            <label>风险等级</label>
            <select class="form-select" name="risk_level">
                <option value="LOW">低风险</option>
                <option value="MEDIUM">中风险</option>
                <option value="HIGH">高风险</option>
            </select>
        </div>

        <button type="submit" class="btn btn-primary">提交变更</button>
    </form>
</div>

<script>
document.getElementById('changeForm').onsubmit = async (e) => {
    e.preventDefault();
    const form = e.target;
    const data = {
        title: form.title.value,
        description: form.description.value,
        devices: Array.from(form.devices.selectedOptions).map(o => o.value),
        commands: form.commands.value.split('\n').filter(c => c.trim()),
        rollback_commands: form.rollback_commands.value.split('\n').filter(c => c.trim()),
        risk_level: form.risk_level.value,
    };

    const resp = await fetch('/api/change/create', {
        method: 'POST',
        headers: {'Content-Type': 'application/json'},
        body: JSON.stringify(data),
    });
    const result = await resp.json();
    alert(`变更工单已创建，ID: ${result.change_id}`);
};
</script>
</body>
</html>

五、转型效果评估

5.1 量化对比

转型前后对比（12 个月后）：

  指标               | 转型前          | 转型后          | 提升
  ──────────────────────────────────────────────────────
  配置备份            | 每月手工 1 次    | 每天自动 1 次    | 30 倍
  备份成功率          | ~90%            | 99.9%           | +9.9%
  单次变更耗时        | 平均 45 分钟     | 平均 8 分钟      | 5.6 倍
  变更成功率          | 85%             | 98%             | +13%
  回滚时间            | 30-60 分钟      | 1-3 分钟         | 20 倍
  巡检覆盖率          | 20%（抽查）     | 100%（全量）     | 5 倍
  巡检耗时            | 4 小时/周       | 5 分钟/天        | 48 倍
  合规检查            | 季度手工 1 次   | 每天自动检查     | 90 倍
  配置错误导致故障     | 每年 3-4 次     | 0 次             | 100%
  新人上手时间        | 3-6 个月        | 1-2 个月         | 3 倍

5.2 团队反馈

运维团队的真实反馈：

工程师 A（8 年经验）：

"以前凌晨 2 点还在逐台 SSH 配置， 现在点一下按钮就完成了。 再也不担心敲错命令了。"

工程师 B（3 年经验）：

"从学 CLI 到学 Python/Ansible， 一开始觉得难，现在觉得"真香"。 以后不会被淘汰了。"

运维经理：

"自动化后，同样的团队做了更多的事情。 以前只能被动救火，现在主动做优化。 ROI 非常清晰。"

新人工程师 C（入职半年）：

"自动化平台让我很快上手了运维工作。 不需要记住所有 CLI 命令， 只要知道在平台上怎么做就行了。"

六、经验与教训

本次转型的关键经验：

成功因素：

1. 高层支持 └─ CTO 明确支持，投入专职人员和预算
2. 从小做起 └─ 备份开始（2 周见效），建立信心 └─ 每个阶段有明确的可交付成果
3. 团队赋能 └─ 组织 Python/Ansible 培训 └─ 鼓励工程师参与开发 └─ 设立"自动化 champion"角色
4. 不追求完美 └─ "够好"就上线，后续迭代优化 └─ 不要等"大平台"建成才用

踩过的坑：

❌ 初期追求大平台，3 个月没产出 → 改为"备份先行，2 周见效" ❌ 脚本只在一台工程师电脑上 → 脚本放入 Git，任何工程师可运行 ❌ 自动化执行前没做配置验证 → 增加 dry-run 和语法检查环节 ❌ 回滚方案事后才补 → 强制每个变更必须预生成回滚脚本 ❌ 忽略非技术因素 → 团队抵触 → 通过培训和激励化解

七、总结与展望

从这篇综合案例中，我们看到了：

  网络自动化转型不是"一蹴而就"的工程，
  ┌─ 是从一个简单的备份脚本开始的旅程
  ├─ 是一个持续 12 个月的渐进过程
  └─ 是工具、流程、人三者协同进化的结果

  关键里程碑：
  ┌──────────────────────────────────────────┐
  │  第 2 周：配置备份自动化 ✅               │
  │  第 4 周：日常巡检自动化 ✅               │
  │  第 8 周：Git 配置管理 ✅                 │
  │  第 12 周：配置模板化 ✅                  │
  │  第 20 周：Ansible 自动化部署 ✅          │
  │  第 30 周：合规检查自动化 ✅              │
  │  第 40 周：CI/CD 流水线 ✅               │
  │  第 50 周：自动化平台 MVP ✅             │
  └──────────────────────────────────────────┘

  未来的方向：
  ┌──────────────────────────────────────────┐
  │  ┌─ AI 辅助排障（基于知识图谱）          │
  │  ├─ Telemetry 全量数据采集               │
  │  ├─ 预测性运维                           │
  │  ├─ 意图驱动网络                         │
  │  └─ 自愈网络                             │
  └──────────────────────────────────────────┘

  给正在读这篇文章的你：
  ┌──────────────────────────────────────────┐
  │  如果你还在犹豫从哪里开始，                │
  │  就从最简单的「配置备份」开始。             │
  │  一行代码都不要写，先试试 Netmiko。         │
  │                                           │
  │  记住：                                    │
  │  最好的开始时间是昨天，                     │
  │  第二好的开始时间就是现在。                 │
  └──────────────────────────────────────────┘

下篇预告：第331篇《中小企业网络改造与平滑迁移》——开启第八篇章「网络案例实战与经验总结」，通过实际网络案例提升综合分析和排障能力。

下篇预告：第331篇《中小企业网络改造与平滑迁移》——开启第八篇章「网络案例实战与经验总结」，通过实际网络案例提升综合分析和排障能力。