第310篇：YANG 数据建模语言

关键词

YANG、数据建模、NETCONF、RESTCONF、数据结构、模块、容器、列表、叶子、数据模型

一、YANG 简介

1.1 什么是 YANG

YANG（Yet Another Next Generation）是一种 数据建模语言，用于定义网络设备配置和状态数据的结构：

YANG 在网络自动化中的位置：

为什么需要 YANG： ┌─ CLI 命令是非结构化的文本 ├─ SNMP MIB 结构老旧，扩展困难 ├─ 不同厂商相同功能的数据模型不同 └─ YANG 提供统一、结构化的数据模型

1.2 YANG 与 SNMP MIB 对比

二、YANG 语法基础

2.1 模块结构

// 第310篇示例：YANG 模块定义
// 文件名：example-interface.yang

module example-interface {
    // 模块标识
    yang-version 1.1;
    namespace "http://example.com/ns/interface";
    prefix intf;

    // 导入其他模块
    import ietf-inet-types {
        prefix inet;
    }

    // 模块描述
    organization "Example Corp";
    contact "support@example.com";
    description
        "This module defines a simple interface model.";

    revision 2025-01-01 {
        description "Initial revision";
        reference "RFC 8344";
    }

    // ===== 容器（Container） =====
    container interfaces {
        description
            "Interface configuration and state.";

        // ===== 列表（List） =====
        list interface {
            key "name";                     // 主键
            description
                "A network interface.";

            // ---- 叶子（Leaf） ----
            leaf name {
                type string;
                description "Interface name.";
            }

            leaf type {
                type enumeration {
                    enum ethernet;
                    enum loopback;
                    enum vlan;
                    enum tunnel;
                }
                mandatory true;
                description "Interface type.";
            }

            leaf enabled {
                type boolean;
                default true;
                description
                    "Whether the interface is enabled.";
            }

            // ---- 叶子列表（Leaf-List） ----
            leaf-list ip-address {
                type inet:ipv4-address;
                description
                    "List of IPv4 addresses.";
            }

            // ---- 嵌套容器 ----
            container statistics {
                config false;               // 只读状态
                description
                    "Interface statistics.";

                leaf in-octets {
                    type uint64;
                    description
                        "Total input octets.";
                }

                leaf out-octets {
                    type uint64;
                    description
                        "Total output octets.";
                }

                leaf in-errors {
                    type uint32;
                    description
                        "Input errors.";
                }
            }

            // ---- 选择（Choice） ----
            choice connection-type {
                description
                    "How this interface connects.";
                case direct {
                    leaf peer-interface {
                        type leafref {
                            path "/interfaces/interface/name";
                        }
                    }
                }
                case vlan {
                    leaf vlan-id {
                        type uint16 {
                            range "1..4094";
                        }
                    }
                }
            }
        }
    }
}

2.2 YANG 核心语句

YANG 核心语法元素：

  1. 容器（Container）
  ┌─ 类似 JSON 的对象 {}
  ├─ 包含子节点
  └─ 自身没有值

  2. 列表（List）
  ┌─ 类似数组 []
  ├─ key 定义主键唯一标识
  ├─ 支持 CRUD 操作
  └─ 有序/无序列表

  3. 叶子（Leaf）
  ┌─ 单个值
  ├─ 有类型约束
  └─ 可设默认值

  4. 叶子列表（Leaf-List）
  ┌─ 同一类型的多个值
  ├─ 类似数组
  └─ 无嵌套结构

  5. 类型（Types）
  ┌─ 内置：string, uint32, boolean, enumeration
  ├─ 派生：range, pattern（正则）
  ├─ 引用：leafref
  ├─ 联合：union
  └─ 自定义：typedef

  6. 约束
  ┌─ mandatory：必填
  ├─ default：默认值
  ├─ config false：只读状态
  ├─ must：XPath 条件
  └─ when：条件约束

2.3 类型定义

// 自定义类型示例

