第312篇：用 NETCONF 采集设备配置

关键词：NETCONF、配置采集、get-config、subtree filter、ncclient、批量采集、配置审计

一、引言

第311篇介绍了 NETCONF 协议的架构和操作模型。在实际运维中，NETCONF 最常见的应用场景之一就是采集设备配置——从网络设备的 running/candidate 数据区中读取结构化配置数据，用于备份、审计、对比分析等目的。

与传统的 SSH + display current-configuration 方式相比，NETCONF 采集具有天然优势：返回的是结构化的 XML 数据（而非纯文本），可以直接解析、查询、对比，无需编写复杂的正则表达式。

二、核心概念

2.1 配置采集的两种方式

2.2 Filter 过滤机制

NETCONF 支持两种 filter 类型，用于精确指定需要采集的 YANG 节点：

2.3 采集流程设计

三、技术细节

3.1 subtree filter 精确采集

subtree filter 是最常用的过滤方式，通过指定 YANG 节点路径来精确控制采集范围：

# scripts/netconf_collect_subtree.py
from ncclient import manager
import xml.etree.ElementTree as ET
import json

DEVICE = {
    "host": "10.1.1.1",
    "port": 830,
    "username": "admin",
    "password": "HuaWei@123",
    "device_params": {"name": "huawei"},
    "hostkey_verify": False,
    "timeout": 60,
}

def xml_to_dict(elem):
    """将 XML 元素递归转换为 Python 字典"""
    result = {}
    for child in elem:
        tag = child.tag.split("}")[-1]  # 去除命名空间前缀
        if len(child) > 0:
            # 有子节点
            if tag in result:
                # 处理 list（同名节点）
                if not isinstance(result[tag], list):
                    result[tag] = [result[tag]]
                result[tag].append(xml_to_dict(child))
            else:
                result[tag] = xml_to_dict(child)
        else:
            # 叶子节点
            value = child.text.strip() if child.text else ""
            # 类型转换
            if value.lower() == "true":
                value = True
            elif value.lower() == "false":
                value = False
            elif value.isdigit():
                value = int(value)
            if tag in result:
                if not isinstance(result[tag], list):
                    result[tag] = [result[tag]]
                result[tag].append(value)
            else:
                result[tag] = value
    return result

# ========== 采集接口配置 ==========
with manager.connect(**DEVICE) as m:

    # 方式一：采集指定接口的配置
    filter_single = """
    <filter type="subtree">
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface>
                <name>GigabitEthernet1/0/1</name>
            </interface>
        </interfaces>
    </filter>
    """
    reply = m.get_config(source="running", filter=filter_single)
    print("=== 单个接口配置 ===")
    config_dict = xml_to_dict(reply.data_element)
    print(json.dumps(config_dict, indent=2, ensure_ascii=False))

    # 方式二：采集所有接口
    filter_all = """
    <filter type="subtree">
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>
    </filter>
    """
    reply = m.get_config(source="running", filter=filter_all)
    print("\n=== 全部接口配置 ===")
    all_interfaces = xml_to_dict(reply.data_element)
    print(json.dumps(all_interfaces, indent=2, ensure_ascii=False))

    # 方式三：采集 VLAN 配置
    filter_vlan = """
    <filter type="subtree">
        <vlans xmlns="urn:huawei:params:xml:ns:yang:huawei-vlan"/>
    </filter>
    """
    reply = m.get_config(source="running", filter=filter_vlan)
    print("\n=== VLAN 配置 ===")
    vlan_config = xml_to_dict(reply.data_element)
    print(json.dumps(vlan_config, indent=2, ensure_ascii=False))

3.2 xpath filter 条件查询

当需要基于条件筛选数据时（如"获取所有状态为 up 的接口"），xpath filter 更加灵活：

# scripts/netconf_collect_xpath.py
from ncclient import manager

