第315篇：gNMI/gNOI：谷歌开源的网络管理协议

关键词：gNMI、gNOI、Google、gRPC、protobuf、Telemetry、网络管理、模型驱动

一、引言

第314篇介绍了 OpenConfig 和 IETF 标准模型。谷歌在大规模网络运营中发现，传统的 NETCONF（XML over SSH）和 RESTCONF（JSON over HTTP）在性能上无法满足超大规模数据中心的需求：XML 解析开销大、HTTP 轮询效率低、缺乏二进制编码支持。

因此，谷歌开发了 gNMI（gRPC Network Management Interface）和 gNOI（gRPC Network Operations Interface），这两项协议使用 gRPC + Protocol Buffers（protobuf）作为传输层，实现了高性能、低延迟的网络管理和监控。

二、核心概念

2.1 gNMI 架构设计

gNMI 是一个基于 gRPC 的双向流式协议，核心设计目标：

  1. 高效编码：使用 protobuf 二进制编码，比 XML/JSON 小 10-100 倍
  2. 流式传输：原生支持服务端推送（subscribe），无需客户端轮询
  3. 统一接口：Set（写）、Get（读）、Subscribe（订阅）三个 RPC 方法覆盖所有操作
  4. YANG 模型兼容：路径编码（PathEncoding）支持 YANG 节点寻址

gNMI 架构设计 - 图示

2.2 gNOI 架构设计

gNOI 是 gNMI 的补充，专注于网络设备的运维操作（operations），而非配置管理：

gNMI 操作 gNOI 操作
Get/Set 配置 系统操作（重启、诊断、证书管理）
Subscribe 状态 文件系统管理（读写文件）
网络诊断（ping、traceroute、ARP 刷新）
日志管理（获取日志）

gNOI 架构设计 - 图示

2.3 gNMI 与 NETCONF/RESTCONF 对比

维度 NETCONF RESTCONF gNMI
传输层 SSH (830) HTTP/HTTPS (443) gRPC/TLS (50051)
编码格式 XML JSON/XML Protobuf (二进制)
编码效率 低（冗长） 中 高（紧凑）
订阅模型 支持但实现复杂 扩展支持 原生流式推送
批量操作 单次 RPC 单次 HTTP 单次 gRPC 调用
双向通信 单向请求-响应 单向请求-响应 双向流
语言生态 有限 通用（HTTP） Go/Python/Java 等
性能 低（解析 XML） 中（解析 JSON） 高（二进制直解）
大规模场景 不适配 部分适配 原生设计

2.4 gNMI 路径编码（Path Encoding）

gNMI 使用路径（Path）来标识 YANG 数据树中的节点：

/YANG模块/YANG容器/YANG列表[键=值]/叶子节点

路径编码方式：

编码类型 说明 示例
PATH YANG 路径编码 /interfaces/interface[name=GigabitEthernet1/0/1]/config/description
ORIGIN 命名空间前缀 oc（OpenConfig）或 ietf
ELEMENT 路径元素列表 interfaces → interface → config → description 
# gNMI Path 示例（Python gnmi 库）
from gnmi.path import Path

# OpenConfig 接口路径
path = Path([
    ("interfaces", "interface"),
    {"name": "GigabitEthernet1/0/1"},
    ("config", "description"),
], origin="oc")

# IETF 接口路径
path = Path([
    ("interfaces", "interface"),
    {"name": "GigabitEthernet1/0/1"},
    "description",
], origin="ietf")

三、技术细节

3.1 gNMI 安装与基础连接

# 安装 Python gNMI 客户端
$ pip install pygnmi

# 或使用 grpcio + protobuf 手动构建
$ pip install grpcio grpcio-tools protobuf

3.1.1 Python pygnmi 基础操作

# scripts/gnmi_basic.py
import pygnmi
import json

# gNMI 连接配置
GNMI_CONFIG = {
    "target": ("10.1.1.1", 50051),
    "username": "admin",
    "password": "HuaWei@123",
    "override": "device-cert",  # 服务器证书名
    "insecure": True,  # 生产环境应使用 TLS 证书
}