module example-types {
    yang-version 1.1;
    namespace "http://example.com/ns/types";
    prefix ex-type;

    // ---- 自定义类型 ----
    typedef vlan-id {
        type uint16 {
            range "1..4094";
        }
        description "VLAN identifier.";
    }

    typedef ipv4-prefix {
        type string {
            pattern
              '([0-9]{1,3}\.){3}[0-9]{1,3}'
            + '/([0-9]|[12][0-9]|3[0-2])';
        }
        description "IPv4 prefix in CIDR notation.";
    }

    typedef admin-state {
        type enumeration {
            enum up {
                value 1;
                description "Administratively up.";
            }
            enum down {
                value 2;
                description "Administratively down.";
            }
        }
        description "Administrative state.";
    }

    typedef interface-ref {
        type leafref {
            path "/if:interfaces/if:interface/if:name";
        }
        description
            "Reference to an existing interface.";
    }

    // ---- 使用自定义类型 ----
    container vlan-config {
        leaf id {
            type vlan-id;
            mandatory true;
        }

        leaf name {
            type string {
                length "1..32";
            }
        }

        leaf admin-status {
            type admin-state;
            default up;
        }

        leaf-list member-interfaces {
            type interface-ref;
        }
    }
}

三、YANG 模块的构建块

3.1 模型分层结构

YANG 数据模型的分层结构：

模块（Module）： ┌─ 最高层次 ├─ 类似 Python 模块 ├─ namespace + prefix └─ 可被其他模块 import

子模块（Submodule）： ┌─ 属于某个模块 ├─ 可以拆分大模块 └─ 使用 belongs-to 关联父模块

增强（Augment）： ┌─ 向已有模块添加新节点 ├─ 厂商扩展标准模型 └─ 保持向前兼容

偏差（Deviation）： ┌─ 声明设备与标准模型的差异 ├─ 不支持某些节点 └─ 约束条件不同

YANG 模块之间的关系： | ietf-interfaces（IETF 标准） ↑ augment huawei-if-extension（华为扩展） ├─ huawei-if-vlan └─ huawei-if-ospf | | | --- | --- |

3.2 标准 YANG 模型

常用的标准 YANG 模型：

  基础模型（RFC）：
  ┌─ ietf-interfaces（RFC 8344）：接口模型
  ├─ ietf-ip（RFC 8344）：IP 地址模型
  ├─ ietf-routing（RFC 8349）：路由模型
  ├─ ietf-ospf：OSPF 模型
  ├─ ietf-bgp：BGP 模型
  ├─ ietf-yang-library：模块清单
  └─ ietf-netconf-monitoring：NETCONF 监控

  华为 YANG 模型：
  ┌─ huawei-ifm：接口管理
  ├─ huawei-vlan：VLAN 配置
  ├─ huawei-ospf：OSPF 配置
  ├─ huawei-bgp：BGP 配置
  ├─ huawei-lldp：LLDP 配置
  └─ huawei-ntp：NTP 配置

  查看设备支持的 YANG 模型：
  NETCONF: <get-schema>
  RESTCONF: GET /restconf/data/ietf-yang-library

四、YANG 与 Python 交互

4.1 使用 pyang 工具

# 安装 pyang（YANG 工具集）
python -m pip install pyang

# 验证 YANG 模块语法
pyang --check example-interface.yang

# 导出为树形结构
pyang -f tree example-interface.yang

# 导出为 YIN（XML 格式）
pyang -f yin example-interface.yang

# 导出为 UML 图
pyang -f uml example-interface.yang -o model.uml

# pyang -f tree 输出示例
module: example-interface
  +--rw interfaces
     +--rw interface* [name]
        +--rw name              string
        +--rw type              enumeration
        +--rw enabled?          boolean
        +--rw ip-address*       inet:ipv4-address
        +--ro statistics
        |  +--ro in-octets?     uint64
        |  +--ro out-octets?    uint64
        |  +--ro in-errors?     uint32
        +--rw (connection-type)?
           +--:(direct)
           |  +--rw peer-interface?      leafref
           +--:(vlan)
              +--rw vlan-id?             uint16