with manager.connect(**DEVICE) as m:

    # 检查是否支持 xpath capability
    xpath_cap = "urn:ietf:params:netconf:capability:xpath:1.0" in m.server_capabilities
    print(f"设备支持 xpath: {xpath_cap}")

    if xpath_cap:
        # 注册命名空间
        m.session.push_default_nsmap({
            "if": "urn:ietf:params:xml:ns:yang:ietf-interfaces",
            "hw": "urn:huawei:params:xml:ns:yang:huawei-system"
        })

        # 查询所有 enabled=true 的接口
        filter_xpath = """
        <filter type="xpath"
                select="if:interfaces/if:interface[if:enabled='true']"/>
        """
        try:
            reply = m.get_config(source="running", filter=filter_xpath)
            print("启用状态的接口:")
            # 解析结果...
        except Exception as e:
            print(f"XPath 查询失败: {e}")

        # 查询 VLAN ID 大于 100 的 VLAN
        filter_vlan_xpath = """
        <filter type="xpath"
                select="hw:vlans/hw:vlan[hw:vlan-id>100]"/>
        """
        try:
            reply = m.get_config(source="running", filter=filter_vlan_xpath)
            print("VLAN ID > 100 的配置:")
        except Exception as e:
            print(f"XPath 查询 VLAN 失败: {e}")

3.3 全量配置采集与运行状态采集

# scripts/netconf_full_collect.py
from ncclient import manager
import xml.etree.ElementTree as ET
from datetime import datetime
import hashlib
import json

def collect_full_config(m, source="running"):
    """采集完整配置（无 filter）"""
    reply = m.get_config(source=source, filter=None)
    return ET.tostring(reply.data_element, pretty_print=True).decode()

def collect_running_state(m, filter_xml):
    """采集运行状态数据（包含 counters、status 等）"""
    reply = m.get(filter=filter_xml)
    return ET.tostring(reply.data_element, pretty_print=True).decode()

def compute_config_hash(config_xml):
    """计算配置哈希值，用于变更检测"""
    return hashlib.sha256(config_xml.encode()).hexdigest()[:16]

with manager.connect(**DEVICE) as m:

    # ========== 全量配置采集 ==========
    full_config = collect_full_config(m, source="running")
    config_hash = compute_config_hash(full_config)

    print(f"配置哈希: {config_hash}")
    print(f"配置大小: {len(full_config)} bytes")

    # 保存配置快照
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    with open(f"configs/config_{timestamp}_{config_hash}.xml", "w", encoding="utf-8") as f:
        f.write(full_config)

    # ========== 接口状态采集 ==========
    state_filter = """
    <filter type="subtree">
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface>
                <name>GigabitEthernet1/0/1</name>
                <oper-status/>
                <statistics/>
            </interface>
        </interfaces>
    </filter>
    """
    state_data = collect_running_state(m, state_filter)
    print(f"\n接口状态数据:\n{state_data}")

    # ========== 路由表采集 ==========
    route_filter = """
    <filter type="subtree">
        <routing xmlns="urn:ietf:params:xml:ns:yang:ietf-routing">
            <route-destination-entries/>
        </routing>
    </filter>
    """
    route_data = collect_running_state(m, route_filter)
    print(f"路由表大小: {len(route_data)} bytes")

3.4 结构化解析：XML → JSON 转换工具

将 NETCONF 返回的 XML 配置转换为 JSON，便于后续处理：

# scripts/netconf_xml_to_json.py
from ncclient import manager
import xmltodict
import json

def netconf_to_json(m, filter_xml, source="running"):
    """将 NETCONF 返回的 XML 转换为 JSON"""
    reply = m.get_config(source=source, filter=filter_xml)
    xml_str = ET.tostring(reply.data_element, pretty_print=False).decode()

    # 使用 xmltodict 转换
    data = xmltodict.parse(xml_str)

    # 去除顶层 rpc-reply 包装
    payload = data.get("rpc-reply", {}).get("data", data)

    return payload

