ryu实例---基于链路质量(时延)的最短路径转发

Posted 2022-12-10 楊木木8023

一、前言

预备知识：

(1) ryu网络拓扑发现：https://www.sdnlab.com/11576.html

(2) ryu基于跳数最短路径转发：https://blog.csdn.net/weixin_40042248/article/details/116977397?spm=1001.2014.3001.5501

(3) ryu网络时延探测：https://blog.csdn.net/weixin_40042248/article/details/117562160?spm=1001.2014.3001.5501

基于链路质量的数据转发，主要考虑了多条路径的不同的时延，根据时延动态调整转发路径，从而使得数据转发拥有最短的时延。

二、原理

关于时延探测原理和网络拓扑的发现原理，这里不再叙述，详细请参考前言部分的链接。

首先获取网络拓扑，将网络的各个节点的连接关系以及端口记录到列表中，然后再获取网络拓扑将其节点和边写入networkx的有向图中，利用时延探测得到每条链路的实时时延，将时延作为边权重更新在有向图中，最后利用shortest_path()方法得到基于权重的最短路径，根据路径获得每一跳的数据转发的输出端口，从而使得控制器正确按照最短时延下发转发命令。

如下图：

(1) 先获取节点关系'1-2':2, '2-3':2, '3-5':2, '1-h1':1, ......，这样就记录了连接关系和端口，方便获取每一跳的输出端口

(2) 初始权重都默认为0，将节点和边写入有向图，G.add_weighted_edges_from([[1,2,0], [2,3,0], ......])

(3) 利用LLDP和echo获取时延，存储，即'2-3-3': 0.5, '2-2-1': 0.3,......，然后更新如第二步的有向图中

(4) 计算出时延最短的路径，获取每一跳的转发端口

三、程序设计

关于获取时延的部分参考前言链接，这里主要介绍拓扑获取更新和转发处理部分，也就是相当于在时延探测程序中加入以下函数。

(1) 获取链路拓扑

通过topology/api提供get方法，获取链路拓扑的switch和link的信息，并将switch作为节点，link作为边，默认初始边权为0，写入有向图中，具体代码如下：

    # 获取网络链路拓扑
    # 即将节点和边信息写入有向图中，默认的权重为0
    @set_ev_cls(events)
    def get_topo(self, ev):
        switch_list = get_switch(self.topology_api_app)
        topo_switches = []
        # 得到每个设备的id，并写入图中作为图的节点
        for switch in switch_list:
            topo_switches.append(switch.dp.id)
        self.G.add_nodes_from(topo_switches)

        link_list = get_link(self.topology_api_app)
        self.links_src_dst = []
        # 将得到的链路的信息作为边写入图中
        # 注意这里links_src_dst时列表里列表，即[[],[],[]]，不能是元组，因为元组不可更改，也就是后面无法更新权重信息
        for link in link_list:
            self.links_src_dst.append([link.src.dpid, link.dst.dpid, 0])
        self.G.add_weighted_edges_from(self.links_src_dst)

        for link in link_list:
            self.links_src_dst.append([link.dst.dpid, link.src.dpid, 0])
        self.G.add_weighted_edges_from(self.links_src_dst)

(2) 更新拓扑信息

这里的更新主要是利用实时的链路时延信息，将时延作为权重信息更新进入有向图中，具体代码如下：

    # 更新拓扑信息，主要更新有向图的边的权重
    # 即，程序获取链路的实时时延，当时延变化时，就将新的时延作为权重写入有向图中
    def update_topo(self):
        # [[1, 2, 0], [3, 2, 0], [2, 1, 0], [2, 3, 0], [2, 1, 0], [2, 3, 0], [1, 2, 0], [3, 2, 0]]
        # '2-3-3': 0.000362396240234375, '2-2-1': 0.001207113265991211, '1-2-2': 0.0004553794860839844, '3-2-2': 0.00015854835510253906
        # 将link_Delay的时延作为权重更新进links_src_dst列表中，然后更新入有向图
        for key in self.link_Delay:
            list = key.split('-')
            l = (int(list[0]), int(list[2]))
            for i in self.links_src_dst:
                if l == (i[0], i[1]):
                    i[2] = self.link_Delay[key]

        self.G.add_weighted_edges_from(self.links_src_dst)

(3) 获取输出的端口