4.2 Python 操作 YANG 数据

#!/usr/bin/env python3
# yang_data_operations.py — YANG 数据操作

# 安装：python -m pip install pyangbind
# 或使用标准 XML/JSON 处理 YANG 编码的数据

import json
import xml.etree.ElementTree as ET

# ===== YANG 数据实例（JSON 编码） =====
# 对应 YANG 模型 example-interface 的数据实例
interface_data = {
    "ietf-interfaces:interfaces": {
        "interface": [
            {
                "name": "GigabitEthernet0/0/1",
                "type": "ethernet",
                "enabled": True,
                "ip-address": ["10.0.1.1", "10.0.1.2"],
                "statistics": {
                    "in-octets": 1024000,
                    "out-octets": 512000,
                    "in-errors": 0,
                },
            },
            {
                "name": "LoopBack0",
                "type": "loopback",
                "enabled": True,
                "ip-address": ["1.1.1.1"],
            },
        ]
    }
}

print("=== YANG 数据实例（JSON）===")
for intf in interface_data["ietf-interfaces:interfaces"]["interface"]:
    print(f"\n接口: {intf['name']}")
    print(f"  类型: {intf['type']}")
    print(f"  启用: {intf['enabled']}")
    print(f"  IP: {', '.join(intf.get('ip-address', []))}")
    if "statistics" in intf:
        s = intf["statistics"]
        print(f"  入流量: {s['in-octets']} bytes")
        print(f"  出流量: {s['out-octets']} bytes")
        print(f"  入错误: {s['in-errors']}")

# ===== XML 编码（NETCONF 格式） =====
def yang_data_to_xml(data_dict, namespace="urn:ietf:params:xml:ns:yang:ietf-interfaces"):
    """将 YANG 数据字典转为 XML（简化版）"""
    root = ET.Element("interfaces", xmlns=namespace)

    for intf in data_dict.get("interface", []):
        iface_elem = ET.SubElement(root, "interface")

        name = ET.SubElement(iface_elem, "name")
        name.text = intf["name"]

        type_elem = ET.SubElement(iface_elem, "type")
        type_elem.text = intf["type"]

        enabled = ET.SubElement(iface_elem, "enabled")
        enabled.text = str(intf.get("enabled", True)).lower()

        for ip in intf.get("ip-address", []):
            ip_elem = ET.SubElement(iface_elem, "ip-address")
            ip_elem.text = ip

    return ET.tostring(root, encoding="unicode", short_empty_elements=False)

xml_output = yang_data_to_xml(interface_data["ietf-interfaces:interfaces"])
print("\n=== XML 编码（NETCONF 格式）===")
print(xml_output)

# ===== 验证数据是否符合 YANG 约束 =====
def validate_vlan_id(vlan_id):
    """验证 VLAN ID 是否符合 YANG 约束"""
    if not isinstance(vlan_id, int):
        return False, "VLAN ID 必须是整数"
    if vlan_id < 1 or vlan_id > 4094:
        return False, "VLAN ID 必须在 1-4094 范围内"
    return True, "OK"

def validate_ip_prefix(prefix):
    """验证 IP 前缀格式"""
    import re
    pattern = r'^(\d{1,3}\.){3}\d{1,3}/(\d|[12]\d|3[0-2])$'
    if not re.match(pattern, prefix):
        return False, "IP 前缀格式无效"
    # 验证每个 octet
    ip_part = prefix.split("/")[0]
    for octet in ip_part.split("."):
        if int(octet) > 255:
            return False, f"IP 段 {octet} 超出 0-255"
    return True, "OK"

# 测试
test_cases = [100, 0, 4095, "abc"]
for tc in test_cases:
    ok, msg = validate_vlan_id(tc)
    print(f"VLAN ID {tc}: {'✓' if ok else '✗'} {msg}")