# 安装依赖: pip install xmltodict
with manager.connect(**DEVICE) as m:
    filter_xml = """
    <filter type="subtree">
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>
    </filter>
    """

    config_json = netconf_to_json(m, filter_xml)

    # 处理接口列表（xmltodict 对单个元素不返回列表）
    interfaces = config_json.get("interfaces", {}).get("interface", [])
    if isinstance(interfaces, dict):
        interfaces = [interfaces]

    for iface in interfaces:
        name = iface.get("name", "?")
        enabled = iface.get("enabled", "?")
        desc = iface.get("description", "")
        print(f"接口: {name}, 状态: {enabled}, 描述: {desc}")

四、实战配置/案例分析

4.1 案例一：多设备配置批量采集与存储

场景：从 200 台设备采集接口 + VLAN + 路由配置，统一存入 JSON 文件。

# scripts/batch_config_collector.py
from ncclient import manager
from ncclient.operations import RPCError
from concurrent.futures import ThreadPoolExecutor, as_completed
import xml.etree.ElementTree as ET
import xmltodict
import yaml
import json
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# 采集配置模板：定义需要采集的 YANG 节点
COLLECTION_TEMPLATES = {
    "interfaces": {
        "filter": """
        <filter type="subtree">
            <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>
        </filter>
        """,
        "source": "running",
    },
    "vlans": {
        "filter": """
        <filter type="subtree">
            <vlans xmlns="urn:huawei:params:xml:ns:yang:huawei-vlan"/>
        </filter>
        """,
        "source": "running",
    },
    "bgp": {
        "filter": """
        <filter type="subtree">
            <bgp xmlns="urn:ietf:params:xml:ns:yang:ietf-bgp"/>
        </filter>
        """,
        "source": "running",
    },
    "interface_state": {
        "filter": """
        <filter type="subtree">
            <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
                <interface>
                    <oper-status/>
                    <statistics/>
                </interface>
            </interfaces>
        </filter>
        """,
        "source": None,  # None 表示用 get 而非 get-config
    },
}

def collect_device_config(device, templates):
    """采集单台设备的多项配置"""
    results = {
        "host": device["host"],
        "hostname": None,
        "timestamp": datetime.now().isoformat(),
        "collections": {}
    }

    try:
        with manager.connect(
            host=device["host"],
            port=device.get("port", 830),
            username=device["username"],
            password=device["password"],
            hostkey_verify=False,
            device_params={"name": device.get("device_type", "huawei")},
            timeout=60
        ) as m:
            # 获取设备主机名
            hostname_reply = m.get_config(
                source="running",
                filter=('subtree',
                    '<system xmlns="urn:huawei:params:xml:ns:yang:huawei-system">'
                    '<hostname/>'
                    '</system>')
            )
            root = ET.fromstring(ET.tostring(hostname_reply.data_element))
            hostname_elem = root.find(".//{*}hostname")
            results["hostname"] = hostname_elem.text if hostname_elem is not None else device["host"]

            # 按模板逐项采集
            for key, template in templates.items():
                try:
                    if template["source"]:
                        reply = m.get_config(
                            source=template["source"],
                            filter=template["filter"]
                        )
                    else:
                        reply = m.get(filter=template["filter"])

                    xml_str = ET.tostring(reply.data_element, pretty_print=False).decode()
                    parsed = xmltodict.parse(xml_str)
                    results["collections"][key] = parsed.get("rpc-reply", {}).get("data", parsed)
                    results["collections"][key]["_status"] = "success"

                except RPCError as e:
                    results["collections"][key] = {
                        "_status": "rpc_error",
                        "_error": str(e.message)
                    }
                except Exception as e:
                    results["collections"][key] = {
                        "_status": "error",
                        "_error": str(e)
                    }

            results["_overall_status"] = "success"

    except Exception as e:
        results["_overall_status"] = "error"
        results["_error"] = str(e)
        logger.error(f"{device['host']} 采集失败: {e}")

    return results