# ========== Get: 读取配置 ==========
def gnmi_get(config_paths, state_paths=None, origin="oc"):
    """读取配置或状态数据"""
    with pygnmi.gnmi_get_request(**GNMI_CONFIG) as gnmi:
        response = gnmi.get(
            substitution={"host": "10.1.1.1"},
            type="all",  # all | config | state
            path=config_paths or [],
            state=state_paths or [],
            origin=origin,
        )
        return response

# 获取所有接口配置
interface_paths = [
    "/interfaces/interface/config/name",
    "/interfaces/interface/config/description",
    "/interfaces/interface/config/enabled",
    "/interfaces/interface/config/type",
]

response = gnmi_get(interface_paths, origin="oc")
print(json.dumps(response, indent=2, ensure_ascii=False))

# 获取特定接口（使用键值过滤）
specific_paths = [
    "/interfaces/interface[name=GigabitEthernet1/0/1]/config/description",
]
response = gnmi_get(specific_paths, origin="oc")

3.1.2 gNMI Set 操作（配置写入）

# scripts/gnmi_set.py
import pygnmi
import json

# ========== Set: 更新配置 ==========
def gnmi_set(update_paths, delete_paths=None, replace=False):
    """设置/更新配置"""
    with pygnmi.gnmi_set_request(**GNMI_CONFIG) as gnmi:
        response = gnmi.set(
            update=update_paths or [],
            delete=delete_paths or [],
            replace=replace,
        )
        return response

# 更新接口描述
update_paths = [
    {
        "path": "/interfaces/interface[name=GigabitEthernet1/0/1]/config/description",
        "val": "Modified-via-gNMI"
    },
    {
        "path": "/interfaces/interface[name=GigabitEthernet1/0/1]/config/enabled",
        "val": True
    },
]

response = gnmi_set(update_paths)
print("Set 操作结果:", json.dumps(response, indent=2))

# 删除配置（如删除 IP 地址）
delete_paths = [
    "/interfaces/interface[name=GigabitEthernet1/0/1]/subinterfaces/"
    "subinterface[index=10]/ipv4/addresses/address[ip=10.10.10.1]",
]
response = gnmi_set(delete_paths=delete_paths)

3.2 gNMI Subscribe（流式订阅）

gNMI 的 Subscribe 是其最强功能——服务端主动推送变更数据：

# scripts/gnmi_subscribe.py
import pygnmi
import json
import time

def gnmi_subscribe(subscription_list, mode="STREAM",
                   encoding="JSON_IETF", sample_mode="SAMPLE",
                   subscribe_mode="ALL", suffix=""):
    """订阅 YANG 数据变更"""
    with pygnmi.gnmi_subscribe_request(**GNMI_CONFIG) as gnmi:
        # 创建订阅
        gnmi.subscribe(
            subscription_list=subscription_list,
            mode=mode,        # STREAM | ONCE | POLL
            encoding=encoding,  # JSON_IETF | JSON | BYTES | PROTO
            sample_mode=sample_mode,  # SAMPLE | ON_CHANGE | TARGET_DEFINED
            subscribe_mode=subscribe_mode,  # ALL | OMIT | ONCE
        )

        # 持续接收推送
        print(f"开始订阅 (mode={mode}, encoding={encoding})...\n")
        try:
            for notification in gnmi.subscribe_response():
                timestamp = notification.get("timestamp", 0)
                time_str = time.strftime("%Y-%m-%d %H:%M:%S",
                                        time.localtime(timestamp / 10**9))

                for update in notification.get("update", []):
                    path = update.get("path", "")
                    val = update.get("val", "")
                    print(f"[{time_str}] {path} = {val}")

                for delete in notification.get("delete", []):
                    path = delete.get("path", "")
                    print(f"[{time_str}] DELETED: {path}")

        except KeyboardInterrupt:
            print("\n订阅已停止")

# ========== 订阅示例 ==========

# 示例1: 订阅接口状态变更（ON_CHANGE 模式）
interface_state_subs = [
    {
        "path": "/interfaces/interface/state/oper-status",
        "subscription_mode": "ON_CHANGE",
        "sample_mode": "SAMPLE",
    },
    {
        "path": "/interfaces/interface/state/counters/in-octets",
        "subscription_mode": "STREAM",
        "subscribes_once": False,
    },
]

gnmi_subscribe(interface_state_subs, mode="STREAM",
              sample_mode="SAMPLE")

