第316篇：CI/CD 网络变更流水线

关键词

CI/CD、持续集成、持续部署、网络变更流水线、GitOps、自动化测试、灰度发布、变更审批

一、CI/CD 引入网络领域

1.1 为什么网络需要 CI/CD

传统网络变更 vs CI/CD 流水线：

  传统变更流程：
  ┌─ 工程师本地修改配置
  ├─ 手工登录设备执行
  ├─ 无版本控制
  ├─ 无自动化测试
  ├─ 人工验证
  └─ 变更记录全靠邮件

  问题：
  ┌─ 人为错误多（漏配、错配）
  ├─ 回滚困难
  ├─ 无法审计
  └─ 周期长（申请→审批→变更窗→验证）

  CI/CD 变更流程：
  ┌─ Git 提交配置代码
  ├─ 自动语法检查 + 合规检查
  ├─ 自动在测试环境验证
  ├─ 自动生成变更差异
  ├─ 审批后自动下发
  └─ 自动验证 + 回滚

1.2 网络 CI/CD 流水线架构

网络 CI/CD 流水线：

二、GitLab CI 实现网络流水线

2.1 GitLab CI 配置文件

# .gitlab-ci.yml — 网络变更 CI/CD 流水线

stages:
  - validate
  - test
  - approve
  - deploy
  - verify

variables:
  # 通用变量
  ANSIBLE_HOST_KEY_CHECKING: "false"
  NETWORK_CONFIG_DIR: "config"
  TEMPLATE_DIR: "templates"

# ===== Stage 1: 验证 =====
validate-yaml:
  stage: validate
  script:
    - python -c "import yaml; yaml.safe_load(open('config/devices.yml'))"
    - echo "✓ YAML 格式验证通过"
  only:
    - master
    - merge_requests

validate-templates:
  stage: validate
  script:
    - python -m pip install jinja2
    - python scripts/validate_templates.py
  only:
    - master
    - merge_requests

validate-ip-conflicts:
  stage: validate
  script:
    - python scripts/check_ip_conflicts.py
  only:
    - master
    - merge_requests

# ===== Stage 2: 测试 =====
test-render:
  stage: test
  script:
    - python scripts/render_configs.py --output rendered/
    - diff -r rendered/ baselines/ || true
  artifacts:
    paths:
      - rendered/

test-compliance:
  stage: test
  script:
    - python scripts/compliance_check.py --config rendered/
  only:
    - master

# ===== Stage 3: 审批 =====
approve:
  stage: approve
  script:
    # 生成变更摘要
    - python scripts/generate_diff_summary.py
    # 发送审批通知
    - python scripts/notify_approvers.py
  when: manual
  only:
    - master

# ===== Stage 4: 部署 =====
deploy-canary:
  stage: deploy
  script:
    - python scripts/deploy_rolling.py --limit "canary" --batch 1
  environment:
    name: production/canary
  when: manual
  only:
    - master

deploy-production:
  stage: deploy
  script:
    - python scripts/deploy_rolling.py --batch 3
  environment:
    name: production
  when: manual
  needs: ["deploy-canary"]
  only:
    - master

# ===== Stage 5: 验证 =====
verify-deployment:
  stage: verify
  script:
    - python scripts/verify_deployment.py
  only:
    - master

2.2 Python 验证脚本

#!/usr/bin/env python3
# scripts/validate_templates.py — 模板验证

import yaml
import os
import sys
from jinja2 import Environment, FileSystemLoader

def validate_templates():
    """验证所有模板可正常渲染"""
    errors = []

    # 加载数据
    with open("config/devices.yml") as f:
        devices = yaml.safe_load(f)

    # 加载模板
    env = Environment(
        loader=FileSystemLoader("templates"),
        trim_blocks=True,
        lstrip_blocks=True,
    )

    for device in devices.get("devices", []):
        template_name = f"{device.get('role', 'default')}.j2"
        try:
            template = env.get_template(template_name)
            output = template.render(device=device)
            if not output.strip():
                errors.append(f"{device['name']}: 渲染结果为空")
        except Exception as e:
            errors.append(f"{device['name']}: {e}")

    if errors:
        print("❌ 模板验证失败:")
        for e in errors:
            print(f"  • {e}")
        sys.exit(1)
    else:
        print(f"✓ 所有模板验证通过 ({len(devices.get('devices', []))} 台设备)")

if __name__ == "__main__":
    validate_templates()

三、Jenkins Pipeline 实现

3.1 Jenkinsfile

// Jenkinsfile — 网络变更流水线