def run_batch_collection(inventory_path, templates, output_dir="collected_configs", max_workers=10):
    """批量采集入口"""
    with open(inventory_path, "r", encoding="utf-8") as f:
        inventory = yaml.safe_load(f)

    devices = inventory.get("devices", [])
    logger.info(f"开始采集 {len(devices)} 台设备...")

    import os
    os.makedirs(output_dir, exist_ok=True)

    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(collect_device_config, dev, templates): dev
            for dev in devices
        }
        for future in as_completed(futures):
            result = future.result()
            results.append(result)

            # 保存每台设备的配置
            safe_name = result["hostname"] or result["host"]
            safe_name = safe_name.replace("/", "_").replace("\\", "_")
            filename = f"{output_dir}/{safe_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"

            with open(filename, "w", encoding="utf-8") as f:
                json.dump(result, f, indent=2, ensure_ascii=False, default=str)

            status = "✓" if result["_overall_status"] == "success" else "✗"
            logger.info(f"{status} {result['host']}: {result['_overall_status']}")

    # 生成汇总报告
    success_count = sum(1 for r in results if r["_overall_status"] == "success")
    report = {
        "timestamp": datetime.now().isoformat(),
        "total_devices": len(devices),
        "success_count": success_count,
        "failure_count": len(devices) - success_count,
        "devices": [
            {"host": r["host"], "status": r["_overall_status"],
             "hostname": r["hostname"]}
            for r in results
        ]
    }

    with open(f"{output_dir}/collection_report.json", "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    logger.info(f"采集完成: {success_count}/{len(devices)} 台成功")
    return results

# 执行
run_batch_collection("devices/inventory.yaml", COLLECTION_TEMPLATES)

4.2 案例二：配置变更检测（Diff）

场景：对比两次采集的配置快照，识别变更内容。

# scripts/config_diff.py
from ncclient import manager
import xmltodict
import json
import difflib

def flatten_dict(d, parent_key="", sep="."):
    """将嵌套字典展平为键值对"""
    items = []
    for k, v in d.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        # 跳过内部字段
        if k.startswith("_"):
            continue
        if isinstance(v, dict):
            items.extend(flatten_dict(v, new_key, sep).items())
        elif isinstance(v, list):
            for i, item in enumerate(v):
                if isinstance(item, dict):
                    items.extend(flatten_dict(item, f"{new_key}[{i}]", sep).items())
                else:
                    items.append((f"{new_key}[{i}]", str(item)))
        else:
            items.append((new_key, str(v)))
    return dict(items)

def compare_configs(old_config, new_config, label="config"):
    """比较两个配置快照，返回变更摘要"""
    old_flat = flatten_dict(old_config)
    new_flat = flatten_dict(new_config)

    old_keys = set(old_flat.keys())
    new_keys = set(new_flat.keys())

    added = new_keys - old_keys
    removed = old_keys - new_keys
    modified = set()

    for key in old_keys & new_keys:
        if old_flat[key] != new_flat[key]:
            modified.add(key)

    changes = {
        "label": label,
        "added_count": len(added),
        "removed_count": len(removed),
        "modified_count": len(modified),
        "total_changes": len(added) + len(removed) + len(modified),
        "details": {
            "added": [{"key": k, "value": new_flat[k]} for k in sorted(added)],
            "removed": [{"key": k, "value": old_flat[k]} for k in sorted(removed)],
            "modified": [
                {"key": k, "old_value": old_flat[k], "new_value": new_flat[k]}
                for k in sorted(modified)
            ]
        }
    }
    return changes

# 使用示例：对比两次采集
old_snapshot = json.load(open("collected_configs/SW-Core-01_20260901_interfaces.json"))
new_snapshot = json.load(open("collected_configs/SW-Core-01_20260906_interfaces.json"))

changes = compare_configs(
    old_snapshot.get("collections", {}).get("interfaces", {}),
    new_snapshot.get("collections", {}).get("interfaces", {}),
    label="SW-Core-01 Interfaces"
)

print(f"变更摘要: {changes['total_changes']} 处变更")
print(f"  新增: {changes['added_count']}")
print(f"  删除: {changes['removed_count']}")
print(f"  修改: {changes['modified_count']}")

if changes["details"]["modified"]:
    print("\n修改详情:")
    for change in changes["details"]["modified"][:10]:  # 只显示前10条
        print(f"  {change['key']}: '{change['old_value']}' → '{change['new_value']}'")

4.3 案例三：故障排查——配置采集超时诊断

故障场景：某台设备 NETCONF 采集经常超时，需要定位原因。

# scripts/netconf_diagnosis.py
from ncclient import manager
import time
import logging

logging.basicConfig(level=logging.DEBUG,
                    format="%(asctime)s [%(levelname)s] %(message)s")

DEVICE_PROBLEMATIC = {
    "host": "10.1.1.50",  # 经常超时的设备
    "port": 830,
    "username": "admin",
    "password": "HuaWei@123",
    "device_params": {"name": "huawei"},
    "hostkey_verify": False,
    "timeout": 300,  # 加大超时时间
}

def diagnose_netconf_session(m):
    """诊断 NETCONF 会话各项操作的性能"""
    diagnostics = {}

    # 测试 1: Hello 协商时间
    start = time.time()
    caps = m.server_capabilities
    diagnostics["hello_time"] = time.time() - start
    diagnostics["capabilities_count"] = len(caps)
    print(f"Hello 协商耗时: {diagnostics['hello_time']:.2f}s")

    # 测试 2: 小 filter 查询（接口配置）
    small_filter = """
    <filter type="subtree">
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
            <interface>
                <name>GigabitEthernet1/0/1</name>
            </interface>
        </interfaces>
    </filter>
    """
    start = time.time()
    reply = m.get_config(source="running", filter=small_filter)
    diagnostics["small_query_time"] = time.time() - start
    diagnostics["small_query_size"] = len(str(reply))
    print(f"小 filter 查询耗时: {diagnostics['small_query_time']:.2f}s, 数据量: {diagnostics['small_query_size']} bytes")

    # 测试 3: 大 filter 查询（全量接口）
    large_filter = """
    <filter type="subtree">
        <interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>
    </filter>
    """
    start = time.time()
    reply = m.get_config(source="running", filter=large_filter)
    diagnostics["large_query_time"] = time.time() - start
    diagnostics["large_query_size"] = len(str(reply))
    print(f"大 filter 查询耗时: {diagnostics['large_query_time']:.2f}s, 数据量: {diagnostics['large_query_size']} bytes")

    # 测试 4: 无 filter 全量查询
    start = time.time()
    try:
        reply = m.get_config(source="running", filter=None)
        diagnostics["full_query_time"] = time.time() - start
        diagnostics["full_query_size"] = len(str(reply))
        print(f"全量查询耗时: {diagnostics['full_query_time']:.2f}s, 数据量: {diagnostics['full_query_size']} bytes")
    except Exception as e:
        diagnostics["full_query_time"] = "timeout"
        diagnostics["full_query_error"] = str(e)
        print(f"全量查询超时: {e}")

    return diagnostics

with manager.connect(**DEVICE_PROBLEMATIC) as m:
    diag = diagnose_netconf_session(m)
    print(f"\n诊断结果: {json.dumps(diag, indent=2)}")

故障根因分析：

四（续）、高级主题

4.4 NETCONF 大容量配置分页采集

当配置数据量超过设备缓冲区时，需要使用分页或流式采集：

# scripts/netconf_paginated_collect.py
from ncclient import manager
from ncclient.transport import SessionListenThread
import xml.etree.ElementTree as ET
import time

def collect_with_writable_runtime(m, max_bytes=5*1024*1024):
    """
    动态调整 NETCONF max-message-size 应对大配置。
    华为设备默认 max-message-size=10MB，部分场景需要调大。
    """
    # 检查当前 max-message-size
    caps = str(m.server_capabilities)
    print(f"设备 Capabilities:\n{caps}")

    # 分模块采集（避免一次性拉取全量配置）
    modules = [
        ("interfaces",
         '<filter type="subtree">'
         '<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>'
         '</filter>'),
        ("vlans",
         '<filter type="subtree">'
         '<vlans xmlns="urn:huawei:params:xml:ns:yang:huawei-vlan"/>'
         '</filter>'),
        ("bgp",
         '<filter type="subtree">'
         '<bgp xmlns="urn:ietf:params:xml:ns:yang:ietf-bgp"/>'
         '</filter>'),
        ("ospf",
         '<filter type="subtree">'
         '<ospf xmlns="urn:ietf:params:xml:ns:yang:ietf-ospf"/>'
         '</filter>'),
        ("isis",
         '<filter type="subtree">'
         '<isis xmlns="urn:ietf:params:xml:ns:yang:ietf-isis"/>'
         '</filter>'),
    ]

    results = {}
    for module_name, filter_xml in modules:
        try:
            start = time.time()
            reply = m.get_config(source="running", filter=filter_xml)
            elapsed = time.time() - start
            data_size = len(str(reply))

            results[module_name] = {
                "status": "success",
                "size_bytes": data_size,
                "elapsed_seconds": round(elapsed, 2),
                "data": ET.tostring(reply.data_element, pretty_print=False).decode()
            }
            print(f"  ✓ {module_name:15s} {data_size:>10,} bytes  ({elapsed:.2f}s)")

        except Exception as e:
            results[module_name] = {
                "status": "error",
                "error": str(e)
            }
            print(f"  ✗ {module_name:15s} 错误: {e}")

    return results

# 使用示例
with manager.connect(
    host="10.1.1.1", port=830,
    username="admin", password="HuaWei@123",
    hostkey_verify=False,
    device_params={"name": "huawei"},
    timeout=300
) as m:
    results = collect_with_writable_runtime(m)

    # 统计
    total_size = sum(r.get("size_bytes", 0) for r in results.values())
    print(f"\n总配置大小: {total_size:,} bytes ({total_size/1024/1024:.2f} MB)")

4.5 NETCONF 采集结果存储策略

# scripts/netconf_config_store.py
import json
import hashlib
import gzip
from pathlib import Path
from datetime import datetime

class ConfigStore:
    """配置快照存储管理器"""

    def __init__(self, base_dir="config_store"):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)

    def store_config(self, hostname, config_xml, metadata=None):
        """存储配置快照（压缩 + 去重）"""
        config_hash = hashlib.sha256(config_xml.encode()).hexdigest()
        short_hash = config_hash[:12]
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

        # 检查是否已有相同配置（去重）
        latest_file = self._get_latest_file(hostname)
        if latest_file and self._read_hash(latest_file) == config_hash:
            print(f"  → {hostname}: 配置无变更，跳过存储")
            return None

        # 存储路径: config_store/{hostname}/{timestamp}_{hash}.xml.gz
        device_dir = self.base_dir / hostname
        device_dir.mkdir(exist_ok=True)

        filename = f"{timestamp}_{short_hash}.xml.gz"
        filepath = device_dir / filename

        # 压缩存储
        with gzip.open(filepath, "wt", encoding="utf-8") as f:
            f.write(config_xml)

        # 元数据
        meta = {
            "hostname": hostname,
            "timestamp": timestamp,
            "sha256": config_hash,
            "size_bytes": len(config_xml),
            "compressed_size": filepath.stat().st_size,
            "metadata": metadata or {}
        }
        meta_file = filepath.with_suffix(".meta.json")
        with open(meta_file, "w", encoding="utf-8") as f:
            json.dump(meta, f, indent=2)

        print(f"  ✓ {hostname}: 存储 {filename} ({len(config_xml):,} → {meta['compressed_size']:,} bytes)")
        return filepath

    def get_latest_config(self, hostname):
        """获取最新配置"""
        latest = self._get_latest_file(hostname)
        if latest:
            with gzip.open(latest, "rt", encoding="utf-8") as f:
                return f.read()
        return None

    def list_versions(self, hostname):
        """列出所有版本"""
        device_dir = self.base_dir / hostname
        if not device_dir.exists():
            return []
        files = sorted(device_dir.glob("*.xml.gz"), reverse=True)
        return [f.stem for f in files]

    def _get_latest_file(self, hostname):
        device_dir = self.base_dir / hostname
        if not device_dir.exists():
            return None
        files = list(device_dir.glob("*.xml.gz"))
        if not files:
            return None
        return max(files, key=lambda f: f.stat().st_mtime)

    def _read_hash(self, filepath):
        meta_file = filepath.with_suffix(".meta.json")
        if meta_file.exists():
            with open(meta_file, "r") as f:
                return json.load(f).get("sha256", "")
        return ""

    def generate_inventory_report(self):
        """生成存储库存报告"""
        report = []
        for device_dir in self.base_dir.iterdir():
            if device_dir.is_dir():
                files = list(device_dir.glob("*.xml.gz"))
                latest = max(files, key=lambda f: f.stat().st_mtime) if files else None
                report.append({
                    "hostname": device_dir.name,
                    "versions": len(files),
                    "latest_file": latest.stem if latest else None,
                    "latest_time": datetime.fromtimestamp(
                        latest.stat().st_mtime
                    ).isoformat() if latest else None
                })
        return report