# 示例2: 一次性获取当前值后订阅
gnmi_subscribe(
    [{"path": "/interfaces/interface/config/enabled"}],
    mode="ONCE",          # 只推送一次当前值
    encoding="JSON_IETF"
)

3.3 Subscribe 模式详解

模式 说明 适用场景
STREAM 持续推送，直到客户端取消订阅 实时监控、指标采集
ONCE 推送一次当前值后断开连接 快照采集、健康检查
POLL 客户端控制推送时机 按需查询
Sample Mode 说明 适用场景
SAMPLE 按固定间隔采样推送 性能指标（CPU、内存、流量）
ON_CHANGE 仅在数据变更时推送 配置变更、接口状态切换
TARGET_DEFINED 由设备决定采样策略 设备优化的推送间隔

Subscribe 模式详解 - 图示

3.4 gNOI 系统操作示例

# scripts/gnoi_operations.py
import grpc
import json
from gnoi.system import system_pb2, system_pb2_grpc

# ========== 建立 gNOI 连接 ==========
channel = grpc.insecure_channel("10.1.1.1:50051")
stub = system_pb2_grpc.SystemStub(channel)

# ========== 获取系统信息 ==========
def get_system_info(stub):
    """获取设备系统信息"""
    response = stub.Uptime(system_pb2.UptimeRequest())
    return {
        "uptime_seconds": response.uptime / 10**9,
        "boot_time": response.boot_time / 10**9,
    }

info = get_system_info(stub)
print(f"系统运行时间: {info['uptime_seconds']:.0f} 秒")

# ========== 执行 Ping 诊断 ==========
def execute_ping(stub, destination, count=4):
    """执行 Ping 诊断"""
    from gnoi.network import network_pb2, network_pb2_grpc

    net_stub = network_pb2_grpc.NetworkStub(stub.channel())
    response = net_stub.Ping(network_pb2.PingRequest(
        host=destination,
        count=count,
        size=64,
        df_bit=True,
    ))

    results = []
    for result in response:
        results.append({
            "ttl": result.ttl,
            "rtt_ms": result.rtt_ms,
            "success": result.success,
        })

    return results

# 执行 Ping
ping_results = execute_ping(stub, "10.2.2.1", count=4)
for r in ping_results:
    status = "✓" if r["success"] else "✗"
    print(f"{status} TTL={r['ttl']}, RTT={r['rtt_ms']:.1f}ms")

# ========== 刷新 ARP 缓存 ==========
def flush_arp(stub):
    """刷新 ARP 缓存"""
    response = net_stub.FlushArp(network_pb2.FlushArpRequest())
    return response

flush_result = flush_arp(stub)
print(f"ARP 缓存刷新结果: {flush_result}")

3.5 使用原生 grpcio 构建 gNMI 客户端

对于需要更精细控制的场景，可以直接使用 grpcio 构建客户端：

# scripts/gnmi_grpc_native.py
import grpc
import json
from google.protobuf import json_format
import gnmi.gnmi_pb2 as gnmi_pb2
import gnmi.gnmi_pb2_grpc as gnmi_grpc

# 建立 gRPC 连接
channel = grpc.insecure_channel("10.1.1.1:50051")
stub = gnmi_grpc.gNMIStub(channel)

# ========== Get 请求 ==========
def gnmi_get_native(stub, paths, dtype="ALL"):
    """原生 gRPC Get 请求"""
    path_list = []
    for p in paths:
        elements = gnmi_pb2.Path().ParseFromString(b"")
        parts = p.strip("/").split("/")
        for part in parts:
            if "=" in part:
                key_part, key_val = part.split("=", 1)
                elements.element.append(key_part)
                elements.key[key_val.split("=")[0]] = key_val.split("=")[1]
            else:
                elements.element.append(part)
        path_list.append(elements)

    request = gnmi_pb2.GetRequest(
        prefix=gnmi_pb2.Path(origin="oc"),
        path=path_list,
        type=dtype,
        encoding="JSON_IETF",
    )

    response = stub.Get(request)
    return response