也就是利用有向图的边权计算源到目的权值和最小的路径，根据路径得到每一跳需要转发数据的端口，比如说控制器收到s1-s5发送的数据，获取最小权重和的路径是s1-s3-s5，此时需要将s1数据转发到s3中，这时的s1-s3的连接是s1的1端口连接的，所以out_port=1。具体代码如下：

    # 获取输出的端口，这里的输出端口是控制器指示数据转发时按照最短权重获得的输出端口进行数据转发
    def get_out_port(self, datapath, src, dst, in_port):
        global out_port
        dpid = datapath.id

        # 开始时，各个主机可能在图中不存在，因为开始ryu只获取了交换机的dpid，并不知道各主机的信息，
        # 所以需要将主机存入图中
        # 同时将记录主机和交换机之间的连接关系和端口
        if src not in self.G:
            self.G.add_node(src)
            self.G.add_weighted_edges_from([[dpid, src, 0]])
            self.G.add_weighted_edges_from([[src, dpid, 0]])
            src_dst = "%s-%s" % (dpid, src)
            self.id_port[src_dst] = in_port

        # 计算出基于最小权重的链路，按照这个转发链路进行数据的转发
        if dst in self.G:
            path = nx.shortest_path(self.G, src, dst, weight='weight')
            next_hop = path[path.index(dpid) + 1]
            for key in self.id_port:
                match_key = "%s-%s" % (dpid, next_hop)
                if key == match_key:
                    out_port = self.id_port[key]
                    # print('key_out_port:', out_port)
            print(path)
        else:
            out_port = datapath.ofproto.OFPP_FLOOD
        return out_port

(4) 处理packet_in消息

packet_in消息主要是控制器处理交换机的数据，本程序主要包括处理LLDP和主机的转发信息两种。具体代码和解释如下：

    # 处理由交换机到来的消息，如LLDP消息和数据转发的消息
    @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
    def packet_in_handler(self, ev):
        msg = ev.msg
        datapath = msg.datapath
        ofp = datapath.ofproto
        ofp_parser = datapath.ofproto_parser
        dpid = datapath.id
        in_port = msg.match['in_port']

        pkt = packet.Packet(msg.data)
        eth = pkt.get_protocols(ethernet.ethernet)[0]

        dst = eth.dst
        src = eth.src

        # try...except，由于packetin中存在LLDP消息和主机的数据转发消息，
        # 二者格式不一样，所以用try...except进行控制，分别处理两种消息；
        try:  # 处理到达的LLDP报文，从而获得LLDP时延
            src_dpid, src_outport = LLDPPacket.lldp_parse(msg.data)  # 获取两个相邻交换机的源交换机dpid和port_no(与目的交换机相连的端口)
            dst_dpid = msg.datapath.id  # 获取目的交换机（第二个），因为来到控制器的消息是由第二个（目的）交换机上传过来的
            if self.switches is None:
                self.switches = lookup_service_brick("switches")  # 获取交换机模块实例

            # 获得key（Port类实例）和data（PortData类实例）
            for port in self.switches.ports.keys():  # 开始获取对应交换机端口的发送时间戳
                if src_dpid == port.dpid and src_outport == port.port_no:  # 匹配key
                    port_data = self.switches.ports[port]  # 获取满足key条件的values值PortData实例，内部保存了发送LLDP报文时的timestamp信息
                    timestamp = port_data.timestamp
                    if timestamp:
                        delay = time.time() - timestamp
                        self._save_delay_data(src=src_dpid, dst=dst_dpid, src_port=src_outport, lldp_dealy=delay)
        except Exception as error:  # 处理到达的主机的转发消息
            out_port = self.get_out_port(datapath, src, dst, in_port)
            actions = [ofp_parser.OFPActionOutput(out_port)]

            # 这里如果使用add_flow()进行了流表的添加，那么程序中的实时更新拓扑的权重就无意义了，转发就会依据流表进行
            # 所以这里不使用add_flow()方法，而是采用hub的形式，也就是每次转发都会请求控制器进行实时计算链路质量

            # 如果执行的动作不是flood，那么此时应该依据流表项进行转发操作，所以需要添加流表到交换机
            # if out_port != ofp.OFPP_FLOOD:
            #     match = ofp_parser.OFPMatch(in_port=in_port, eth_dst=dst, eth_src=src)
            #     self.add_flow(datapath=datapath, priority=1, match=match, actions=actions)

            data = None
            if msg.buffer_id == ofp.OFP_NO_BUFFER:
                data = msg.data
            # 控制器指导执行的命令
            out = ofp_parser.OFPPacketOut(datapath=datapath, buffer_id=msg.buffer_id,
                                          in_port=in_port, actions=actions, data=data)
            datapath.send_msg(out)

上述便是整个程序的关键部分，时延探测见前言链接，完整代码后面会给出GitHub链接。

四、实验验证

(1) 实验拓扑

(2) 启动ryu程序

ryu-manager quality_path_forward_yjl.py --observe-links

（3）ping验证

在mininet命令行输入h1 ping h3，建议在启动ryu程序之后过10几秒再进行验证测试，因为获取拓扑计算实验需要一段时间。

可以看见主机之间能够相互通信，接下来看ryu的打印日志，如下所示：

可以发现主机之间的通信路径会根据时延信息变化，各位同学可以自行进行权重变化验证是否会根据时延不同切换链路，这里不再验证。

以上便是利用网络时延进行数据转发的实例，如有不当和疑问，请各位同学指出，谢谢。

完整源码的GitHub地址：

https://github.com/Yang-Jianlin/ryu/blob/master/ryu/app/quality_path_forward_yjl.py