test_prefixes = ["10.0.1.0/24", "256.0.0.0/8", "10.0.0.0/33"]
for tp in test_prefixes:
    ok, msg = validate_ip_prefix(tp)
    print(f"前缀 {tp}: {'✓' if ok else '✗'} {msg}")

五、YANG 模型的增删改查

5.1 操作模型数据

#!/usr/bin/env python3
# yang_crud.py — YANG 模型数据的增删改查

import json
from copy import deepcopy

class YangDataStore:
    """简单的 YANG 数据存储（模拟设备）"""

    def __init__(self):
        self.data = {
            "interfaces": {
                "interface": []
            }
        }

    def get_interface(self, name):
        """查询接口（Read）"""
        for intf in self.data["interfaces"]["interface"]:
            if intf["name"] == name:
                return deepcopy(intf)
        return None

    def list_interfaces(self):
        """列出所有接口（List）"""
        return deepcopy(self.data["interfaces"]["interface"])

    def create_interface(self, interface):
        """创建接口（Create）"""
        # 检查是否已存在
        existing = self.get_interface(interface["name"])
        if existing:
            raise ValueError(f"接口 {interface['name']} 已存在")

        # 验证必填字段
        required = ["name", "type"]
        for field in required:
            if field not in interface:
                raise ValueError(f"缺少必填字段: {field}")

        self.data["interfaces"]["interface"].append(deepcopy(interface))
        return True

    def update_interface(self, name, updates):
        """更新接口（Update）"""
        for intf in self.data["interfaces"]["interface"]:
            if intf["name"] == name:
                intf.update(updates)
                return True
        raise ValueError(f"接口 {name} 不存在")

    def delete_interface(self, name):
        """删除接口（Delete）"""
        original_len = len(self.data["interfaces"]["interface"])
        self.data["interfaces"]["interface"] = [
            i for i in self.data["interfaces"]["interface"]
            if i["name"] != name
        ]
        if len(self.data["interfaces"]["interface"]) == original_len:
            raise ValueError(f"接口 {name} 不存在")
        return True

# ===== 测试 =====
store = YangDataStore()

# Create
store.create_interface({
    "name": "GigabitEthernet0/0/1",
    "type": "ethernet",
    "enabled": True,
    "ip-address": ["10.0.1.1"],
})
print("✓ 创建接口 GE0/0/1")

# Read
intf = store.get_interface("GigabitEthernet0/0/1")
print(f"  查询结果: {intf['name']}, {intf['type']}")

# Update
store.update_interface("GigabitEthernet0/0/1", {
    "enabled": False,
    "description": "Maintenance"
})
updated = store.get_interface("GigabitEthernet0/0/1")
print(f"  更新状态: enabled={updated['enabled']}")

# List
print(f"  接口列表: {[i['name'] for i in store.list_interfaces()]}")

# Delete
store.delete_interface("GigabitEthernet0/0/1")
print(f"  删除后数量: {len(store.list_interfaces())}")

5.2 YANG 数据过滤与查询

#!/usr/bin/env python3
# yang_filter.py — YANG 数据过滤与查询

# 模拟 YANG 模型数据
devices_data = {
    "device": [
        {
            "name": "CORE-RT01",
            "role": "core",
            "interfaces": {
                "interface": [
                    {"name": "GE0/0/1", "type": "ethernet", "ip": "10.0.12.1/30"},
                    {"name": "LoopBack0", "type": "loopback", "ip": "1.1.1.1/32"},
                ]
            },
        },
        {
            "name": "ACC-SW01",
            "role": "access",
            "interfaces": {
                "interface": [
                    {"name": "GE0/0/1", "type": "ethernet", "mode": "trunk"},
                    {"name": "GE0/0/2", "type": "ethernet", "mode": "access"},
                ]
            },
        },
    ]
}