# ========== Subscribe 流 ==========
def gnmi_subscribe_native(stub, subscriptions):
    """原生 gRPC Subscribe 流"""
    sub_list = []
    for sub in subscriptions:
        elements = gnmi_pb2.Path()
        parts = sub["path"].strip("/").split("/")
        for part in parts:
            elements.element.append(part)

        sub_list.append(gnmi_pb2.Subscription(
            path=elements,
            mode=gnmi_pb2.SubscriptionMode.ON_CHANGE,
        ))

    request = gnmi_pb2.SubscribeRequest()
    request.subscribe.Add(subscription=sub_list)

    # 接收流式响应
    for response in stub.Subscribe(iter([request])):
        if response.HasField("update"):
            for notification in response.update.notifications:
                for update in notification.updates:
                    val = json_format.MessageToDict(update.val)
                    print(f"Path: {update.path}")
                    print(f"Value: {val}")

四、Nornir 网络自动化框架集成

4.1 Nornir 项目结构与 gNMI Task

Nornir 是一个 Python 网络自动化框架，与 gNMI 结合可实现批量设备管理：

nornir-gnmi/
- inventory/
  - hosts.yaml           # 设备清单
  - groups.yaml          # 分组变量
  - defaults.yaml        # 默认变量
- groups/
  - core_switches.yaml   # 核心交换机变量
  - access_switches.yaml # 接入交换机变量
- playbooks/
  - gnmi_tasks/
    - __init__.py
    - collect_metrics.py  # gNMI指标采集
    - apply_config.py     # gNMI配置下发
    - monitor_changes.py  # gNMI变更监控
  - gather_inventory.yaml # 采集Inventory
  - deploy_config.yaml    # 部署配置
- nornir.yaml            # Nornir配置文件
- requirements.txt

4.2 Nornir Inventory 配置

# inventory/hosts.yaml
---
core-sw-01:
  hostname: 192.168.1.1
  groups:
    - core_switches
  data:
    gnmi_port: 50051
    site: beijing
    rack: A01

core-sw-02:
  hostname: 192.168.1.2
  groups:
    - core_switches
  data:
    gnmi_port: 50051
    site: beijing
    rack: A02

agg-sw-01:
  hostname: 192.168.1.10
  groups:
    - aggregation_switches
  data:
    gnmi_port: 50051
    site: beijing
    rack: B01

agg-sw-02:
  hostname: 192.168.1.11
  groups:
    - aggregation_switches
  data:
    gnmi_port: 50051
    site: beijing
    rack: B02

# inventory/groups.yaml
---
core_switches:
  groups: []
  vars:
    platform: huawei
    gnmi_username: admin
    gnmi_password: H@wa123!
    device_role: core
    metrics_interval: 30

aggregation_switches:
  groups: []
  vars:
    platform: huawei
    gnmi_username: admin
    gnmi_password: H@wa123!
    device_role: aggregation
    metrics_interval: 60

# nornir.yaml
---
inventory:
  plugin: SimpleInventory
  options:
    host_file: "inventory/hosts.yaml"
    group_file: "inventory/groups.yaml"
    defaults_file: "inventory/defaults.yaml"

runner:
  plugin: ThreadPool
  options:
    num_workers: 20

logging:
  level: INFO
  format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

4.3 Nornir gNMI Task 编写

# playbooks/gnmi_tasks/collect_metrics.py
from nornir import InitNornir
from nornir.core.task import Task, Result
from nornir_utils.plugins.functions import print_result
import pygnmi
import json

def gnmi_collect_metrics(task: Task, paths: list, **kwargs) -> Result:
    """Nornir Task: 使用gNMI采集设备指标"""
    try:
        gnmi_config = {
            "target": (task.host.hostname, task.host.data.get("gnmi_port", 50051)),
            "username": task.host.get("gnmi_username"),
            "password": task.host.get("gnmi_password"),
            "insecure": True,
        }

        subscription_list = [
            {"path": p, "subscription_mode": "STREAM"}
            for p in paths
        ]

        with pygnmi.gnmi_subscribe_request(**gnmi_config) as gnmi:
            gnmi.subscribe(
                subscription_list=subscription_list,
                mode="ONCE",
                encoding="JSON_IETF",
            )

            metrics = {}
            for notification in gnmi.subscribe_response():
                for update in notification.get("update", []):
                    metrics.update(update.get("val", {}))

            return Result(
                host=task.host,
                changed=True,
                metrics=metrics,
            )
    except Exception as e:
        return Result(
            host=task.host,
            changed=False,
            failed=True,
            message=f"gNMI采集失败: {str(e)}",
        )

