第317篇：Network-as-Code 网络即代码

关键词

Network-as-Code、GitOps、声明式网络、基础设施即代码、配置即代码、版本控制、网络 DevOps

一、Network-as-Code 理念

1.1 什么是 Network-as-Code

Network-as-Code（网络即代码）是将 网络配置视为软件代码 来管理的理念和方法论：

传统网络 vs Network-as-Code：

  传统网络：
  ┌─ 配置是手工输入的 CLI 命令
  ├─ 变更靠工程师记忆和经验
  ├─ 状态存放在设备本地
  ├─ 没有版本控制
  └─ 配置漂移是常态

  Network-as-Code：
  ┌─ 配置是 Git 仓库中的代码文件
  ├─ 变更通过 PR/MR 审查
  ├─ 期望状态声明在代码中
  ├─ 自动同步到设备
  └─ 配置漂移自动修复

  核心理念：
  ┌─ 声明式：描述"应该是什么"，而非"怎么配置"
  ├─ 版本化：所有变更可追溯
  ├─ 自动化：从代码到设备自动同步
  ├─ 可重复：相同输入产生相同输出
  └─ 可测试：变更前可验证

1.2 GitOps 网络模型

GitOps for Network：

二、GitOps 目录结构

2.1 标准仓库结构

network-gitops/
├── README.md
├── .gitlab-ci.yml          # CI/CD 流水线
│
├── config/                 # 网络配置数据（YAML）
│   ├── global.yml          # 全局参数
│   ├── devices.yml         # 设备清单
│   ├── sites/
│   │   ├── beijing.yml     # 北京站点
│   │   └── shanghai.yml    # 上海站点
│   ├── vlans/
│   │   ├── prod.yml        # 生产 VLAN
│   │   └── dev.yml         # 开发 VLAN
│   └── bgp/
│       ├── asn.yml         # AS 号规划
│       └── peers.yml       # BGP 邻居
│
├── templates/              # Jinja2 配置模板
│   ├── base/
│   │   ├── base.j2
│   │   └── macros.j2
│   ├── huawei/
│   │   ├── core_router.j2
│   │   ├── access_switch.j2
│   │   └── firewall.j2
│   └── cisco/
│       ├── ios_xr_router.j2
│       └── nxos_switch.j2
│
├── policies/               # 策略定义
│   ├── compliance.rego     # OPA 合规策略
│   ├── security.rego       # 安全策略
│   └── naming.rego         # 命名规范
│
├── tests/                  # 自动化测试
│   ├── test_render.py
│   ├── test_compliance.py
│   └── test_connectivity.py
│
├── scripts/                # 自动化脚本
│   ├── deploy.py
│   ├── verify.py
│   ├── validate.py
│   └── rollback.py
│
├── baselines/              # 基线配置
│   ├── CORE-SW01.cfg
│   └── ACC-SW01.cfg
│
└── rendered/               # 渲染输出（.gitignore）

2.2 声明式配置示例

# config/global.yml — 全局配置参数
---
ntp:
  servers:
    - 203.0.113.1
    - 203.0.113.2
  timezone: "CST+08:00"

snmp:
  location: "Beijing-DC"
  contact: "network@corp.com"
  communities:
    - name: "public-ro"
      type: read-only
      acl: "ACL-SNMP-RO"

syslog:
  servers:
    - 10.0.0.100
    - 10.0.0.101
  facility: local5

dns:
  servers:
    - 10.0.0.10
    - 10.0.0.11
  domain: corp.local

# config/devices.yml — 设备清单
---
devices:
  - name: CORE-SW01
    role: core
    mgmt_ip: 10.255.1.1
    device_type: huawei
    model: CE12808
    site: beijing
    interfaces:
      - name: 40GE0/0/1
        description: "Spine-Leaf-01"
        mode: trunk
        vlans: all
      - name: 40GE0/0/2
        description: "Spine-Leaf-02"
        mode: trunk
        vlans: all

  - name: ACC-SW01
    role: access
    mgmt_ip: 10.255.2.1
    device_type: huawei
    model: S5735-L48T4XE
    site: beijing
    upstream: CORE-SW01
    vlans: [10, 20, 30]
    interfaces:
      - name: GigabitEthernet0/0/1
        description: "Uplink"
        mode: trunk
        vlans: "10 20 30"
      - name: GigabitEthernet0/0/2
        description: "Office-F1"
        mode: access
        vlan: 10
      - name: GigabitEthernet0/0/3
        description: "Office-F2"
        mode: access
        vlan: 20

三、声明式同步引擎

3.1 期望状态与实际状态

#!/usr/bin/env python3
# scripts/gitops_sync.py — GitOps 同步引擎