# XPath 风格的过滤查询
def xpath_filter(data, path, condition=None):
    """模拟 XPath 查询"""
    parts = path.strip("/").split("/")
    current = data

    for part in parts:
        if isinstance(current, dict) and part in current:
            current = current[part]
        elif isinstance(current, list):
            # 遍历列表
            results = []
            for item in current:
                if isinstance(item, dict):
                    item_result = item.get(part)
                    if item_result:
                        if isinstance(item_result, list):
                            results.extend(item_result)
                        else:
                            results.append(item_result)
            current = results
        else:
            return []

    # 应用条件过滤
    if condition and isinstance(current, list):
        return [item for item in current
                if all(item.get(k) == v for k, v in condition.items())]
    return current

# 查询示例
# 1. 查询所有设备
all_devices = xpath_filter(devices_data, "/device")
print("=== 所有设备 ===")
for d in all_devices:
    print(f"  {d['name']} ({d['role']})")

# 2. 查询核心设备
core_devices = xpath_filter(devices_data, "/device", {"role": "core"})
print("\n=== 核心设备 ===")
for d in core_devices:
    print(f"  {d['name']}")

# 3. 查询所有 type=ethernet 的接口
all_ifs = []
for dev in devices_data["device"]:
    for intf in dev["interfaces"]["interface"]:
        if intf.get("type") == "ethernet":
            all_ifs.append({**intf, "device": dev["name"]})

print("\n=== 所有以太网接口 ===")
for i in all_ifs:
    print(f"  {i['device']}: {i['name']}")

六、YANG 模型的实际应用

6.1 标准 IETF 接口模型示例

// RFC 8344: ietf-interfaces 简化版
module ietf-interfaces {
    yang-version 1.1;
    namespace "urn:ietf:params:xml:ns:yang:ietf-interfaces";
    prefix if;

    import ietf-yang-types {
        prefix yang;
    }

    container interfaces {
        list interface {
            key "name";

            leaf name {
                type string;
            }

            leaf description {
                type string;
            }

            leaf type {
                type identityref {
                    base interface-type;
                }
                mandatory true;
            }

            leaf enabled {
                type boolean;
                default true;
            }

            leaf link-up-down-trap-enable {
                type enumeration {
                    enum enabled;
                    enum disabled;
                }
            }

            container ipv4 {
                leaf enabled {
                    type boolean;
                    default true;
                }
                leaf mtu {
                    type uint16 {
                        range "68..1500";
                    }
                }
                list address {
                    key "ip";
                    leaf ip {
                        type yang:ipv4-address;
                    }
                    leaf prefix-length {
                        type uint8 {
                            range "0..32";
                        }
                        mandatory true;
                    }
                }
            }

            container statistics {
                config false;
                leaf discontinuity-time {
                    type yang:date-and-time;
                }
                leaf in-octets {
                    type yang:counter64;
                }
                leaf in-errors {
                    type yang:counter32;
                }
                leaf out-octets {
                    type yang:counter64;
                }
            }
        }
    }
}

6.2 基于 YANG 的配置操作

#!/usr/bin/env python3
# yang_config_ops.py — 基于 YANG 模型的配置操作

from copy import deepcopy

# 模拟设备数据库
device_db = {}

def yang_validate(data, schema):
    """YANG 数据校验"""
    errors = []

    if not isinstance(data, dict):
        errors.append("数据必须是字典")

    for field, rules in schema.items():
        value = data.get(field)

        if rules.get("mandatory") and value is None:
            errors.append(f"缺少必填字段: {field}")
            continue

        if value is not None:
            # 类型检查
            expected_type = rules["type"]
            if not isinstance(value, expected_type):
                errors.append(
                    f"{field}: 期望 {expected_type.__name__}, "
                    f"实际 {type(value).__name__}"
                )

            # 范围检查
            if "range" in rules and isinstance(value, (int, float)):
                min_v, max_v = rules["range"]
                if value < min_v or value > max_v:
                    errors.append(
                        f"{field}: 值 {value} 超出范围 [{min_v}, {max_v}]"
                    )

            # 模式检查
            if "pattern" in rules and isinstance(value, str):
                import re
                if not re.match(rules["pattern"], value):
                    errors.append(f"{field}: 格式不匹配 {rules['pattern']}")

    return errors