def gnmi_apply_config(task: Task, config_path: str, config_data: dict, **kwargs) -> Result:
    """Nornir Task: 使用gNMI下发配置"""
    try:
        gnmi_config = {
            "target": (task.host.hostname, task.host.data.get("gnmi_port", 50051)),
            "username": task.host.get("gnmi_username"),
            "password": task.host.get("gnmi_password"),
            "insecure": True,
        }

        with pygnmi.gnmi_set_request(**gnmi_config) as gnmi:
            gnmi.set(
                update={config_path: config_data},
                encoding="JSON_IETF",
            )

        return Result(
            host=task.host,
            changed=True,
            message="配置下发成功",
        )
    except Exception as e:
        return Result(
            host=task.host,
            changed=False,
            failed=True,
            message=f"配置下发失败: {str(e)}",
        )

4.4 Nornir Filter 使用

# playbooks/gather_inventory.yaml
---
- name: 采集核心交换机指标
  inventory: core
  tasks:
    - name: 采集接口流量指标
      task: gnmi_collect_metrics
      vars:
        paths:
          - "/interfaces/interface/state/counters/in-octets"
          - "/interfaces/interface/state/counters/out-octets"
          - "/interfaces/interface/state/oper-status"

- name: 仅采集北京站点的设备
  inventory: core
  filter:
    data:
      site: beijing
  tasks:
    - name: 采集北京站点设备
      task: gnmi_collect_metrics
      vars:
        paths:
          - "/interfaces/interface/state/counters/in-octets"

- name: 采集所有核心交换机
  inventory: all
  filter:
    groups:
      - core_switches
  tasks:
    - name: 核心交换机全量指标
      task: gnmi_collect_metrics
      vars:
        paths:
          - "/interfaces/interface/state/counters/in-octets"
          - "/components/component/state/resource-utilization/cpu"

# Python Filter 示例
from nornir import InitNornir
from nornir.filter import F

nr = InitNornir(config_file="nornir.yaml")

# 筛选北京站点的核心交换机
beijing_core = nr.filter(F(data__site="beijing") & F(groups__name="core_switches"))

# 筛选特定机架的设备
rack_a = nr.filter(F(data__rack__startswith="A"))

# 排除特定主机
exclude_list = nr.filter(hostname__notcontains="sw-02")

# 组合筛选
target_devices = nr.filter(
    (F(data__site="beijing") | F(data__site="shanghai")) &
    F(groups__name="core_switches") &
    F(hostname__notcontains="old")
)

print(f"目标设备数量: {len(target_devices.inventory.hosts)}")
for host in target_devices.inventory.hosts.values():
    print(f"  - {host.name} ({host.hostname})")

4.5 Nornir 批量操作示例

# playbooks/batch_operations.py
from nornir import InitNornir
from nornir.core.filter import F
from nornir_utils.plugins.functions import print_result
from playbooks.gnmi_tasks import collect_metrics, apply_config

def batch_gather_metrics():
    """批量采集所有设备指标"""
    nr = InitNornir(config_file="nornir.yaml")

    # 定义采集路径
    interface_paths = [
        "/interfaces/interface/state/counters/in-octets",
        "/interfaces/interface/state/counters/out-octets",
        "/interfaces/interface/state/oper-status",
    ]

    system_paths = [
        "/components/component/state/resource-utilization/cpu",
        "/components/component/state/resource-utilization/memory",
    ]

    # 并行采集接口指标
    result = nr.run(
        name="采集接口流量指标",
        task=collect_metrics.gnmi_collect_metrics,
        paths=interface_paths,
    )
    print_result(result)

    # 并行采集系统资源指标
    result = nr.run(
        name="采集系统资源指标",
        task=collect_metrics.gnmi_collect_metrics,
        paths=system_paths,
    )
    print_result(result)

def batch_deploy_config():
    """批量下发配置"""
    nr = InitNornir(config_file="nornir.yaml")

    # 仅对核心交换机下发描述配置
    core = nr.filter(F(groups__name="core_switches"))

    config_data = {
        "description": "Managed by Nornir+gNMI",
    }

    result = core.run(
        name="批量更新接口描述",
        task=apply_config.gnmi_apply_config,
        config_path="/interfaces/interface/config/description",
        config_data=config_data,
    )
    print_result(result)

