Ryu 源码阅读

SimpleMonitor13，SimpleSwitchRest13 两个类都继承自 SimpleSwitch13，分别扩展了流量监控和北向 RESTful API 功能，这篇以实验+源码剖析的方式简单总结下。

我的新书《LangChain编程从入门到实践》 已经开售！推荐正在学习AI应用开发的朋友购买阅读！

Switching Hub

• 传统交换机
  • 学习连接到传统交换机的主机的mac地址，并把其存在mac地址表中
  • 对于已经记录下来的mac地址，若是收到送往该mac地址的数据包时，就往对应的端口进行转发
  • 对于mac地址表中没有的数据包，则进行flooding
• OpenFlow 交换机实现传统交换机功能
  • 对接收到的数据包向指定的端口转发
  • 把接收到的数据包发送给控制器（Packet-In)
  • 把从控制器接收到的数据包转发到指定的端口（Packet-Out)

# simple_switch_13.py  实现一个带MAC地址学习功能的二层交换机
from ryu.base import app_manager
from ryu.controller import ofp_event
from ryu.controller.handler import CONFIG_DISPATCHER, MAIN_DISPATCHER
from ryu.controller.handler import set_ev_cls
from ryu.ofproto import ofproto_v1_3
from ryu.lib.packet import packet
from ryu.lib.packet import ethernet
from ryu.lib.packet import ether_types


class SimpleSwitch13(app_manager.RyuApp):
    OFP_VERSIONS = [ofproto_v1_3.OFP_VERSION]

    def __init__(self, *args, **kwargs):
        super(SimpleSwitch13, self).__init__(*args, **kwargs)
        # MAC地址表的定义
        self.mac_to_port = {}
    # 控制器向OpenFlow交换机发送Features Request消息，请求OpenFlow交换机上传自己的详细参数。
    # OpenFlow交换机收到请求后，向控制器发送Features Reply消息，详细汇报自身参数，
    # 包括支持的buffer、流表数以及Actions等
    @set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)
    def switch_features_handler(self, ev):
        # ev.msg是用来存对应事件的OpenFlow消息类别实体，这里指的是OFPSwitchFeatures
        # datapath是和控制器通信的实体单元
        datapath = ev.msg.datapath
        ofproto = datapath.ofproto
        parser = datapath.ofproto_parser
        # install table-miss flow entry
        #
        # We specify NO BUFFER to max_len of the output action due to
        # OVS bug. At this moment, if we specify a lesser number, e.g.,
        # 128, OVS will send Packet-In with invalid buffer_id and
        # truncated packet data. In that case, we cannot output packets
        # correctly.  The bug has been fixed in OVS v2.1.0.