4.6 NETCONF 采集性能基准测试

# scripts/netconf_bench.py
from ncclient import manager
import time
import statistics

def benchmark_netconf(device, iterations=5):
    """NETCONF 采集性能基准测试"""
    tests = {
        "small_filter": (
            '<filter type="subtree">'
            '<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">'
            '<interface><name>GE1/0/1</name></interface>'
            '</interfaces></filter>',
            "单接口配置"
        ),
        "medium_filter": (
            '<filter type="subtree">'
            '<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>'
            '</filter>',
            "全部接口配置"
        ),
        "large_filter": (
            '<filter type="subtree">'
            '<bgp xmlns="urn:ietf:params:xml:ns:yang:ietf-bgp"/>'
            '</filter>',
            "BGP 全量配置"
        ),
    }

    results = {}
    with manager.connect(
        host=device["host"], port=830,
        username=device["username"], password=device["password"],
        hostkey_verify=False,
        device_params={"name": device.get("vendor", "huawei")},
        timeout=120
    ) as m:
        for test_name, (filter_xml, description) in tests.items():
            times = []
            sizes = []

            for i in range(iterations):
                # Warm-up（第一次不计入统计）
                if i == 0:
                    m.get_config(source="running", filter=filter_xml)
                    continue

                start = time.perf_counter()
                reply = m.get_config(source="running", filter=filter_xml)
                elapsed = time.perf_counter() - start

                times.append(elapsed)
                sizes.append(len(str(reply)))

            results[description] = {
                "mean_time_ms": round(statistics.mean(times) * 1000, 2),
                "min_time_ms": round(min(times) * 1000, 2),
                "max_time_ms": round(max(times) * 1000, 2),
                "mean_size_bytes": statistics.mean(sizes),
                "iterations": iterations - 1,
            }

    # 输出结果
    print(f"\n{'='*70}")
    print(f"NETCONF 性能基准测试: {device['host']}")
    print(f"{'='*70}")
    for desc, stats in results.items():
        print(f"\n{desc}:")
        print(f"  平均耗时: {stats['mean_time_ms']:>8.2f} ms")
        print(f"  最小/最大: {stats['min_time_ms']:>8.2f} / {stats['max_time_ms']:>8.2f} ms")
        print(f"  平均数据量: {stats['mean_size_bytes']:>10,.0f} bytes")

    return results

五、总结

思考题

1. get-config 和 get 在采集结果上有什么区别？什么场景下必须用 get 而非 get-config？
2. subtree filter 中如果指定了一个不存在的 YANG 节点路径，NETCONF 会返回什么？
3. 批量采集 200 台设备时，串行采集和并行采集各有什么优缺点？如何选择合适的并发数？

下篇预告：第313篇：用 RESTCONF 修改接口配置