第322篇：CI/CD 在网络变更中的应用

关键词：CI/CD、持续集成、持续交付、网络变更、GitLab CI、Jenkins、灰度发布、配置验证

一、引言

传统网络变更流程存在明显痛点：变更窗口集中在深夜、回滚困难、缺乏自动化验证、人工操作易出错。这些问题的根源在于网络变更长期被视为"手工操作"，而非"可验证的软件交付"。

CI/CD（Continuous Integration / Continuous Delivery）理念源自软件工程，其核心思想是：任何变更都应经过自动化构建 → 自动化测试 → 自动化部署的完整流程。将 CI/CD 引入网络变更，意味着每一条配置指令在到达生产设备之前，都要经历严格的自动化校验。

二、网络 CI/CD 架构设计

2.1 整体流程

整体流程 - 图示

2.2 核心组件

组件 工具选型 职责
版本控制 Git（GitLab / GitHub） 配置代码存储、分支管理、MR 评审
CI/CD 引擎 GitLab CI / Jenkins Pipeline 编排、阶段触发、通知
静态分析 Batfish、Avisynth 配置语法检查、冲突检测
单元/集成测试 pytest + pytest-network 接口状态、路由表、ACL 验证
仿真环境 Containerlab、EVE-NG 变更前隔离验证
部署工具 Ansible、 NAPALM 配置下发、回滚

三、GitLab CI 配置变更 Pipeline

3.1 .gitlab-ci.yml 配置

stages:
  - lint
  - unit-test
  - integration-test
  - staging-deploy
  - review
  - production-deploy

variables:
  PYTEST_ADDOPTS: "-v --tb=short"
  ANSIBLE_HOST_KEY_CHECKING: "False"