# OpenFlow1.3版本为处理table miss事件专门引入的条目。并规定每一个flow table必须要支持table-miss flow entry去处理table miss情况。
# table-miss flow entry具备最低的优先级(0);必须至少能够支持使用CONTROLLER保留端口发送包，使用Clear-Actions指令丢包；
# table-miss flow entry和其他flow entry具有相同的特性：默认不存在，控制器可以随时添加或丢弃该条目，也可以到期；如果使用CONTROLLER保留端口发生数据包，Packet-In发送原因必须标明table-miss。

        match = parser.OFPMatch()
        # OFPActionOutput() ：使用一个packet_out 消息去指定你想从交换机的哪个端口发送出数据包。在该应用中，按照标准指定通过CONTROLLER保留端口发生，所以选择了OFPP_CONTROLLER端口。第二个参数OFPCML_NO_BUFFER，指明：消息中必须包含完整的包,而不会被缓冲。
        actions = [parser.OFPActionOutput(ofproto.OFPP_CONTROLLER,
                                          ofproto.OFPCML_NO_BUFFER)]
        # 下发Table-miss Flow Entry优先级为0的流表,即如果报文都没有匹配的话,则匹配该报文,并将其发送给控制器
        self.add_flow(datapath, 0, match, actions)

    # add_flow方法用来发送Flow Mod消息
    def add_flow(self, datapath, priority, match, actions, buffer_id=None):
        ofproto = datapath.ofproto
        parser = datapath.ofproto_parser
        # Instruction是1.0版本中Actions的拓展，Instruction主要负责将流表转发到其他Table，或者进行其他转发操作，# OFPIT_APPLY_ACTIONS: 立即应用actions操作到交换机
        inst = [parser.OFPInstructionActions(ofproto.OFPIT_APPLY_ACTIONS,
                                             actions)]
        if buffer_id:
            mod = parser.OFPFlowMod(datapath=datapath, buffer_id=buffer_id,
                                    priority=priority, match=match,
                                    instructions=inst)
        else:
            mod = parser.OFPFlowMod(datapath=datapath, priority=priority,
                                    match=match, instructions=inst)
        datapath.send_msg(mod)

    # 处理Packet-in数据包，交换机状态为正常状态
    @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
    def _packet_in_handler(self, ev):
        # If you hit this you might want to increase
        # the "miss_send_length" of your switch
        if ev.msg.msg_len < ev.msg.total_len:
            self.logger.debug("packet truncated: only %s of %s bytes",
                              ev.msg.msg_len, ev.msg.total_len)
        msg = ev.msg
        datapath = msg.datapath
        ofproto = datapath.ofproto
        parser = datapath.ofproto_parser
        # match 用来存储数据包的Meta元数据
        in_port = msg.match['in_port']
        # data 接受数据包本身的信息
        pkt = packet.Packet(msg.data)
        eth = pkt.get_protocols(ethernet.ethernet)[0]

        if eth.ethertype == ether_types.ETH_TYPE_LLDP:
            # ignore lldp packet
            return
        dst = eth.dst
        src = eth.src

        # 更新mac地址表,每台交换机独立的dpid对应一个表
        dpid = format(datapath.id, "d").zfill(16)
        self.mac_to_port.setdefault(dpid, {})

        self.logger.info("packet in %s %s %s %s", dpid, src, dst, in_port)

        # learn a mac address to avoid FLOOD next time.
        self.mac_to_port[dpid][src] = in_port

        if dst in self.mac_to_port[dpid]:
            out_port = self.mac_to_port[dpid][dst]
        else:
            out_port = ofproto.OFPP_FLOOD

        actions = [parser.OFPActionOutput(out_port)]

        # install a flow to avoid packet_in next time
        if out_port != ofproto.OFPP_FLOOD:
            match = parser.OFPMatch(in_port=in_port, eth_dst=dst, eth_src=src)
            # verify if we have a valid buffer_id, if yes avoid to send both
            # flow_mod & packet_out
            if msg.buffer_id != ofproto.OFP_NO_BUFFER:
                self.add_flow(datapath, 1, match, actions, msg.buffer_id)
                return
            else:
                self.add_flow(datapath, 1, match, actions)
        data = None
        if msg.buffer_id == ofproto.OFP_NO_BUFFER:
            data = msg.data

        out = parser.OFPPacketOut(datapath=datapath, buffer_id=msg.buffer_id,
                                  in_port=in_port, actions=actions, data=data)
        datapath.send_msg(out)

实验测试

mininet端执行

$ sudo mn --topo single,3 --mac --switch ovsk,protocols=OpenFlow13 --controller remote
*** Creating network
*** Adding controller
Unable to contact the remote controller at 127.0.0.1:6653
Unable to contact the remote controller at 127.0.0.1:6633
Setting remote controller to 127.0.0.1:6653
*** Adding hosts:
h1 h2 h3
*** Adding switches:
s1
*** Adding links:
(h1, s1) (h2, s1) (h3, s1)
*** Configuring hosts
h1 h2 h3
*** Starting controller
c0
*** Starting 1 switches
s1 ...
*** Starting CLI:

ovs端执行

$ sudo ovs-vsctl show
0f735673-a50c-4059-95bf-6ee3fd50c39c
    Bridge "s1"
        Controller "tcp:127.0.0.1:6653"
        Controller "ptcp:6654"
        fail_mode: secure
        Port "s1"
            Interface "s1"
                type: internal
        Port "s1-eth2"
            Interface "s1-eth2"
        Port "s1-eth1"
            Interface "s1-eth1"
        Port "s1-eth3"
            Interface "s1-eth3"
    ovs_version: "2.9.5"