# 执行批量操作
# batch_gather_metrics()
# batch_deploy_config()

4.6 案例一：gNMI 大规模指标采集

场景：从 1000 台路由器采集接口流量、CPU、内存指标。

# scripts/gnmi_metrics_collector.py
import pygnmi
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

# 指标采集路径定义（OpenConfig）
METRICS_PATHS = [
    # 接口流量统计
    "/interfaces/interface/state/counters/in-octets",
    "/interfaces/interface/state/counters/out-octets",
    "/interfaces/interface/state/counters/in-unicast-pkts",
    "/interfaces/interface/state/counters/out-unicast-pkts",
    "/interfaces/interface/state/counters/in-errors",
    "/interfaces/interface/state/counters/out-errors",
    # 接口状态
    "/interfaces/interface/state/oper-status",
    # 系统资源
    "/components/component/state/resource-utilization/cpu",
    "/components/component/state/resource-utilization/memory",
]

def collect_metrics_once(device_config, metrics_paths):
    """使用 ONCE 模式采集一次当前指标"""
    subscription_list = [
        {"path": p, "subscription_mode": "STREAM"}
        for p in metrics_paths
    ]

    with pygnmi.gnmi_subscribe_request(**device_config) as gnmi:
        gnmi.subscribe(
            subscription_list=subscription_list,
            mode="ONCE",        # 只采集一次
            encoding="JSON_IETF",
            sample_mode="SAMPLE",
        )

        metrics = {}
        for notification in gnmi.subscribe_response():
            for update in notification.get("update", []):
                path = update.get("path", "")
                val = update.get("val")
                # 提取接口名和指标名
                metrics[path] = val

        return metrics

# 批量采集
devices = [
    {**GNMI_CONFIG, "target": (f"10.1.1.{i}", 50051)}
    for i in range(1, 11)
]

results = {}
with ThreadPoolExecutor(max_workers=20) as executor:
    futures = {
        executor.submit(collect_metrics_once, dev, METRICS_PATHS): dev
        for dev in devices
    }
    for future in as_completed(futures):
        dev = futures[future]
        host = dev["target"][0]
        try:
            metrics = future.result()
            results[host] = metrics
            print(f"✓ {host}: 采集 {len(metrics)} 个指标")
        except Exception as e:
            results[host] = {"error": str(e)}
            print(f"✗ {host}: {e}")

# 保存结果
with open("gnmi_metrics.json", "w", encoding="utf-8") as f:
    json.dump(results, f, indent=2, ensure_ascii=False, default=str)

4.2 案例二：gNMI 配置变更实时监控

# scripts/gnmi_change_monitor.py
import pygnmi
import json
from datetime import datetime

def monitor_config_changes(device_config, paths_to_watch):
    """实时监控配置变更"""
    subscription_list = [
        {
            "path": p,
            "subscription_mode": "ON_CHANGE",
        }
        for p in paths_to_watch
    ]

    with pygnmi.gnmi_subscribe_request(**device_config) as gnmi:
        gnmi.subscribe(
            subscription_list=subscription_list,
            mode="STREAM",
            encoding="JSON_IETF",
            sample_mode="SAMPLE",
        )

        print("=" * 60)
        print("配置变更监控启动")
        print(f"监控路径: {len(paths_to_watch)} 个")
        print("=" * 60)

        try:
            for notification in gnmi.subscribe_response():
                timestamp = notification.get("timestamp", 0)
                ts_str = datetime.fromtimestamp(timestamp / 10**9).strftime(
                    "%Y-%m-%d %H:%M:%S"
                )

                for update in notification.get("update", []):
                    path = update.get("path", "")
                    val = update.get("val")
                    print(f"\n[{ts_str}] 变更检测:")
                    print(f"  路径: {path}")
                    print(f"  值: {val}")
                    print(f"  ---")

                    # 触发告警逻辑
                    if "enabled" in path and val is False:
                        print(f"  ⚠️  告警: 接口被禁用!")

                for delete in notification.get("delete", []):
                    path = delete.get("path", "")
                    print(f"\n[{ts_str}] 删除检测:")
                    print(f"  路径: {path}")
                    print(f"  ⚠️  配置节点被删除!")

        except KeyboardInterrupt:
            print("\n监控已停止")