"""
GitOps 同步引擎
比较 Git 中声明的期望状态与设备实际状态
自动修复配置漂移
"""

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 json

log = setup_logger("gitops")

class GitOpsSync:
    """GitOps 同步引擎"""

    def __init__(self, repo_path: str):
        self.repo_path = repo_path
        self.connector = DeviceConnector()

    def load_desired_state(self) -> dict:
        """加载期望状态（从 Git）"""
        devices_file = os.path.join(self.repo_path, "config/devices.yml")
        with open(devices_file) as f:
            return yaml.safe_load(f)

    def get_actual_state(self, device: dict) -> dict:
        """获取设备实际状态"""
        conn = self.connector.connect_netmiko(device)
        if not conn:
            return {"status": "unreachable"}

        actual = {"status": "connected"}
        try:
            # 主机名
            hostname = conn.send_command(
                "display current-configuration | include sysname"
            )
            actual["hostname"] = hostname.replace("sysname", "").strip()

            # VLAN
            vlan_output = conn.send_command("display vlan")
            actual["vlans"] = [
                line.split()[0] for line in vlan_output.split("\n")
                if line.strip() and line.split()[0].isdigit()
            ]

            # 接口配置
            intf_output = conn.send_command("display current-configuration interface")
            actual["interfaces"] = intf_output

        except Exception as e:
            actual["status"] = f"error: {e}"
        finally:
            self.connector.close(conn)

        return actual

    def diff_state(self, desired: dict, actual: dict) -> list:
        """比较期望状态与实际状态的差异"""
        diffs = []

        # 比较 VLAN
        desired_vlans = set(
            str(v["id"]) for v in desired.get("vlans", [])
        )
        actual_vlans = set(actual.get("vlans", []))

        missing_vlans = desired_vlans - actual_vlans
        extra_vlans = actual_vlans - desired_vlans

        for vlan in missing_vlans:
            diffs.append({
                "type": "add",
                "object": f"vlan {vlan}",
                "desired": f"VLAN {vlan} 应存在",
                "actual": "不存在",
            })
        for vlan in extra_vlans:
            diffs.append({
                "type": "remove",
                "object": f"vlan {vlan}",
                "desired": f"VLAN {vlan} 不应存在",
                "actual": "存在",
            })

        return diffs

    def auto_remediate(self, device: dict, diffs: list) -> bool:
        """自动修复配置漂移"""
        conn = self.connector.connect_netmiko(device)
        if not conn:
            return False

        commands = []
        for diff in diffs:
            if diff["type"] == "add" and "vlan" in diff["object"]:
                vlan_id = diff["object"].split()[-1]
                commands.append(f"vlan {vlan_id}")
                commands.append("quit")
            elif diff["type"] == "remove" and "vlan" in diff["object"]:
                vlan_id = diff["object"].split()[-1]
                commands.append(f"undo vlan {vlan_id}")

        if commands:
            try:
                conn.send_config_set(commands)
                conn.save()
                log.info(f"✓ {device['name']}: 已修复 {len(commands)//2} 个漂移")
                return True
            except Exception as e:
                log.error(f"✗ {device['name']}: 修复失败 {e}")
                return False
            finally:
                self.connector.close(conn)
        return True

    def sync(self):
        """执行同步"""
        log.info("=== GitOps 同步开始 ===")

        desired = self.load_desired_state()
        total_drifts = 0

        for dev in desired.get("devices", []):
            desired_state = dev
            actual_state = self.get_actual_state(dev)

            if actual_state.get("status") != "connected":
                log.warning(f"  {dev['name']}: 无法获取实际状态")
                continue

            diffs = self.diff_state(desired_state, actual_state)
            if diffs:
                log.info(f"  {dev['name']}: 发现 {len(diffs)} 个漂移")
                for d in diffs:
                    log.info(f"    {d['type']}: {d['object']}")
                self.auto_remediate(dev, diffs)
                total_drifts += len(diffs)
            else:
                log.info(f"  ✓ {dev['name']}: 配置一致")

        log.info(f"=== 同步完成，共修复 {total_drifts} 个漂移 ===")

if __name__ == "__main__":
    sync = GitOpsSync(repo_path=".")
    sync.sync()

四、策略即代码

4.1 OPA 合规策略

# policies/compliance.rego — OPA 合规策略

package network.compliance

# 所有设备必须配置 NTP
deny[msg] {
    device := input.devices[_]
    not device.config.ntp
    msg = sprintf("%v: 缺少 NTP 配置", [device.name])
}

# 必须使用 SSH v2
deny[msg] {
    device := input.devices[_]
    ssh := device.config.ssh
    ssh.version != 2
    msg = sprintf("%v: SSH 版本必须为 v2, 当前为 %v",
                  [device.name, ssh.version])
}