因为此时的交换机中没有任何可匹配的转发策略，在mininet端执行pingall，结果全丢包

mininet> pingall
*** Ping: testing ping reachability
h1 -> X X
h2 -> X X
h3 -> X X
*** Results: 100% dropped (0/6 received)

再开一个终端，开启ryu

执行ryu-manager --verbose ryu.app.simple_switch_13

• 在ovs端查看流表，Table-miss Flow Entry 已加入OVS

  $ sudo ovs-ofctl -O OpenFlow13 dump-flows s1
  cookie=0x0, duration=28.824s, table=0, n_packets=0, n_bytes=0, priority=0 actions=CONTROLLER:65535

• mininet端

  mininet> h1 ping -c1 h2
  PING 10.0.0.2 (10.0.0.2) 56(84) bytes of data.
  64 bytes from 10.0.0.2: icmp_seq=1 ttl=64 time=5.18 ms
  
  --- 10.0.0.2 ping statistics ---
  1 packets transmitted, 1 received, 0% packet loss, time 0ms
  rtt min/avg/max/mdev = 5.183/5.183/5.183/0.000 ms

• ovs端再次查看OVS流表

  $ sudo ovs-ofctl -O OpenFlow13 dump-flows s1
  cookie=0x0, duration=190.004s, table=0, n_packets=2, n_bytes=140, priority=1,in_port="s1-eth2",dl_dst=00:00:00:00:00:01 actions=output:"s1-eth1"
  cookie=0x0, duration=190.003s, table=0, n_packets=1, n_bytes=42, priority=1,in_port="s1-eth1",dl_dst=00:00:00:00:00:02 actions=output:"s1-eth2"
  cookie=0x0, duration=648.460s, table=0, n_packets=6, n_bytes=392, priority=0 actions=CONTROLLER:65535

• ryu端查看

  EVENT ofp_event->SimpleSwitch13 EventOFPPacketIn
  packet in 1 00:00:00:00:00:01 ff:ff:ff:ff:ff:ff 1
  EVENT ofp_event->SimpleSwitch13 EventOFPPacketIn
  packet in 1 00:00:00:00:00:02 00:00:00:00:00:01 2
  EVENT ofp_event->SimpleSwitch13 EventOFPPacketIn
  packet in 1 00:00:00:00:00:01 00:00:00:00:00:02 1

Traffic Monitor

from operator import attrgetter

from ryu.app import simple_switch_13
from ryu.controller import ofp_event
from ryu.controller.handler import MAIN_DISPATCHER, DEAD_DISPATCHER
from ryu.controller.handler import set_ev_cls
from ryu.lib import hub