# 监控接口配置变更
watch_paths = [
    "/interfaces/interface/config/enabled",
    "/interfaces/interface/config/description",
    "/interfaces/interface/subinterfaces/subinterface/ipv4/addresses/address/config/ip",
    "/interfaces/interface/subinterfaces/subinterface/ipv4/addresses/address/config/prefix-length",
]

monitor_config_changes(GNMI_CONFIG, watch_paths)

4.3 案例三：故障案例——gNMI Subscribe 连接频繁断开

故障现象：gNMI Subscribe 流在运行几分钟后频繁断开，需要不断重连。

排查过程：

# scripts/gnmi_connection_diagnosis.py
import pygnmi
import time

def test_subscribe_stability(device_config, duration=300):
    """测试 Subscribe 连接的稳定性"""
    start_time = time.time()
    reconnect_count = 0
    notifications_received = 0

    while time.time() - start_time < duration:
        try:
            with pygnmi.gnmi_subscribe_request(**device_config) as gnmi:
                gnmi.subscribe(
                    subscription_list=[
                        {"path": "/interfaces/interface/state/oper-status",
                         "subscription_mode": "ON_CHANGE"}
                    ],
                    mode="STREAM",
                    encoding="JSON_IETF",
                    aliases=["interface-status"],
                    use_aliases=True,
                )

                for notification in gnmi.subscribe_response():
                    notifications_received += 1

        except Exception as e:
            reconnect_count += 1
            elapsed = time.time() - start_time
            print(f"[{elapsed:.0f}s] 连接断开 ({e}), "
                  f"已重连 {reconnect_count} 次, "
                  f"已接收 {notifications_received} 条通知")
            time.sleep(2)  # 等待后重连

    return {
        "duration": duration,
        "reconnect_count": reconnect_count,
        "notifications_received": notifications_received,
    }

result = test_subscribe_stability(GNMI_CONFIG, duration=600)
print(f"\n稳定性测试结果: {json.dumps(result, indent=2)}")

根因分析与解决方案：

根因 解决方案
gRPC Keepalive 超时 调整 keepalive_time_ms、keepalive_timeout_ms、keepalive_permit_without_calls
设备侧订阅限制 减少订阅路径数量，使用别名（alias）合并路径
网络中间设备断开空闲连接 配置 TCP keepalive，避免连接被防火墙超时断开
设备内存泄漏 联系设备厂商，使用 sample_mode=TARGET_DEFINED 减少推送频率
Subscribe 响应积压 客户端消费不够快，增加客户端处理并发

五、总结

协议 最佳场景 关键特性
gNMI 大规模网络监控、实时指标采集 gRPC + protobuf，高效编码，流式推送
gNOI 运维操作（重启、诊断、文件管理） 标准化运维接口，替代 vendor CLI
NETCONF 配置管理（中小规模） XML + SSH，成熟但效率低
RESTCONF API 集成、Web 前端 JSON + HTTP，易用但轮询效率低
最佳实践 说明
优先 protobuf 大规模场景使用 protobuf 编码，减少传输和解析开销
ON_CHANGE 监控 配置变更监控使用 ON_CHANGE 模式，避免无效推送
别名使用 大量路径订阅时使用 alias，降低设备侧处理负担
TLS 加密 生产环境必须使用 TLS，gRPC 原生支持 mTLS
重连机制 Subscribe 流需实现自动重连，处理网络中断
编码选择 调试时用 JSON_IETF 便于阅读，生产用 BYTES 或 PROTO 性能最优
Nornir集成 使用Nornir框架实现批量gNMI操作，Inventory管理设备清单
Filter筛选 按站点/角色/机架筛选目标设备，精准执行任务

思考题

  1. gNMI 的 Subscribe 流与 NETCONF 的 <subscribe> 在实现上有什么本质区别？
  2. 为什么 gNMI 选择 protobuf 而非 JSON 作为编码格式？在什么场景下 JSON_IETF 编码仍然有用？
  3. gNOI 和 gNMI 的职责边界是什么？为什么谷歌将它们设计为两个独立的 gRPC service？

下篇预告：第316篇：Telemetry 订阅与流式数据推送