# SNMP v3 优先
warn[msg] {
    device := input.devices[_]
    snmp := device.config.snmp
    snmp.version == "v2c"
    msg = sprintf("%v: 建议升级 SNMP 到 v3", [device.name])
}

# 接口命名规范检查
deny[msg] {
    device := input.devices[_]
    intf := device.config.interfaces[_]
    not re_match("^[A-Z][a-z]+[0-9/]+$", intf.name)
    msg = sprintf("%v: 接口名 %v 不符合命名规范",
                  [device.name, intf.name])
}

4.2 Python 策略检查

#!/usr/bin/env python3
# scripts/policy_check.py — 策略检查

import yaml
import re

class PolicyEngine:
    """策略引擎"""

    def __init__(self, policies: list):
        self.policies = policies

    def check_device(self, device: dict) -> list:
        """检查单台设备"""
        violations = []

        for policy in self.policies:
            rule = policy.get("rule", "")
            field = policy.get("field", "")

            if rule == "required":
                # 必填字段检查
                if field not in device and field not in device.get("config", {}):
                    violations.append({
                        "device": device["name"],
                        "policy": policy["name"],
                        "severity": policy.get("severity", "error"),
                        "message": f"缺少必填字段: {field}",
                    })

            elif rule == "pattern":
                # 正则匹配
                value = device.get(field, "")
                pattern = policy.get("pattern", "")
                if value and not re.match(pattern, value):
                    violations.append({
                        "device": device["name"],
                        "policy": policy["name"],
                        "severity": policy.get("severity", "error"),
                        "message": f"{field}='{value}' 不匹配 {pattern}",
                    })

            elif rule == "value_in":
                # 值范围检查
                value = device.get(field)
                allowed = policy.get("values", [])
                if value and value not in allowed:
                    violations.append({
                        "device": device["name"],
                        "policy": policy["name"],
                        "severity": policy.get("severity", "error"),
                        "message": f"{field}='{value}' 不在允许值 {allowed} 中",
                    })

        return violations

# 策略定义
POLICIES = [
    {"name": "hostname-format", "rule": "pattern",
     "field": "name", "pattern": r"^[A-Z0-9-]+$",
     "severity": "error"},
    {"name": "ntp-required", "rule": "required",
     "field": "ntp_server", "severity": "error"},
    {"name": "snmp-required", "rule": "required",
     "field": "snmp_community", "severity": "warning"},
    {"name": "device-type", "rule": "value_in",
     "field": "device_type", "values": ["huawei", "cisco_ios", "juniper_junos"],
     "severity": "error"},
]

engine = PolicyEngine(POLICIES)
devices = yaml.safe_load("""
devices:
  - name: ACC-SW01
    device_type: huawei
    ntp_server: 203.0.113.1
  - name: invalid-name
    device_type: unknown
""")

for dev in devices.get("devices", []):
    violations = engine.check_device(dev)
    for v in violations:
        print(f"[{v['severity'].upper()}] {v['device']}: {v['message']}")

五、Network-as-Code 落地路径

5.1 成熟度模型

Network-as-Code 成熟度：

  L1: 手工脚本
  ┌─ 配置脚本保存在本地
  ├─ 工程师手动执行
  └─ 无版本控制

  L2: 代码入 Git
  ┌─ 配置数据和模板入 Git
  ├─ 手动触发部署
  ├─ 有版本历史
  └─ CI 做语法检查

  L3: 自动化流水线
  ┌─ Git Push → CI/CD 自动部署
  ├─ 有审批流程
  ├─ 自动测试
  └─ 灰度发布

  L4: GitOps 自愈
  ┌─ Git 作为单一事实源
  ├─ 自动检测配置漂移
  ├─ 自动修复漂移
  └─ 全网状态 Git 中一致

  L5: 自治网络
  ┌─ AI 驱动配置优化
  ├─ 自动规划
  ├─ 自动优化
  └─ 闭环自愈

六、总结

Network-as-Code 核心价值：

  配置即代码
  ┌─ YAML 描述期望状态
  ├─ Jinja2 模板生成配置
  └─ Git 版本控制所有变更

  声明式管理
  ┌─ 描述"要什么"而非"怎么做"
  ├─ 自动修复配置漂移
  └─ 单一事实源

  流水线自动化
  ┌─ CI：验证 + 测试
  ├─ CD：审批 + 灰度 + 部署
  └─ GitOps：持续同步

  工程文化转变
  ┌─ 从"消防员"到"软件工程师"
  ├─ 从"CLI 高手"到"代码作者"
  └─ 从"手工操作"到"自动化编排"

下篇预告：第318篇 — Git 配置管理，将介绍 Git 在网络自动化中的应用，包括分支策略、配置版本管理和团队协作最佳实践。