class SimpleMonitor13(simple_switch_13.SimpleSwitch13):

    def __init__(self, *args, **kwargs):
        super(SimpleMonitor13, self).__init__(*args, **kwargs)
        # 增添datapaths字典，存储交换机id
        self.datapaths = {}
        # 引入hub.spawn()函数启动一个新线程，输入为一个新的方法_monitor
        self.monitor_thread = hub.spawn(self._monitor)
    # 创建一个EventOFPStateChange监听事件，监听MAIN_DISPATCHER,DEAD_DISPATCHER两种情况。
    @set_ev_cls(ofp_event.EventOFPStateChange,
                [MAIN_DISPATCHER, DEAD_DISPATCHER])
    def _state_change_handler(self, ev):
        datapath = ev.datapath
        # 监听如果为MAIN_DISPATCHER,并且datapath.id不在datapath列表中，则证明是新加入的交换机
        if ev.state == MAIN_DISPATCHER:
            if datapath.id not in self.datapaths:
                self.logger.debug('register datapath: %016x', datapath.id)
                self.datapaths[datapath.id] = datapath
        # 如果为DEAD_DISPATCHER，并且datapath.id在datapath列表中，则证明是掉线的交换机
        elif ev.state == DEAD_DISPATCHER:
            if datapath.id in self.datapaths:
                self.logger.debug('unregister datapath: %016x', datapath.id)
                del self.datapaths[datapath.id]
    # _monitor方法，循环不断向datapath列表中的交换机发送Flow状态请求，和Port状态请求
    def _monitor(self):
        while True:
            for dp in self.datapaths.values():
                self._request_stats(dp)
            hub.sleep(10)

    def _request_stats(self, datapath):
        self.logger.debug('send stats request: %016x', datapath.id)
        ofproto = datapath.ofproto
        parser = datapath.ofproto_parser

        req = parser.OFPFlowStatsRequest(datapath)
        datapath.send_msg(req)

        req = parser.OFPPortStatsRequest(datapath, 0, ofproto.OFPP_ANY)
        datapath.send_msg(req)
    # 刚刚发送了请求，现在监听其回复
    @set_ev_cls(ofp_event.EventOFPFlowStatsReply, MAIN_DISPATCHER)
    def _flow_stats_reply_handler(self, ev):
        body = ev.msg.body

        self.logger.info('datapath         '
                         'in-port  eth-dst           '
                         'out-port packets  bytes')
        self.logger.info('---------------- '
                         '-------- ----------------- '
                         '-------- -------- --------')
        # 其中sorted ... lambda语法，指元组的排列顺序按照先in_port再eth_dst
        for stat in sorted([flow for flow in body if flow.priority == 1],
                           key=lambda flow: (flow.match['in_port'],
                                             flow.match['eth_dst'])):
            self.logger.info('%016x %8x %17s %8x %8d %8d',
                             ev.msg.datapath.id,
                             stat.match['in_port'], stat.match['eth_dst'],
                             stat.instructions[0].actions[0].port,
                             stat.packet_count, stat.byte_count)

    @set_ev_cls(ofp_event.EventOFPPortStatsReply, MAIN_DISPATCHER)
    def _port_stats_reply_handler(self, ev):
        body = ev.msg.body

        self.logger.info('datapath         port     '
                         'rx-pkts  rx-bytes rx-error '
                         'tx-pkts  tx-bytes tx-error')
        self.logger.info('---------------- -------- '
                         '-------- -------- -------- '
                         '-------- -------- --------')
        for stat in sorted(body, key=attrgetter('port_no')):
            self.logger.info('%016x %8x %8d %8d %8d %8d %8d %8d',
                             ev.msg.datapath.id, stat.port_no,
                             stat.rx_packets, stat.rx_bytes, stat.rx_errors,
                             stat.tx_packets, stat.tx_bytes, stat.tx_errors)

实验测试

• mininet端
  执行sudo mn --topo single,3 --mac --switch ovsk,protocols=OpenFlow13 --controller remote
• ryu端
  执行ryu-manager --verbose ryu.app.simple_monitor_13
• mininet端
  执行h1 ping -c1 h2，可观察到实时的流量检测

REST Linkage

RYU提供了一个类似WSGI的web服务器功能。借助这个功能，我们可以创建一个REST API，基于创建的REST API，可以快速的将RYU系统与其他系统或者是浏览器相连接。

import json

from ryu.app import simple_switch_13
from ryu.controller import ofp_event
from ryu.controller.handler import CONFIG_DISPATCHER
from ryu.controller.handler import set_ev_cls
from ryu.app.wsgi import ControllerBase
from ryu.app.wsgi import Response
from ryu.app.wsgi import route
from ryu.app.wsgi import WSGIApplication
from ryu.lib import dpid as dpid_lib

simple_switch_instance_name = 'simple_switch_api_app'
url = '/simpleswitch/mactable/{dpid}'