pipeline {
    agent any

    parameters {
        string(name: 'CHANGE_DESCRIPTION',
               defaultValue: '',
               description: '变更描述')
        choice(name: 'DEPLOY_MODE',
               choices: ['dry-run', 'canary', 'full'],
               description: '部署模式')
    }

    stages {
        stage('代码检查') {
            steps {
                sh 'python -m pip install -r requirements.txt'
                sh 'python scripts/validate_yaml.py'
                sh 'python scripts/validate_templates.py'
            }
        }

        stage('生成配置') {
            steps {
                sh 'python scripts/render_configs.py --output output/'
                sh 'python scripts/generate_diff.py --output output/diff.txt'
            }
            post {
                success {
                    archiveArtifacts artifacts: 'output/**/*'
                }
            }
        }

        stage('审批') {
            when {
                expression { params.DEPLOY_MODE != 'dry-run' }
            }
            steps {
                script {
                    // 生成审批请求
                    def change_summary = readFile('output/diff.txt')
                    mail to: 'network-team@corp.com',
                         subject: "[审批] 网络变更 - ${params.CHANGE_DESCRIPTION}",
                         body: "变更摘要:\n\n${change_summary}\n\n请审批: ${env.BUILD_URL}"
                }
                input message: '是否批准变更？',
                      ok: '批准',
                      submitterParameter: 'approver'
            }
        }

        stage('灰度部署') {
            when {
                expression {
                    params.DEPLOY_MODE == 'canary' ||
                    params.DEPLOY_MODE == 'full'
                }
            }
            steps {
                // 先部署 1 台
                sh 'python scripts/deploy.py --limit "canary-group" --check'
                sh 'python scripts/deploy.py --limit "canary-group"'
                sh 'python scripts/verify.py --limit "canary-group"'
            }
        }

        stage('全量部署') {
            when {
                expression { params.DEPLOY_MODE == 'full' }
            }
            steps {
                sh 'python scripts/deploy.py --batch 3'
            }
        }

        stage('验证') {
            steps {
                sh 'python scripts/verify_deployment.py'
            }
            post {
                success {
                    echo '✓ 变更验证通过'
                }
                failure {
                    echo '✗ 变更验证失败，触发回滚'
                    sh 'python scripts/rollback.py'
                }
            }
        }
    }

    post {
        always {
            // 发送变更结果
            emailext(
                to: 'network-team@corp.com',
                subject: "[完成] 网络变更 - ${params.CHANGE_DESCRIPTION}",
                body: "变更结果: ${currentBuild.currentResult}\n\n详情: ${env.BUILD_URL}"
            )
        }
    }
}

四、灰度发布与回滚

4.1 分批部署

#!/usr/bin/env python3
# scripts/deploy_rolling.py — 滚动部署与灰度发布

import argparse
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger
import yaml
import time

log = setup_logger("rolling_deploy")

# 灰度分组
CANARY_GROUP = ["ACC-SW01", "ACC-SW02"]  # 先验证的设备

def deploy_device(device, config_file):
    """部署单台设备"""
    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return False, "连接失败"

    # 先备份
    backup = connector.backup_config(conn)
    backup_file = f"backup/pre_deploy_{device['name']}.cfg"
    with open(backup_file, "w") as f:
        f.write(backup)

    try:
        # 读取配置
        with open(config_file) as f:
            config = f.read()

        # 应用配置
        output = conn.send_config_set(config.split("\n"))
        conn.save()

        log.info(f"✓ {device['name']}: 部署成功")
        return True, "成功"

    except Exception as e:
        log.error(f"✗ {device['name']}: {e}")
        # 自动回滚
        with open(backup_file) as f:
            backup_config = f.read()
        conn.send_config_set(backup_config.split("\n"))
        conn.save()
        return False, f"已回滚: {e}"
    finally:
        connector.close(conn)

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--batch", type=int, default=1,
                       help="每批设备数")
    parser.add_argument("--limit", help="设备范围（canary/all/设备名）")
    parser.add_argument("--check", action="store_true",
                       help="仅检查不下发")
    args = parser.parse_args()

    connector = DeviceConnector()

    # 确定目标设备
    if args.limit == "canary":
        devices = [d for d in connector.devices
                   if d["name"] in CANARY_GROUP]
    elif args.limit:
        dev = connector.get_device(args.limit)
        devices = [dev] if dev else []
    else:
        devices = connector.devices

    log.info(f"目标设备: {[d['name'] for d in devices]}")
    log.info(f"批次大小: {args.batch}")

    if args.check:
        log.info("--- 检查模式 ---")
        for dev in devices:
            log.info(f"  将部署: {dev['name']} ({dev['mgmt_ip']})")
        return

    # 分批部署
    for i in range(0, len(devices), args.batch):
        batch = devices[i:i + args.batch]
        log.info(f"\n=== 部署批次 {i//args.batch + 1}: "
                 f"{[d['name'] for d in batch]} ===")

        for dev in batch:
            config_file = f"rendered/{dev['name']}_config.cfg"
            if not os.path.exists(config_file):
                log.warning(f"跳过 {dev['name']}: 配置文件不存在")
                continue

            ok, msg = deploy_device(dev, config_file)
            status = "✓" if ok else "✗"
            log.info(f"[{status}] {dev['name']}: {msg}")

            # 批次内等待验证
            if ok and args.batch == 1:
                time.sleep(5)
                log.info(f"  等待验证...")

    log.info("=== 部署完成 ===")