# 接口 YANG 模式的简化表示
INTERFACE_YANG_SCHEMA = {
    "name": {"type": str, "mandatory": True, "pattern": r"^[A-Za-z0-9/]+$"},
    "type": {"type": str, "mandatory": True},
    "enabled": {"type": bool, "mandatory": False},
    "mtu": {"type": int, "mandatory": False, "range": (68, 9216)},
    "description": {"type": str, "mandatory": False, "pattern": r"^[\w\-\s./]+$"},
    "ip_address": {"type": str, "mandatory": False},
}

# 测试数据
test_configs = [
    {"name": "GE0/0/1", "type": "ethernet", "enabled": True, "mtu": 1500},
    {"name": "Invalid/Name!", "type": "ethernet"},   # 名称格式错误
    {"name": "GE0/0/2", "type": "ethernet", "mtu": 99999},  # MTU 超范围
    {"name": "LoopBack0", "type": "loopback", "ip_address": "10.0.0.1"},
]

for config in test_configs:
    errors = yang_validate(config, INTERFACE_YANG_SCHEMA)
    if errors:
        print(f"✗ 校验失败 {config.get('name', '?')}:")
        for err in errors:
            print(f"  - {err}")
    else:
        print(f"✓ 校验通过: {config['name']}")
        device_db[config["name"]] = deepcopy(config)

七、YANG 最佳实践

7.1 模型设计原则

YANG 模型设计原则：

  1. 单一职责
  ┌─ 一个模块只负责一个功能域
  ├─ 接口模型只管接口
  ├─ 路由模型只管路由
  └─ 避免大而全的模块

  2. 扩展性
  ┌─ 使用 augment 扩展现有模型
  ├─ 不修改标准模型
  ├─ 厂商扩展加独立前缀
  └─ 保持向后兼容

  3. 命名规范
  ┌─ 模块名：ietf/huawei/example- + 功能名
  ├─ 容器名：名词复数
  ├─ 列表名：名词单数
  ├─ 叶子名：小写+连字符
  └─ 类型名：名词+描述

  4. 文档
  ┌─ 每个模块写 description
  ├─ 每个容器/list/leaf 写 description
  ├─ revision 记录变更历史
  └─ reference 引用标准文档

7.2 YANG 工具链

YANG 工具链：

  开发工具：
  ┌─ pyang：YANG 语法检查 + 格式转换
  ├─ yanglint：YANG 数据验证
  ├─ YANG Designer：可视化建模
  └─ VS Code YANG 插件

  代码生成：
  ┌─ pyangbind：YANG → Python 绑定的类
  ├─ yang2dsdl：YANG → XML Schema
  ├─ sysrepo：YANG → C 语言 API
  └─ yangc：YANG → C 语言结构体

  测试工具：
  ┌─ yangvalidator：在线验证
  ├─ Netopeer2：NETCONF 模拟器
  ├─ confd：YANG 基础服务器
  └─ yang-explorer：YANG 浏览器

八、总结

YANG 在网络自动化中的核心地位：

统一的配置数据模型 ┌─ 跨厂商接口抽象 ├─ 配置与状态统一建模 ├─ 支持 CRUD 操作 └─ 类型约束保证数据质量

自动化协议的基础 ┌─ NETCONF 操作基于 YANG 路径 ├─ RESTCONF URL 映射 YANG 节点 ├─ gNMI Path 引用 YANG 路径 └─ Telemetry 数据按 YANG 编码

YANG 驱动的自动化链路： | YANG 模型 定义 | → | 数据实例 (XML/JSON) 生成 | → | NETCONF/ RESTCONF 写入设备 | | --- | --- | --- | --- | --- | 标准/厂商 Python 脚本 API 交互

下篇预告：第311篇 — NETCONF/RESTCONF 协议，将介绍如何通过 NETCONF 和 RESTCONF 协议基于 YANG 模型对网络设备进行编程化配置管理。