# 继承SimpleSwitch13的功能，即具备父类交换机的基本功能。
# 注册WSGI服务
# 配置mac_to_port
class SimpleSwitchRest13(simple_switch_13.SimpleSwitch13):

    _CONTEXTS = {'wsgi': WSGIApplication}

    def __init__(self, *args, **kwargs):
        super(SimpleSwitchRest13, self).__init__(*args, **kwargs)
        self.switches = {}
        # 通过上一步设置的_CONTEXTS成员变量，可以通过kwargs进行实例化一个WSGIApplication。
        # 同时使用register方法注册该服务到controller类上。
        wsgi = kwargs['wsgi']
        wsgi.register(SimpleSwitchController,
                      {simple_switch_instance_name: self})

    # 重写父类的switch_features_handler函数，存储datapath到switches，初始化MAC 地址表
    @set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)
    def switch_features_handler(self, ev):
        super(SimpleSwitchRest13, self).switch_features_handler(ev)
        datapath = ev.msg.datapath
        self.switches[datapath.id] = datapath
        self.mac_to_port.setdefault(datapath.id, {})
    # 该方法将MAC地址和端口注册到指定的交换机。该方法主要被REST API的PUT方法所调用
    def set_mac_to_port(self, dpid, entry):
        mac_table = self.mac_to_port.setdefault(dpid, {})
        datapath = self.switches.get(dpid)

        entry_port = entry['port']
        entry_mac = entry['mac']

        if datapath is not None:
            parser = datapath.ofproto_parser
            if entry_port not in mac_table.values():

                for mac, port in mac_table.items():

                    # from known device to new device
                    actions = [parser.OFPActionOutput(entry_port)]
                    match = parser.OFPMatch(in_port=port, eth_dst=entry_mac)
                    self.add_flow(datapath, 1, match, actions)

                    # from new device to known device
                    actions = [parser.OFPActionOutput(port)]
                    match = parser.OFPMatch(in_port=entry_port, eth_dst=mac)
                    self.add_flow(datapath, 1, match, actions)

                mac_table.update({entry_mac: entry_port})
        return mac_table


class SimpleSwitchController(ControllerBase):

    def __init__(self, req, link, data, **config):
        super(SimpleSwitchController, self).__init__(req, link, data, **config)
        self.simple_switch_app = data[simple_switch_instance_name]
    # 借助route装饰器关联方法和URL。参数如下：
    # 第一个参数：任何自定义名称
    # 第二个参数：指明URL
    # 第三个参数：指定http方法
    # 第四个参数：指明指定位置的格式，URL(/simpleswitch/mactable/{dpid} 匹配DPID_PATTERN的描述
    @route('simpleswitch', url, methods=['GET'],
           requirements={'dpid': dpid_lib.DPID_PATTERN})
    # 当使用GET方式访问到该REST API接口时，调用list_mac_table函数
    def list_mac_table(self, req, **kwargs):

        simple_switch = self.simple_switch_app
        dpid = kwargs['dpid']

        if dpid not in simple_switch.mac_to_port:
            return Response(status=404)

        mac_table = simple_switch.mac_to_port.get(dpid, {})
        body = json.dumps(mac_table)
        return Response(content_type='application/json', text=body)

    @route('simpleswitch', url, methods=['PUT'],
           requirements={'dpid': dpid_lib.DPID_PATTERN})
    def put_mac_table(self, req, **kwargs):

        simple_switch = self.simple_switch_app
        dpid = kwargs['dpid']
        try:
            new_entry = req.json if req.body else {}
        except ValueError:
            raise Response(status=400)

        if dpid not in simple_switch.mac_to_port:
            return Response(status=404)

        try:
            mac_table = simple_switch.set_mac_to_port(dpid, new_entry)
            body = json.dumps(mac_table)
            return Response(content_type='application/json', text=body)
        except Exception as e:
            return Response(status=500)

实验测试

• mininet端
  执行sudo mn --topo single,3 --mac --switch ovsk,protocols=OpenFlow13 --controller remote
• ryu端
  执行ryu-manager --verbose ryu.app.simple_switch_rest_13
• mininet端
  执行h1 ping -c1 h2，并观察ryu端的变化

使用curl调用 RESTful API 获取mac地址表
curl -X GET http://127.0.0.1:8080/simpleswitch/mactable /0000000000000001
使用curl调用 RESTful API 进行mac地址表的注册
curl -X PUT -d '{"mac" : "00:00:00:00:00:03", "port" : 3}' http://127.0.0.1:8080/simpleswitch/mactable/0000000000000001
查看流表可以看得到刚刚PUT的请求，也转化为流表的形式下发了

参考链接

RYUBook