if __name__ == "__main__":
    main()

4.2 自动回滚

#!/usr/bin/env python3
# scripts/rollback.py — 自动回滚

import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

from lib.connector import DeviceConnector
from lib.logger import setup_logger

log = setup_logger("rollback")

def rollback_device(device):
    """回滚单台设备"""
    backup_file = f"backup/pre_deploy_{device['name']}.cfg"
    if not os.path.exists(backup_file):
        log.error(f"{device['name']}: 无备份文件，无法回滚")
        return False

    connector = DeviceConnector()
    conn = connector.connect_netmiko(device)
    if not conn:
        return False

    try:
        with open(backup_file) as f:
            config = f.read()

        conn.send_config_set(config.split("\n"))
        conn.save()
        log.info(f"✓ {device['name']}: 回滚成功")
        return True
    except Exception as e:
        log.error(f"✗ {device['name']}: 回滚失败 {e}")
        return False
    finally:
        connector.close(conn)

def main():
    log.info("=== 全网回滚 ===")
    connector = DeviceConnector()
    success = 0

    for dev in connector.devices:
        if rollback_device(dev):
            success += 1

    log.info(f"回滚完成: {success}/{len(connector.devices)}")

if __name__ == "__main__":
    main()

五、变更审批与通知

5.1 Webhook 集成

#!/usr/bin/env python3
# scripts/notify_approvers.py — 审批通知

import requests
import json
import sys

def send_approval_request(change_info):
    """发送审批请求"""
    # 示例：发送到企业微信/钉钉/Slack

    webhook_url = "https://hooks.example.com/network-approval"

    message = {
        "msgtype": "markdown",
        "markdown": {
            "title": "网络变更审批请求",
            "text": f"""
### 网络变更审批请求

**变更描述**: {change_info.get('description', 'N/A')}
**变更范围**: {change_info.get('scope', 'N/A')}
**变更文件**: {change_info.get('files', 'N/A')}

**变更摘要**:

{change_info.get('diff_summary', 'N/A')[:500]}

请点击链接审批: {change_info.get('approval_url', '#')}
            """
        }
    }

    try:
        resp = requests.post(webhook_url, json=message, timeout=10)
        print(f"✓ 审批通知已发送: {resp.status_code}")
    except Exception as e:
        print(f"✗ 通知发送失败: {e}")

def generate_change_ticket(change_info):
    """自动生成变更工单"""
    # 示例：创建 ITSM 工单
    ticket = {
        "title": f"[自动] 网络配置变更 - {change_info['description']}",
        "description": change_info.get('diff_summary', ''),
        "category": "Network Change",
        "priority": change_info.get('priority', 'Normal'),
        "affected_items": change_info.get('affected_devices', []),
        "requester": "CI/CD Pipeline",
    }

    # 调用 ITSM API
    # resp = requests.post("https://itsm.corp.com/api/tickets", json=ticket)

    print(f"变更工单已创建: {ticket['title']}")
    return ticket

if __name__ == "__main__":
    change = {
        "description": "新增 VLAN 100-200",
        "scope": "ACC-SW01, ACC-SW02",
        "files": "config/vlans.yml",
        "diff_summary": "+ 创建 VLAN 100 (SALES)\n+ 创建 VLAN 200 (ENG)",
        "approval_url": "https://git.corp.com/network/merge_requests/42",
        "affected_devices": ["ACC-SW01", "ACC-SW02"],
    }
    send_approval_request(change)
    generate_change_ticket(change)

六、最佳实践

网络 CI/CD 实施要点：

  1. 版本控制
  ┌─ 所有配置代码入 Git
  ├─ 分支策略（develop → staging → master）
  ├─ 每次变更必须 MR/PR
  └─ 标签对应发布版本

  2. 自动化测试
  ┌─ 语法检查（YAML/JSON/Jinja2）
  ├─ IP 冲突检测
  ├─ 合规策略检查
  ├─ 渲染结果 diff
  └─ 沙箱环境验证

  3. 安全部署
  ┌─ 灰度发布（先 1 台→10%→50%→全量）
  ├─ 自动回滚（部署失败 5 分钟内）
  ├─ 变更窗口（仅允许维护窗口执行）
  └─ 所有操作记录审计日志

  4. 持续改进
  ┌─ 统计部署成功率
  ├─ 分析回滚原因
  ├─ 优化测试覆盖率
  └─ 缩短变更周期

下篇预告：第317篇 — Network-as-Code，将介绍将网络配置视为代码的管理理念，以及 GitOps 在网络自动化中的应用。