第321篇：全网拓扑自动发现与绘图

关键词：拓扑发现、LLDP、CDP、SNMP、Graphviz、NetworkX、自动绘图、依赖第320篇IPAM自动化

一、引言

在网络规模从几十台设备膨胀到上千台后，手工绘制拓扑图不仅效率极低，而且极难保持准确性。第320篇我们介绍了 IPAM 自动化的核心思路：将 IP 地址资源纳入程序化管控。而拓扑自动发现正是 IPAM 的自然延伸——当你知道哪些 IP 属于哪些设备后，下一步就是搞清楚这些设备之间是如何连接的。

全网拓扑自动发现的核心逻辑非常简单：利用设备自身产生的协议报文（LLDP/CDP/ARP/CDP）和可采集的管理数据（SNMP Net-SNMP MIB、接口描述），自动构建出设备间的连接关系，再用绘图引擎渲染成可视化拓扑图。这一流程贯穿"数据采集 → 关系推理 → 图结构存储 → 可视化渲染"四个阶段。

二、拓扑发现的数据源

2.1 LLDP 与 CDP 邻居发现

LLDP（Link Layer Discovery Protocol，IEEE 802.1AB）是业界标准链路层发现协议，CDP（Cisco Discovery Protocol）是 Cisco 私有协议。两者原理相同：每隔一定周期向直连对端发送包含本地接口名、系统名称、管理 IP 等信息的 TLV（Type-Length-Value）报文。

# 从华为设备提取 LLDP 邻居信息
import paramiko

