如何通过程序查询Ryu控制器的网络状态信息？

完全可以通过Ryu的REST API或北向接口获取你需要的所有网络状态信息，以下是对应需求的具体实现方案和Python示例：

前置准备 #

启动Ryu时必须加载对应的REST模块，才能开启API接口：

ryu-manager --observe-links ryu.app.rest_topology ryu.app.rest_flow ryu.app.rest_conf_switch ryu.app.rest_application

--observe-links参数用于自动发现网络拓扑，是获取链路、主机信息的必要配置。

一、获取活跃流列表（转发/路由表） #

通过Ryu的/stats/flow REST端点可以获取所有交换机的流表信息，包含匹配规则、动作、优先级、流量统计等数据：

import requests

# 替换为你的Ryu控制器IP和端口（默认8080）
RYU_API_BASE = "http://127.0.0.1:8080"

def get_active_flows():
    """获取所有交换机的活跃流表"""
    resp = requests.get(f"{RYU_API_BASE}/stats/flow")
    if resp.status_code == 200:
        return resp.json()
    return f"请求失败，状态码: {resp.status_code}"

# 调用示例
flow_data = get_active_flows()
print(flow_data)

返回的JSON结构以交换机DPID为键，每个DPID对应一组流表项，可直接解析提取转发规则。

二、获取当前使用的协议 #

可以通过查询Ryu加载的应用列表推断运行的协议，或获取交换机的OpenFlow协商版本：

方案1：从已加载应用推断协议 #

def get_running_protocols():
    """获取控制器运行的网络协议"""
    resp = requests.get(f"{RYU_API_BASE}/v1.0/applications")
    if resp.status_code != 200:
        return f"请求失败，状态码: {resp.status_code}"
    
    app_list = resp.json()
    protocols = []
    protocol_mapping = {
        "ospf": "OSPF",
        "bgp": "BGP",
        "stp": "STP",
        "openflow": "OpenFlow"
    }
    
    for app in app_list:
        app_name = app["name"].lower()
        for key, proto in protocol_mapping.items():
            if key in app_name and proto not in protocols:
                protocols.append(proto)
    return protocols

# 调用示例
running_protos = get_running_protocols()
print(f"当前运行的协议: {running_protos}")

方案2：获取交换机OpenFlow版本 #

def get_switch_of_version(dpid):
    """获取指定交换机的OpenFlow协商版本"""
    resp = requests.get(f"{RYU_API_BASE}/conf/switches/{dpid}")
    if resp.status_code == 200:
        return resp.json().get("openflow_version")
    return f"请求失败，状态码: {resp.status_code}"

三、主机间连通性测试（类似Mininet pingall） #

Ryu本身没有直接的pingall接口，但可以结合拓扑信息+Mininet/ComNetsEmu的命令行工具实现：

import subprocess
from typing import List, Dict

def get_topology_hosts() -> List[Dict]:
    """获取拓扑中的所有主机信息"""
    resp = requests.get(f"{RYU_API_BASE}/v1.0/topology/hosts")
    return resp.json() if resp.status_code == 200 else []

def test_host_connectivity(host1: str, host2: str) -> bool:
    """测试两个主机的连通性（基于Mininet mnexec工具）"""
    # mnexec用于在Mininet命名空间中执行命令
    result = subprocess.run(
        ["mnexec", "-a", host1, "ping", "-c", "1", "-W", "1", host2],
        capture_output=True,
        text=True
    )
    return result.returncode == 0

def pingall_test():
    """执行类似Mininet pingall的全量连通性测试"""
    hosts = get_topology_hosts()
    if not hosts:
        print("未发现任何主机")
        return
    
    host_ids = [host["host_id"] for host in hosts]
    print("=== 主机连通性测试结果 ===")
    for i in range(len(host_ids)):
        for j in range(i+1, len(host_ids)):
            h1, h2 = host_ids[i], host_ids[j]
            reachable = test_host_connectivity(h1, h2)
            status = "✅ 可达" if reachable else "❌ 不可达"
            print(f"{h1} -> {h2}: {status}")

# 调用示例
pingall_test()

四、其他有用的网络信息 #

1. 完整拓扑结构（交换机、链路、主机） #

def get_full_topology():
    """获取完整的网络拓扑信息"""
    # 获取交换机列表
    switches_resp = requests.get(f"{RYU_API_BASE}/v1.0/topology/switches")
    switches = switches_resp.json() if switches_resp.status_code == 200 else []
    
    # 获取链路列表
    links_resp = requests.get(f"{RYU_API_BASE}/v1.0/topology/links")
    links = links_resp.json() if links_resp.status_code == 200 else []
    
    # 获取主机列表
    hosts = get_topology_hosts()
    
    return {
        "switches": switches,
        "links": links,
        "hosts": hosts
    }

2. 交换机端口状态 #

def get_switch_port_stats(dpid: str):
    """获取指定交换机的端口状态和流量统计"""
    resp = requests.get(f"{RYU_API_BASE}/stats/port/{dpid}")
    if resp.status_code == 200:
        return resp.json()
    return f"请求失败，状态码: {resp.status_code}"

内容的提问来源于stack exchange，提问作者Lucia Pintor