# 第一阶段：静态检查
config_lint:
  stage: lint
  image: python:3.11-slim
  script:
    - pip install pybatfish cfn-lint yamllint
    - yamllint configs/*.yaml
    - python scripts/batfish_check.py --config configs/$DEVICE_GROUP/
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"

# 第二阶段：单元测试
unit_test:
  stage: unit-test
  image: python:3.11-slim
  script:
    - pip install -r requirements-test.txt
    - pytest tests/unit/ -v
  needs:
    - config_lint

# 第三阶段：集成测试
integration_test:
  stage: integration-test
  image: python:3.11-slim
  script:
    - pip install -r requirements-test.txt
    - pytest tests/integration/ -v --junitxml=report.xml
  artifacts:
    reports:
      junit: report.xml
  needs:
    - unit_test

# 第四阶段：仿真环境部署
staging_deploy:
  stage: staging-deploy
  image: ansible/ansible:latest
  script:
    - ansible-playbook playbooks/staging/deploy.yml
      -e "env=staging"
      -e "config_ref=$CI_MERGE_REQUEST_SOURCE_BRANCH_NAME"
  environment:
    name: staging
  needs:
    - integration_test

# 第五阶段：人工审批
review:
  stage: review
  image: alpine:latest
  script:
    - echo "Waiting for manual approval..."
  rules:
    - if: $CI_MERGE_REQUEST_EVENT_TYPE == "approve"
      when: manual
  needs:
    - staging_deploy

# 第六阶段：生产环境灰度部署
prod_deploy:
  stage: production-deploy
  image: ansible/ansible:latest
  script:
    # 灰度：先部署 20% 设备
    - ansible-playbook playbooks/prod/deploy.yml
      -e "env=production"
      -e "config_ref=$CI_MERGE_REQUEST_SOURCE_BRANCH_NAME"
      -e "deploy_batch=canary"
      --limit "group[0:2]"
    # 等待验证期
    - sleep 300
    # 全量部署
    - ansible-playbook playbooks/prod/deploy.yml
      -e "env=production"
      -e "config_ref=$CI_MERGE_REQUEST_SOURCE_BRANCH_NAME"
      -e "deploy_batch=full"
  environment:
    name: production
  rules:
    - if: $CI_MERGE_REQUEST_EVENT_TYPE == "approve"
      when: manual
  needs:
    - review

3.2 Batfish 静态分析检查

Batfish 是目前业界最强的网络配置分析工具，可在不连接真实设备的情况下检测配置问题：

# scripts/batfish_check.py
from pybatfish.client import commands as bf

def run_batfish_check(config_dir):
    """运行 Batfish 配置检查"""
    # 初始化 Batfish 会话
    bf.init_session()

    # 创建工作空间
    workspace = bf.create_workspace("network_validation")
    bf.set_workspace(workspace)

    # 初始化网络快照
    snapshot = bf.init_snapshot(config_dir)

    # 检查项 1: BGP 会话是否可建立
    result = bf.q.bgp_session_status().answer()
    failed_sessions = result.frame[
        result.frame['Status'] != 'ESTABLISHED'
    ]
    if not failed_sessions.empty:
        print("BGP 会话异常:")
        print(failed_sessions.to_string())
        return False

    # 检查项 2: ACL 是否阻断预期流量
    result = bf.q.tracingOptions(
        headers={'ip': {'dstIp': '10.0.0.0/8'}},
        edgeRoles='ToDatacenterRouters'
    ).answer()

    # 检查项 3: 路由泄露检测
    result = bf.q.routeLeak(
        source='External',
        destination='Internal'
    ).answer()

    bf.delete_workspace(workspace)
    return True

if __name__ == '__main__':
    import sys
    success = run_batfish_check(sys.argv[1])
    sys.exit(0 if success else 1)

四、灰度发布与回滚机制

4.1 灰度部署策略

# inventory/grayscale.yml
all:
  children:
    canary_group:      # 金丝雀组（首批）
      hosts:
        PE-BJ-01:
        PE-SH-01:
    wave_1_group:      # 第一波
      hosts:
        PE-BJ-02:
        PE-SH-02:
        PE-GZ-01:
    wave_2_group:      # 第二波
      hosts:
        PE-BJ-03:
        PE-SH-03:
        PE-GZ-02:

# scripts/grayscale_deploy.py
import time
import ansible_runner

GRAYSCALE_CONFIG = {
    'waves': [
        {'name': 'canary', 'inventory': 'canary_group', 'wait_seconds': 300},
        {'name': 'wave_1', 'inventory': 'wave_1_group', 'wait_seconds': 600},
        {'name': 'wave_2', 'inventory': 'wave_2_group', 'wait_seconds': 600},
    ],
    'rollback_on_failure': True,
    'health_check': 'check_device_health.py'
}

def grayscale_deploy(config_template):
    """灰度部署主流程"""
    for wave in GRAYSCALE_CONFIG['waves']:
        print(f"=== 部署批次: {wave['name']} ===")

        # 1. 下发配置
        result = deploy_batch(wave['inventory'], config_template)

        if not result['success']:
            # 部署失败，立即回滚当前批次及之前批次
            rollback(wave['name'])
            raise Exception(f"批次 {wave['name']} 部署失败")

        # 2. 等待验证期
        print(f"等待 {wave['wait_seconds']} 秒验证期...")
        time.sleep(wave['wait_seconds'])

        # 3. 健康检查
        health = run_health_check(wave['inventory'])
        if not health['healthy']:
            rollback(wave['name'])
            raise Exception(f"批次 {wave['name']} 健康检查失败")

        print(f"批次 {wave['name']} 验证通过，继续下一批")

    print("全部批次部署完成")

4.2 自动回滚机制

# playbooks/prod/rollback.yml
---
- name: 自动回滚网络配置
  hosts: "{{ target_group }}"
  gather_facts: no
  vars:
    backup_dir: "/opt/network-backups/{{ inventory_hostname }}"

  tasks:
    - name: 获取最近一次成功备份
      ansible.builtin.find:
        paths: "{{ backup_dir }}"
        patterns: "*.cfg"
      register: backups
      delegate_to: localhost

    - name: 回滚配置
      ansible.netcommon.config:
        src: "{{ backups.files[0].path }}"
        backup: yes
        backup_options:
          dir: "{{ backup_dir }}/rollback_{{ ansible_date_time.iso8601_basic_short }}"

    - name: 验证回滚后配置
      ansible.builtin.command:
        cmd: "python scripts/verify_config.py {{ inventory_hostname }}"
      delegate_to: localhost
      register: verify_result

    - name: 失败告警
      ansible.builtin.uri:
        url: "https://alerts.example.com/webhook/rollback"
        method: POST
        body: |
          {
            "device": "{{ inventory_hostname }}",
            "rollback_time": "{{ ansible_date_time.iso8601 }}",
            "reason": "自动回滚触发"
          }
        body_format: json
      when: verify_result.rc != 0

五、自动化运维平台架构设计

5.1 前后端分离架构

前后端分离架构 - 图示

5.2 API 接口设计

# api/main.py - FastAPI 自动化运维平台
"""
网络自动化运维平台 REST API
支持：设备管理、变更管理、配置采集、健康检查
"""
from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
from datetime import datetime
import uuid

app = FastAPI(
    title="网络自动化运维平台",
    version="1.0.0",
    docs_url="/api/docs"
)


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

class DeviceCreate(BaseModel):
    name: str = Field(..., description="设备名称")
    ip: str = Field(..., description="管理 IP")
    platform: str = Field(default='huawei_vrpv8', description="设备平台")
    role: str = Field(default='access', description="设备角色")
    username: str = Field(default='admin')
    password: str
    site: str = Field(default='default', description="站点")
    tags: List[str] = Field(default_factory=list)


class DeviceUpdate(BaseModel):
    name: Optional[str] = None
    ip: Optional[str] = None
    role: Optional[str] = None
    tags: Optional[List[str]] = None


class ChangeRequest(BaseModel):
    title: str = Field(..., description="变更标题")
    description: str = Field(..., description="变更描述")
    device_ids: List[str] = Field(..., description="目标设备 ID 列表")
    config_template: str = Field(..., description="配置模板路径")
    config_params: Dict = Field(default_factory=dict, description="模板参数")
    change_type: str = Field(default='P2', description="变更类型 P1-P4")
    change_window_start: Optional[datetime] = None
    rollback_on_failure: bool = Field(default=True)


class HealthCheckResult(BaseModel):
    device_name: str
    cpu_usage: float
    memory_usage: float
    bgp_sessions: int
    interface_status: Dict
    health_score: float
    timestamp: str


# ============ 设备管理 API ============

@app.post("/api/v1/devices", response_model=Dict, status_code=201)
async def create_device(device: DeviceCreate):
    """创建设备"""
    # 保存到数据库
    db_device = save_device_to_db(device)
    return {"id": db_device['id'], "name": device.name}


@app.get("/api/v1/devices", response_model=List[Dict])
async def list_devices(role: Optional[str] = None,
                       site: Optional[str] = None,
                       tag: Optional[str] = None):
    """设备列表（支持过滤）"""
    return query_devices(role=role, site=site, tag=tag)


@app.get("/api/v1/devices/{device_id}", response_model=Dict)
async def get_device(device_id: str):
    """获取设备详情"""
    device = get_device_from_db(device_id)
    if not device:
        raise HTTPException(status_code=404, detail="设备不存在")
    return device


@app.put("/api/v1/devices/{device_id}", response_model=Dict)
async def update_device(device_id: str, device: DeviceUpdate):
    """更新设备信息"""
    return update_device_in_db(device_id, device)


@app.delete("/api/v1/devices/{device_id}")
async def delete_device(device_id: str):
    """删除设备"""
    delete_device_from_db(device_id)
    return {"status": "deleted"}


@app.post("/api/v1/devices/{device_id}/collect", response_model=Dict)
async def collect_device(device_id: str, background_tasks: BackgroundTasks):
    """触发单设备信息采��"""
    background_tasks.add_task(run_collection, device_id)
    return {"task_id": str(uuid.uuid4()), "status": "started"}


@app.post("/api/v1/devices/batch-collect", response_model=Dict)
async def batch_collect(devices: List[str], background_tasks: BackgroundTasks):
    """批量设备信息采集"""
    task_id = str(uuid.uuid4())
    background_tasks.add_task(run_batch_collection, devices, task_id)
    return {"task_id": task_id, "status": "started"}


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

@app.post("/api/v1/changes", response_model=Dict, status_code=201)
async def create_change(change: ChangeRequest, background_tasks: BackgroundTasks):
    """创建变更请求"""
    change_id = str(uuid.uuid4())
    save_change_request(change_id, change)
    return {"change_id": change_id, "status": "created"}


@app.post("/api/v1/changes/{change_id}/execute")
async def execute_change(change_id: str, background_tasks: BackgroundTasks):
    """执行变更"""
    background_tasks.add_task(run_change_execution, change_id)
    return {"status": "executing", "change_id": change_id}


@app.get("/api/v1/changes/{change_id}/status", response_model=Dict)
async def get_change_status(change_id: str):
    """获取变更状态"""
    return get_change_status_from_db(change_id)


@app.post("/api/v1/changes/{change_id}/rollback")
async def rollback_change(change_id: str):
    """回滚变更"""
    run_rollback(change_id)
    return {"status": "rolling_back", "change_id": change_id}


@app.get("/api/v1/changes", response_model=List[Dict])
async def list_changes(status: Optional[str] = None,
                       limit: int = 50):
    """变更列表"""
    return query_changes(status=status, limit=limit)


# ============ 健康检查 API ============

@app.get("/api/v1/health", response_model=Dict)
async def system_health():
    """平台健康状态"""
    return {
        "status": "healthy",
        "database": "connected",
        "redis": "connected",
        "workers_online": get_worker_count(),
    }


@app.get("/api/v1/devices/{device_id}/health", response_model=HealthCheckResult)
async def device_health(device_id: str):
    """设备健康状态"""
    return get_device_health(device_id)


@app.get("/api/v1/devices/health-summary", response_model=Dict)
async def health_summary():
    """设备健康汇总"""
    return {
        "total_devices": get_total_devices(),
        "healthy": get_healthy_count(),
        "warning": get_warning_count(),
        "critical": get_critical_count(),
    }


# ============ 认证与权限 ============

class LoginRequest(BaseModel):
    username: str
    password: str


class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"
    expires_in: int = 3600


@app.post("/api/v1/auth/login", response_model=TokenResponse)
async def login(credentials: LoginRequest):
    """用户登录"""
    token = generate_jwt_token(credentials.username)
    return TokenResponse(access_token=token)

5.3 数据库模型设计

# models/database.py - SQLAlchemy 数据库模型
"""
自动化运维平台数据库模型
"""
from sqlalchemy import (
    Column, String, Integer, Float, Boolean, Text, DateTime,
    ForeignKey, JSON, Enum as SAEnum, Index
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.create_engine import create_engine
from datetime import datetime

Base = declarative_base()


class Device(Base):
    """设备表"""
    __tablename__ = 'devices'

    id = Column(String(36), primary_key=True)
    name = Column(String(128), unique=True, nullable=False, index=True)
    ip = Column(String(45), nullable=False)
    platform = Column(String(64), default='huawei_vrpv8')
    role = Column(String(32), default='access')
    site = Column(String(64), default='default', index=True)
    username = Column(String(64))
    password_encrypted = Column(String(256))
    port = Column(Integer, default=22)
    tags = Column(JSON, default=list)
    last_collected_at = Column(DateTime)
    last_health_score = Column(Float)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

    # 关联
    collections = relationship("DeviceCollection", back_populates="device")
    changes = relationship("ChangeRequest", back_populates="device")


class DeviceCollection(Base):
    """设备采集记录表"""
    __tablename__ = 'device_collections'

    id = Column(String(36), primary_key=True)
    device_id = Column(String(36), ForeignKey('devices.id'), index=True)
    task_id = Column(String(36), index=True)
    cpu_usage = Column(Float)
    memory_usage = Column(Float)
    bgp_sessions = Column(Integer)
    ospf_neighbors = Column(Integer)
    interface_count = Column(Integer)
    up_interfaces = Column(Integer)
    down_interfaces = Column(Integer)
    temperature = Column(Float)
    health_score = Column(Float)
    raw_data = Column(JSON)
    errors = Column(JSON, default=list)
    collected_at = Column(DateTime, default=datetime.now)

    device = relationship("Device", back_populates="collections")


class ChangeRequest(Base):
    """变更请求表"""
    __tablename__ = 'change_requests'

    id = Column(String(36), primary_key=True)
    title = Column(String(256), nullable=False)
    description = Column(Text)
    change_type = Column(String(8), default='P2')
    status = Column(String(32), default='draft')
    # draft | submitted | approved | executing | success | failed | rolled_back
    requestor = Column(String(64))
    approver = Column(String(64))
    approved_at = Column(DateTime)
    executed_at = Column(DateTime)
    completed_at = Column(DateTime)
    config_template = Column(String(256))
    config_params = Column(JSON)
    rollback_on_failure = Column(Boolean, default=True)
    pre_backup_path = Column(String(512))
    result = Column(JSON)
    error_message = Column(Text)
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)


class ChangeDeviceMapping(Base):
    """变更-设备映射表"""
    __tablename__ = 'change_device_mappings'

    id = Column(String(36), primary_key=True)
    change_id = Column(String(36), ForeignKey('change_requests.id'), index=True)
    device_id = Column(String(36), ForeignKey('devices.id'), index=True)
    deploy_status = Column(String(32), default='pending')
    # pending | deploying | success | failed | rolled_back
    deploy_output = Column(Text)
    error_message = Column(Text)
    deployed_at = Column(DateTime)


class AuditLog(Base):
    """审计日志表"""
    __tablename__ = 'audit_logs'

    id = Column(String(36), primary_key=True)
    user = Column(String(64))
    action = Column(String(64))  # create_device | deploy_config | rollback
    resource_type = Column(String(32))  # device | change | collection
    resource_id = Column(String(36))
    details = Column(JSON)
    ip_address = Column(String(45))
    created_at = Column(DateTime, default=datetime.now)


# 数据库初始化
engine = create_engine('postgresql://admin:password@localhost:5432/network_automation')
Session = sessionmaker(bind=engine)
Base.metadata.create_all(engine)

5.4 部署配置脚本

# docker-compose.deploy.yml - 生产环境部署
version: '3.8'

services:
  # API 网关
  nginx:
    image: nginx:alpine
    ports:
      - "443:443"
      - "80:80"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf
      - ./nginx/certs:/etc/nginx/certs
    depends_on:
      - api-server
    networks:
      - platform-net

  # API 服务（2 实例实现高可用）
  api-server:
    build:
      context: ./backend
      dockerfile: Dockerfile
    environment:
      - DATABASE_URL=postgresql://admin:${DB_PASSWORD}@postgres:5432/network_automation
      - REDIS_URL=redis://redis:6379/0
      - JWT_SECRET=${JWT_SECRET}
    depends_on:
      - postgres
      - redis
    deploy:
      replicas: 2
      resources:
        limits:
          cpus: '1.0'
          memory: 512M
    networks:
      - platform-net

  # Celery Worker 集群
  celery-worker:
    build:
      context: ./backend
      dockerfile: Dockerfile
    command: celery -A tasks worker -l info -c 10
    environment:
      - DATABASE_URL=postgresql://admin:${DB_PASSWORD}@postgres:5432/network_automation
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - postgres
      - redis
    deploy:
      replicas: 3
    networks:
      - platform-net

  # 数据库
  postgres:
    image: postgres:15-alpine
    environment:
      - POSTGRES_DB=network_automation
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - pg_data:/var/lib/postgresql/data
      - ./scripts/init_db.sql:/docker-entrypoint-initdb.d/init.sql
    ports:
      - "127.0.0.1:5432:5432"
    networks:
      - platform-net

  # Redis
  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes --requirepass ${REDIS_PASSWORD}
    volumes:
      - redis_data:/data
    ports:
      - "127.0.0.1:6379:6379"
    networks:
      - platform-net

  # 前端
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
    ports:
      - "8080:80"
    networks:
      - platform-net

volumes:
  pg_data:
  redis_data:

networks:
  platform-net:
    driver: bridge

#!/bin/bash
# deploy.sh - 一键部署脚本
set -e

echo "=== 网络自动化运维平台部署脚本 ==="

# 检查依赖
command -v docker >/dev/null 2>&1 || { echo "Docker 未安装"; exit 1; }
command -v docker-compose >/dev/null 2>&1 || { echo "docker-compose 未安装"; exit 1; }

# 环境变量
source .env || { echo "请创建 .env 文件"; exit 1; }

# 构建镜像
echo "[1/4] 构建 Docker 镜像..."
docker-compose -f docker-compose.deploy.yml build

# 启动服务
echo "[2/4] 启动服务..."
docker-compose -f docker-compose.deploy.yml up -d

# 等待服务就绪
echo "[3/4] 等待服务就绪..."
sleep 10

# 运行数据库迁移
echo "[4/4] 运行数据库迁移..."
docker-compose -f docker-compose.deploy.yml exec api-server \
  flask db upgrade

echo ""
echo "=== 部署完成 ==="
echo "前端访问: http://localhost:8080"
echo "API 文档: http://localhost/api/docs"
echo "Swagger:  http://localhost/api/docs"

六、总结

维度 传统变更 CI/CD 变更 改善幅度
变更准备 手工编写配置脚本 Git 分支 + MR 模板 标准化程度大幅提升
变更验证 变更窗口临时验证 多层自动化测试 故障率降低 70%+
变更执行 深夜窗口逐台操作 Pipeline 灰度自动下发 时间从小时级降至分钟级
回滚机制 手工回退配置 一键回滚 + 自动触发 回滚时间从30分钟降至2分钟
变更审计 变更后的事后回顾 全流程 Git 记录 + CI 日志 可追溯性 100%

网络 CI/CD Pipeline 视图

网络 CI/CD Pipeline 视图 - 图示

故障案例：BGP 配置变更事故与 CI/CD 防护

某运营商在一次 BGP 路由策略变更中，因一条 ACL 规则编写错误，导致核心链路路由被意外过滤，造成 4 个地市断网 2 小时。若使用 CI/CD Pipeline，可在以下环节拦截该问题：

  1. Lint 阶段：Batfish 检测到 ACL 规则与现网路由表不匹配，阻断 Pipeline
  2. 仿真阶段：EVE-NG 仿真环境中验证 BGP 会话状态异常
  3. 灰度阶段：即使前两步漏过，金丝雀组仅影响 2 台设备，健康检查发现路由丢失后立即回滚

思考题

  1. 在网络 CI/CD Pipeline 中，如何平衡"自动化程度"与"人工审核"？哪些变更类型可以完全自动化，哪些必须保留人工审批环节？
  2. 灰度部署的批次划分有哪些策略？在核心网（如 PE 设备）与接入网（如 CE/AG 设备）的灰度策略应有什么区别？

下篇预告：第323篇《网络即代码（Network as Code）理念》——将网络配置视为代码，用软件工程的方法管理网络的全生命周期。