def get_lldp_neighbors(hostname, username, password):
    """获取设备 LLDP 邻居表"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname, username=username, password=password)

    # 执行 display lldp neighbor
    stdin, stdout, stderr = ssh.exec_command('display lldp neighbor')
    output = stdout.read().decode()
    ssh.close()

    # 解析输出
    neighbors = parse_lldp_table(output)
    return neighbors

def parse_lldp_table(raw_output):
    """解析 LLDP neighbor 表格"""
    neighbors = []
    lines = raw_output.strip().split('\n')
    for line in lines:
        parts = line.split()
        if len(parts) >= 4:
            neighbors.append({
                'local_interface': parts[1],
                'remote_device': parts[2],
                'remote_interface': parts[3],
                'status': parts[4] if len(parts) > 4 else 'up'
            })
    return neighbors

2.2 SNMP 采集接口与 ARP 表

SNMP 可采集的 MIB 节点为拓扑发现提供补充信息：

# 使用 pysnmp 采集 LLDP 数据
from pysnmp.hlapi import *

def get_lldp_via_snmp(hostname, community='public'):
    """通过 SNMP 采集 LLDP 邻居信息"""
    neighbors = []

    # 轮询 lldpRemTable
    for (err_status, err_index, err_indication, snmp_data) in next(
        getCmd(SnmpEngine(),
               CommunityData(community),
               UdpTransportTarget((hostname, 161)),
               ContextData(),
               ObjectType(ObjectIdentity('1.0.8802.1.1.2.1.4.1.1.4')),   # lldpRemLocalPortNum
               ObjectIdentity('1.0.8802.1.1.2.1.4.1.1.9'),               # lldpRemChassisId
               ObjectIdentity('1.0.8802.1.1.2.1.4.1.1.11'))):            # lldpRemPortId

        if not err_status:
            for oid, val in snmp_data:
                pass  # 解析为邻居关系
    return neighbors

2.3 ARP 表与 MAC 地址表推导

当 LLDP/CDP 不可用时，可通过 ARP 表 + MAC 地址表交叉关联推导连接关系：

1. 在设备 A 的 ARP 表中查到某 IP 对应的 MAC 地址
2. 在设备 A 的 MAC 地址表中查到该 MAC 对应的入接口
3. 若该 MAC 属于设备 B，则推断 A 与该接口相连的设备为 B

此方法的精度低于 LLDP，但在纯二层交换场景下仍具有实用价值。

三、图结构建模与存储

3.1 NetworkX 图模型

Python 的 NetworkX 库提供了完善的图论数据结构，非常适合存储和处理拓扑关系：

import networkx as nx

# 创建无向图表示网络拓扑
G = nx.Graph()

# 添加设备节点
G.add_node('PE-BJ-01', role='PE', vendor='Huawei', ip='10.1.1.1')
G.add_node('PE-BJ-02', role='PE', vendor='Huawei', ip='10.1.1.2')
G.add_node('P-BJ-01', role='P', vendor='Huawei', ip='10.1.1.3')

# 添加链路边
G.add_edge('PE-BJ-01', 'PE-BJ-02',
           src_port='GE1/0/0', dst_port='GE1/0/0',
           state='up', protocol='lldp', bandwidth='10G')

G.add_edge('PE-BJ-01', 'P-BJ-01',
           src_port='GE1/0/1', dst_port='GE2/0/0',
           state='up', protocol='lldp', bandwidth='10G')

# 拓扑指标
print(f"设备数: {G.number_of_nodes()}")
print(f"链路数: {G.number_of_edges()}")
print(f"连通分量: {nx.number_connected_components(G)}")
print(f"平均度: {sum(dict(G.degree()).values()) / G.number_of_nodes():.2f}")

3.2 图数据库存储

对于大型网络（数千节点），关系型数据库查询效率较低，图数据库是更好的选择：

# 使用 Neo4j 存储拓扑
from neo4j import GraphDatabase

driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))

def save_topology_to_neo4j(G):
    """将 NetworkX 图写入 Neo4j"""
    with driver.session() as session:
        # 写入节点
        for node, attrs in G.nodes(data=True):
            session.run(
                "MERGE (d:Device {name: $name}) "
                "SET d.role = $role, d.vendor = $vendor, d.ip = $ip",
                name=node, role=attrs.get('role'),
                vendor=attrs.get('vendor'), ip=attrs.get('ip')
            )

        # 写入边
        for src, dst, attrs in G.edges(data=True):
            session.run(
                "MATCH (a:Device {name: $src}), (b:Device {name: $dst}) "
                "MERGE (a)-[l:CONNECTED_TO]->(b) "
                "SET l.src_port = $sp, l.dst_port = $dp, l.state = $st",
                src=src, dst=dst,
                sp=attrs.get('src_port'), dp=attrs.get('dst_port'),
                st=attrs.get('state')
            )

save_topology_to_neo4j(G)

3.3 与 IPAM 数据联动

第320篇的 IPAM 系统提供设备与 IP 的映射关系，拓扑发现系统可直接消费 IPAM 输出：

{
  "discovery_summary": {
    "source_ipam": "ipam_database.json",
    "total_devices": 48,
    "discovered_links": 72,
    "protocols": ["lldp", "cdp", "arp_inference"],
    "timestamp": "2026-09-06T10:30:00Z"
  },
  "edges": [
    {
      "source": "PE-BJ-01",
      "source_ip": "10.1.1.1",
      "source_port": "GE1/0/0",
      "target": "PE-BJ-02",
      "target_ip": "10.1.1.2",
      "target_port": "GE1/0/0",
      "protocol": "lldp",
      "state": "up"
    }
  ]
}

四、自动化绘图方案

4.1 Graphviz 自动布局

Graphviz 是最经典的自动布局绘图引擎，支持 dot、neato、fdp 等多种布局算法：

from graphviz import Digraph

def render_topology_graphviz(G, output_file='topology'):
    """使用 Graphviz 渲染拓扑图"""
    dot = Digraph(comment='Network Topology')
    dot.attr(rankdir='LR', size='20,15', fontsize='12')

    # 定义节点样式
    role_styles = {
        'PE': {'shape': 'box', 'style': 'filled', 'fillcolor': '#4CAF50'},
        'P':  {'shape': 'ellipse', 'style': 'filled', 'fillcolor': '#2196F3'},
        'CE': {'shape': 'diamond', 'style': 'filled', 'fillcolor': '#FF9800'},
        'AG': {'shape': 'box', 'style': 'filled', 'fillcolor': '#9C27B0'}
    }

    # 添加节点
    for node, attrs in G.nodes(data=True):
        role = attrs.get('role', '')
        style = role_styles.get(role, {'shape': 'ellipse'})
        label = f"{node}\n{attrs.get('ip', '')}"
        dot.node(node, label=label, **style)

    # 添加边
    for src, dst, attrs in G.edges(data=True):
        label = f"{attrs.get('src_port', '')}\n↔\n{attrs.get('dst_port', '')}"
        color = 'green' if attrs.get('state') == 'up' else 'red'
        dot.edge(src, dst, label=label, color=color, fontsize='10')

    dot.render(output_file, format='png', cleanup=True)
    return f'{output_file}.png'

render_topology_graphviz(G)

4.2 Mermaid 交互式拓扑

Mermaid 可生成嵌入 Markdown 的交互式拓扑图：

flowchart TD
    A["PE-BJ-01<br/>10.1.1.1<br/>GE1/0/0"] -->|"10G up"| B["PE-BJ-02<br/>10.1.1.2<br/>GE1/0/0"]
    A -->|"10G up"| C["P-BJ-01<br/>10.1.1.3<br/>GE2/0/0"]
    B -->|"10G up"| C
    C -->|"10G up"| D["P-BJ-02<br/>10.1.1.4"]
    D -->|"10G up"| B

    style A fill:#4CAF50,color:#fff
    style B fill:#4CAF50,color:#fff
    style C fill:#2196F3,color:#fff
    style D fill:#2196F3,color:#fff

4.3 动态拓扑 Web 展示

结合 D3.js 或 Cytoscape.js 可构建可拖拽、缩放的动态拓扑页面：

// Cytoscape.js 示例
var cy = cytoscape({
    container: document.getElementById('cy'),
    elements: [
        // 从后端 API 加载节点和边
    ],
    style: [
        {
            selector: 'node',
            style: {
                'label': 'data(label)',
                'background-color': 'data(color)',
                'width': 60,
                'height': 60
            }
        },
        {
            selector: 'edge',
            style: {
                'width': 2,
                'line-color': 'data(state) === "up" ? "#4CAF50" : "#F44336"'
            }
        }
    ],
    layout: { name: 'cose' }
});

五、完整发现流程与增量更新

5.1 端到端发现管线

将采集、解析、存储、渲染串联为一个自动化流程：

#!/usr/bin/env python3
"""
topology_discovery.py — 全网拓扑自动发现管线
"""
import yaml
import logging
import networkx as nx
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
from datetime import datetime
from pathlib import Path

class TopologyDiscoverer:
    """拓扑自动发现器"""

    def __init__(self, inventory_file='inventory.yaml'):
        with open(inventory_file) as f:
            self.inventory = yaml.safe_load(f)
        self.graph = nx.Graph()
        self.logger = logging.getLogger(__name__)

    def discover_all(self, max_workers=10):
        """并行发现所有设备拓扑"""
        devices = self.inventory.get('devices', [])
        all_edges = []
        errors = {}

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_map = {
                executor.submit(self._discover_device, d): d['name']
                for d in devices
            }
            for future in as_completed(future_map):
                name = future_map[future]
                try:
                    edges = future.result()
                    all_edges.extend(edges)
                except Exception as e:
                    errors[name] = str(e)
                    self.logger.error(f"发现 {name} 失败: {e}")

        # 构建图
        self._build_graph(all_edges)
        return errors

    def _discover_device(self, device):
        """单设备拓扑发现"""
        neighbors = get_lldp_neighbors(
            device['host'], device['username'], device['password']
        )
        edges = []
        for n in neighbors:
            edges.append({
                'source': device['name'],
                'source_port': n['local_interface'],
                'target': n['remote_device'],
                'target_port': n['remote_interface'],
                'state': n['status'],
                'protocol': 'lldp'
            })
        return edges

    def _build_graph(self, edges):
        """从边列表构建 NetworkX 图"""
        for e in edges:
            self.graph.add_edge(e['source'], e['target'],
                               src_port=e['source_port'],
                               dst_port=e['target_port'],
                               state=e['state'],
                               protocol=e['protocol'])
        # 补全节点属性
        for d in self.inventory.get('devices', []):
            if d['name'] in self.graph:
                self.graph.nodes[d['name']].update({
                    'role': d.get('role', ''),
                    'vendor': d.get('vendor', ''),
                    'ip': d.get('host', '')
                })

    def export_json(self, output_file='topology.json'):
        """导出为 JSON"""
        data = {
            'timestamp': datetime.now().isoformat(),
            'devices': self.graph.number_of_nodes(),
            'links': self.graph.number_of_edges(),
            'connected_components': nx.number_connected_components(self.graph),
            'nodes': [
                {'name': n, **attrs}
                for n, attrs in self.graph.nodes(data=True)
            ],
            'edges': [
                {'source': s, 'target': t, **attrs}
                for s, t, attrs in self.graph.edges(data=True)
            ]
        }
        Path(output_file).write_text(json.dumps(data, indent=2, ensure_ascii=False))
        return data

    def export_graphml(self, output_file='topology.graphml'):
        """导出 GraphML 格式（支持 Gephi 等工具打开）"""
        nx.write_graphml(self.graph, output_file)

    def get_shortest_path(self, src, dst):
        """计算最短路径（跳数）"""
        try:
            path = nx.shortest_path(self.graph, src, dst)
            return path
        except nx.NetworkXNoPath:
            return None

    def find_critical_links(self):
        """识别桥接链路（移除后导致网络分裂）"""
        bridges = list(nx.bridges(self.graph))
        return bridges

    def get_device_neighbors(self, device_name):
        """获取某设备的直连邻居"""
        if device_name not in self.graph:
            return []
        return list(self.graph.neighbors(device_name))

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    td = TopologyDiscoverer()
    errors = td.discover_all()
    data = td.export_json()
    td.export_graphml()
    render_topology_graphviz(td.graph)

    print(f"\n发现结果: {data['devices']} 设备, {data['links']} 链路")
    print(f"连通分量: {data['connected_components']}")
    print(f"桥接链路: {len(td.find_critical_links())}")
    if errors:
        print(f"失败设备: {list(errors.keys())}")

5.2 拓扑变更检测

每次发现后与上次快照对比，输出变更报告：

def compare_topology(old_graph, new_graph):
    """对比两次拓扑发现的差异"""
    old_edges = set((min(s,t), max(s,t)) for s,t in old_graph.edges())
    new_edges = set((min(s,t), max(s,t)) for s,t in new_graph.edges())

    added = new_edges - old_edges
    removed = old_edges - new_edges

    old_nodes = set(old_graph.nodes())
    new_nodes = set(new_graph.nodes())
    devices_added = new_nodes - old_nodes
    devices_removed = old_nodes - new_nodes

    report = {
        'timestamp': datetime.now().isoformat(),
        'topology_changed': bool(added or removed or devices_added or devices_removed),
        'devices_added': list(devices_added),
        'devices_removed': list(devices_removed),
        'links_added': [dict(source=s, target=t) for s,t in added],
        'links_removed': [dict(source=s, target=t) for s,t in removed],
        'summary': f"设备 {len(devices_added)}+{len(devices_removed)}-, "
                   f"链路 {len(added)}+{len(removed)}-"
    }

    if report['topology_changed']:
        print(f"⚠ 拓扑变更: {report['summary']}")
        for item in report['links_removed']:
            print(f"  ❌ 链路断开: {item['source']} ↔ {item['target']}")
        for item in report['links_added']:
            print(f"  ✅ 链路新增: {item['source']} ↔ {item['target']}")

    return report

5.3 关键链路分析

利用图论算法识别网络中的薄弱环节：

def analyze_network_health(G):
    """网络健康度分析"""
    analysis = {}

    # 1. 桥接链路（桥接边：移除后网络分裂）
    bridges = list(nx.bridges(G))
    analysis['bridges'] = bridges
    analysis['bridge_count'] = len(bridges)

    # 2. 孤立节点
    isolated = list(nx.isolated_nodes(G))
    analysis['isolated_nodes'] = isolated

    # 3. 度数分布（度数为1的设备为网络边缘）
    degrees = dict(G.degree())
    analysis['leaf_nodes'] = [n for n, d in degrees.items() if d == 1]
    analysis['avg_degree'] = sum(degrees.values()) / len(degrees) if degrees else 0

    # 4. 直径（最长最短路径）
    try:
        analysis['diameter'] = nx.diameter(G)
    except nx.NetworkXError:
        analysis['diameter'] = None  # 不连通

    # 5. 中心性分析（最重要的设备）
    centrality = nx.betweenness_centrality(G)
    analysis['most_critical'] = sorted(
        centrality.items(), key=lambda x: -x[1]
    )[:5]

    return analysis

五（续）、高级主题

5.4 BGP 逻辑拓扑发现

LLDP/CDP 只能发现物理直连邻居。要绘制 BGP 逻辑拓扑（路由传播路径），需要采集 BGP 邻居和路由表：

# scripts/bgp_topology_discovery.py
import napalm
import networkx as nx
from collections import defaultdict

def discover_bgp_topology(devices_yaml):
    """基于 NAPALM 采集 BGP 逻辑拓扑"""
    import yaml

    with open(devices_yaml) as f:
        inventory = yaml.safe_load(f)

    # 创建有向图（BGP 路由传播是有方向的）
    bgp_graph = nx.DiGraph()
    route_paths = defaultdict(list)

    for group in inventory.get("groups", []):
        for device in group.get("devices", []):
            driver = napalm.get_network_driver(device.get("driver", "huawei"))
            dev = driver(
                hostname=device["host"],
                username=device["username"],
                password=device["password"]
            )
            try:
                dev.open()
                bgp_info = dev.get_bgp_neighbors()
                bgp_routes = dev.get_bgp_routes()

                # 添加设备节点
                global_info = bgp_info.get("global", {})
                bgp_graph.add_node(
                    device["name"],
                    router_id=global_info.get("router_id", ""),
                    local_as=global_info.get("local_as", 0),
                    vendor=device.get("vendor", "unknown")
                )

                # 添加 BGP 对等关系边
                for peer_ip, peer_info in global_info.get("neighbors", {}).items():
                    peer_name = f"AS{peer_info.get('remote_as', '?')}-{peer_ip}"
                    if peer_info.get("is_up"):
                        bgp_graph.add_edge(
                            device["name"], peer_name,
                            type="eBGP" if peer_info.get("remote_as") != global_info.get("local_as") else "iBGP",
                            state="up",
                            remote_as=peer_info.get("remote_as")
                        )

                # 记录路由路径
                for prefix, routes in bgp_routes.items():
                    for route in routes:
                        route_paths[prefix].append({
                            "device": device["name"],
                            "next_hop": route.get("next_hop", ""),
                            "as_path": route.get("as_path", ""),
                            "local_pref": route.get("local_pref", 0),
                        })

                dev.close()

            except Exception as e:
                print(f"[-] {device['name']}: {e}")

    return bgp_graph, route_paths

def find_as_path(prefix, device_name, bgp_graph, route_paths):
    """查找某前缀在指定设备上的 AS Path"""
    paths = route_paths.get(prefix, [])
    for p in paths:
        if p["device"] == device_name:
            return p
    return None

# 使用示例
bgp_g, routes = discover_bgp_topology("inventory.yaml")
print(f"BGP 拓扑: {bgp_g.number_of_nodes()} 节点, {bgp_g.number_of_edges()} 对等关系")

# 查询特定前缀的路径
result = find_as_path("10.1.1.0/24", "PE-BJ-01", bgp_g, routes)
if result:
    print(f"AS Path: {result['as_path']}, Next-Hop: {result['next_hop']}")

5.5 拓扑数据导出格式对比

# 多格式导出
def export_topology(G, output_prefix="topology"):
    """导出拓扑为多种格式"""
    # JSON
    import json
    data = {
        "nodes": [{"id": n, **attrs} for n, attrs in G.nodes(data=True)],
        "edges": [{"source": s, "target": t, **attrs} for s, t, attrs in G.edges(data=True)]
    }
    with open(f"{output_prefix}.json", "w") as f:
        json.dump(data, f, indent=2)

    # GraphML
    nx.write_graphml(G, f"{output_prefix}.graphml")

    # CSV 边表
    import csv
    with open(f"{output_prefix}_edges.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["source", "target", "src_port", "dst_port", "state", "bandwidth"])
        for s, t, attrs in G.edges(data=True):
            writer.writerow([
                s, t,
                attrs.get("src_port", ""),
                attrs.get("dst_port", ""),
                attrs.get("state", ""),
                attrs.get("bandwidth", "")
            ])

    # DOT
    from graphviz import Digraph
    dot = Digraph()
    for n in G.nodes():
        dot.node(n, label=f"{n}\n{G.nodes[n].get('role', '')}")
    for s, t in G.edges():
        dot.edge(s, t)
    dot.render(f"{output_prefix}", format="png", cleanup=True)

    print(f"导出完成: .json, .graphml, _edges.csv, .png")

export_topology(G)

5.6 大规模拓扑发现优化

六、总结

自动化发现完整流程

故障案例：拓扑发现异常排查

某金融客户部署拓扑自动发现后，发现 30% 的链路未能发现。排查过程如下：

1. 检查 LLDP 状态：登录未发现链路的设备，display lldp enable 确认全局 LLDP 已开启，但部分接口未执行 lldp enable
2. 检查 SNMP 权限：部分老设备 SNMP community 为默认值，与采集器配置不一致
3. 检查防火墙策略：核心交换机启用了 ACL，过滤了 LLDP 目的 MAC 地址 01:80:C2:00:00:0E
4. 修复后重新采集：逐台补全 LLDP 接口配置、统一 SNMP community、放行 LLDP 报文，最终发现率提升至 99.2%

思考题

1. LLDP 只能发现直连邻居，如果需要绘制跨越多跳的逻辑拓扑（如 BGP 路由路径），应如何设计数据采集策略？
2. 在拓扑自动发现系统中，如何处理"链路来回振荡"（flapping）导致的数据噪声？
3. 拓扑变更检测中，如何区分"正常变更"（如计划内的设备替换）和"异常变更"（如未授权的配置修改）？
4. 当网络规模达到万级节点时，NetworkX 的性能是否足够？有哪些替代方案？

下篇预告：第322篇《CI/CD 在网络变更中的应用》——将软件工程的持续集成/持续交付理念引入网络变更，实现配置变更的自动化测试与灰度发布。