add traceroute source ip registration api

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

ryu/base/app_manager.py

@@ -58,7 +58,9 @@ class RyuApp(object):
         super(RyuApp, self).__init__()
         self.name = self.__class__.__name__
         self.event_handlers = {}
+        self.all_events_handlers = []
         self.observers = {}
+        self.observers_all = []
         self.threads = []
         self.events = hub.Queue(128)
         self.replies = hub.Queue()
@@ -67,15 +69,22 @@ class RyuApp(object):
 
     def register_handler(self, ev_cls, handler):
         assert callable(handler)
-        self.event_handlers.setdefault(ev_cls, [])
+        if ev_cls is None:
-        self.event_handlers[ev_cls].append(handler)
+            self.all_events_handlers.append(handler)
+        else:
+            self.event_handlers.setdefault(ev_cls, [])
+            self.event_handlers[ev_cls].append(handler)
 
     def register_observer(self, ev_cls, name, states=None):
         states = states or []
         self.observers.setdefault(ev_cls, {})[name] = states
 
+    def register_observer_all(self, name):
+        self.observers_all.append(name)
+
     def get_handlers(self, ev):
-        return self.event_handlers.get(ev.__class__, [])
+        handlers = self.event_handlers.get(ev.__class__, [])
+        return set(self.all_events_handlers) | set(handlers)
 
     def get_observers(self, ev, state):
         observers = []
@@ -118,7 +127,9 @@ class RyuApp(object):
                       (self.name, name, ev.__class__.__name__))
 
     def send_event_to_observers(self, ev, state=None):
-        for observer in self.get_observers(ev, state):
+        observers = set(self.get_observers(ev, state))
+        observers |= set(self.observers_all)
+        for observer in observers:
             self.send_event(observer, ev)
 
     def reply_to_request(self, req, rep):
@@ -220,6 +231,14 @@ class AppManager(object):
                         brick.register_observer(m.ev_cls, i.name,
                                                 m.dispatchers)
 
+            for _k, m in inspect.getmembers(i, inspect.ismethod):
+                if not hasattr(m, 'observer_all'):
+                    continue
+                name = m.observer_all
+                if name in SERVICE_BRICKS:
+                    brick = SERVICE_BRICKS[name]
+                    brick.register_observer_all(i.name)
+
         for brick, i in SERVICE_BRICKS.items():
             LOG.debug("BRICK %s" % brick)
             for ev_cls, list in i.observers.items():

ryu/controller/api.py

@@ -0,0 +1,293 @@
+import eventlet
+import sys
+import inspect
+import netaddr
+from ryu.base import app_manager
+from ryu.controller import handler
+from ryu.controller import dpset
+from ryu.controller import ofp_event
+from ryu.ofproto.ofproto_parser import MsgBase
+from ryu.lib.rpc import RpcSession, RpcMessage
+from ryu.ofproto import ofproto_parser
+from ryu.ofproto import ofproto_v1_0
+from ryu.ofproto import ofproto_v1_0_parser
+from ryu.ofproto import ofproto_v1_2
+from ryu.ofproto import ofproto_v1_2_parser
+from ryu.ofproto import ofproto_v1_3
+from ryu.ofproto import ofproto_v1_3_parser
+from ryu.lib.packet import packet
+from ryu.lib.packet import ethernet
+from ryu.lib.packet import arp
+from ryu.lib.packet import vlan
+from ryu.lib.packet import ipv4
+from ryu.lib.packet import ipv6
+from ryu.lib.packet import udp
+from ryu.lib.packet import tcp
+from ryu.lib.packet import icmp
+from ryu.lib import mac
+from ryu.lib import ip as ip_lib
+import netaddr
+import array
+
+
+def find_ofp_cls(ofp_version, name):
+    parser_name = 'ryu.ofproto.ofproto_v1_' + str(ofp_version - 1) + \
+        '_parser'
+    mod = sys.modules[parser_name]
+    for i in inspect.getmembers(mod, lambda cls: (inspect.isclass(cls))):
+        if i[0] == name:
+            return i[1]
+    return None
+
+traceroute_source = {}
+
+class OFWireRpcSession(object):
+    def __init__(self, socket, dpset):
+        self.socket = socket
+        self.dpset = dpset
+        self.session = RpcSession()
+        self.pool = eventlet.GreenPool()
+        self.send_queue = eventlet.queue.Queue()
+        self.pool.spawn_n(self._send)
+
+    def _send(self):
+        while True:
+            m = self.send_queue.get()
+            self.socket.sendall(m)
+
+    def _ofp_handle_matchfield(self, clses, params):
+        matchfield = getattr(clses, 'make')(**params)
+        return matchfield
+
+    def _ofp_handle_match(self, clses, params):
+        match = clses()
+        for k, v in params.items():
+            if hasattr(match, 'set_' + k):
+                if k.startswith('ipv4_') or k.startswith('arp_spa') or k.startswith('arp_tpa'):
+                    if k.endswith('_masked'):
+                        addr = netaddr.IPNetwork(v).ip
+                        mask = netaddr.IPNetwork(v).netmask
+                        getattr(match, 'set_' + k)(int(addr), int(mask))
+                        continue
+                    else:
+                        v = int(netaddr.IPNetwork(v).ip)
+                getattr(match, 'set_' + k)(v)
+        return match
+
+    def _ofp_handle_params(self, dp, params):
+        for k, v in params.items():
+            if type(v) == dict:
+                self._ofp_handle_params(dp, v)
+                clses = find_ofp_cls(dp.ofproto.OFP_VERSION, k)
+                if clses is not None:
+                    if issubclass(clses, MsgBase):
+                        ins = clses(dp, **v)
+                    else:
+                        if k == 'OFPMatch' and dp.ofproto.OFP_VERSION > 2:
+                            ins = self._ofp_handle_match(clses, v)
+                        elif k == 'OFPMatchField':
+                            ins = self._ofp_handle_matchfield(clses, v)
+                        else:
+                            ins = clses(**v)
+                    params[k] = ins
+                    return ins
+                else:
+                    for key, value in v.items():
+                        params[k] = value
+                        break
+            elif type(v) == list:
+                ins = []
+                for i in v:
+                    ins.append(self._ofp_handle_params(dp, i))
+                params[k] = ins
+            else:
+                pass
+
+    def ofp_handle_request(self, msg):
+        dpid = msg[3][0]
+        params = msg[3][1]
+        dp = self.dpset.get(int(dpid))
+        if dp is None:
+            # hack
+            for k, v in self.dpset.get_all():
+                dp = v
+                break
+
+        self._ofp_handle_params(dp, params)
+        result = {}
+        for k, v in params.items():
+            dp.send_msg(v)
+            result = {'xid': v.xid}
+        r = self.session.create_response(msg[1], 0, result)
+        self.send_queue.put(r)
+
+    def _tr_handle_notify(self, msg):
+        params = msg[2][0]
+        traceroute_source[params['vlan']] = {
+            'ip': params['ip'],
+            'port': params['port']
+            }
+        print traceroute_source
+
+    def serve(self):
+        while True:
+            ret = self.socket.recv(4096)
+            if len(ret) == 0:
+                break
+            messages = self.session.get_messages(ret)
+            for m in messages:
+                if m[0] == RpcMessage.REQUEST:
+                    if m[2] == 'ofp':
+                        self.ofp_handle_request(m)
+                elif m[0] == RpcMessage.RESPONSE:
+                    pass
+                elif m[0] == RpcMessage.NOTIFY:
+                    if m[1] == 'traceroute':
+                        self._tr_handle_notify(m)
+                else:
+                    print "invalid type", m
+
+
+class RPCApi(app_manager.RyuApp):
+    _CONTEXTS = {
+        'dpset': dpset.DPSet,
+    }
+
+    def __init__(self, *args, **kwargs):
+        self.dpset = kwargs['dpset']
+        super(RPCApi, self).__init__(*args, **kwargs)
+        self.server = eventlet.listen(('', 50001))
+        self.pool = eventlet.GreenPool()
+        self.pool.spawn_n(self.serve)
+        self.sessions = []
+
+    def serve(self):
+        while True:
+            sock, address = self.server.accept()
+            session = OFWireRpcSession(sock, self.dpset)
+            self.sessions.append(session)
+            self.pool.spawn_n(session.serve)
+
+    def _ofp_handle_ob(self, dp, msg, params):
+        clses = find_ofp_cls(dp.ofproto.OFP_VERSION,
+                             msg.__class__.__name__)
+        if clses:
+            params[msg.__class__.__name__] = {}
+            _params = params[msg.__class__.__name__]
+        else:
+            _params = params
+
+        for i in ofproto_parser.ofp_attr(msg):
+            if i.startswith('_'):
+                continue
+            elif i == 'parser':
+                continue
+            elif i == 'serialize':
+                continue
+
+            v = getattr(msg, i)
+            if type(v) == dict:
+                _params[i] = {}
+                for key in v.keys():
+                    _params[i][key] = {}
+                    self._ofp_handle_ob(dp, v[key], _params[i][key])
+            elif type(v) == list:
+                ins = []
+                for j in v:
+                    d = {}
+                    self._ofp_handle_ob(dp, j, d)
+                    ins.append(d)
+                _params[i] = ins
+            elif type(v).__name__ == 'builtin_function_or_method':
+                pass
+            else:
+                _params[i] = v
+
+    # @handler.observe_all_events('ofp_event')
+    # def _handler(self, ev):
+    #     if hasattr(ev, 'msg'):
+    #         ofmsg = ev.msg
+    #         params = {}
+    #         self._ofp_handle_ob(ofmsg.datapath, ofmsg, params)
+    #         for k in params.keys():
+    #             params[k]['xid'] = ofmsg.xid
+    #         # for s in self.sessions:
+    #         #     m = s.session.create_notification('ofp', params)
+    #         #     s.send_queue.put(m)
+
+    def handle_traceroute(self, msg):
+        dp = msg.datapath
+        in_port = None
+        for f in msg.match.fields:
+            if f.header == ofproto_v1_2.OXM_OF_IN_PORT:
+                in_port = f.value
+
+        if in_port is None:
+            print "in_port is missing"
+            return
+
+        pkt = packet.Packet(msg.data)
+        if not ipv4.ipv4 in pkt:
+            print "ip header doesn't exit"
+            return
+
+        if vlan.vlan in pkt:
+            o_vlan = pkt.get_protocols(vlan.vlan)[0]
+            vlan_p = vlan.vlan(vid=o_vlan.vid)
+        else:
+            print "vlan header doesn't exit"
+            return
+
+        o_eth = pkt.get_protocols(ethernet.ethernet)[0]
+        print mac.haddr_to_str(o_eth.dst), mac.haddr_to_str(o_eth.src)
+        eth = ethernet.ethernet(o_eth.src, o_eth.dst, o_eth.ethertype)
+        o_ip = pkt.get_protocols(ipv4.ipv4)[0]
+        print ip_lib.ipv4_to_str(o_ip.dst), ip_lib.ipv4_to_str(o_ip.src)
+        # needs to set src properly for either side (vlan or mpls)
+        # ip = ipv4.ipv4(src=ip_lib.ipv4_to_bin(V1_GS_IP), dst=o_ip.src,
+        #                proto=1)
+
+        src_ip = traceroute_source[o_vlan.vid]['ip']
+        in_port = traceroute_source[o_vlan.vid]['port']
+        
+        ip = ipv4.ipv4(src=ip_lib.ipv4_to_bin(src_ip), dst=o_ip.src, proto=1)
+        ip_offset = o_eth.length
+        # vlan header
+        ip_offset += 4
+        data = msg.data[ip_offset:ip_offset +
+                        (o_ip.header_length * 4 + 8)]
+        ic = icmp.icmp(icmp.ICMP_TIME_EXCEEDED, 0, 0,
+                       icmp.TimeExceeded(data))
+
+        proto_list = [eth, vlan_p, ip, ic]
+
+        p = packet.Packet(protocols=proto_list)
+        p.serialize()
+        self.send_openflow_packet(dp=dp, packet=p.data,
+                                  port_no=ofproto_v1_2.OFPP_TABLE,
+                                  inport=in_port)
+
+    def send_openflow_packet(self, dp, packet, port_no,
+                             inport=ofproto_v1_2.OFPP_CONTROLLER):
+
+        actions = [dp.ofproto_parser.OFPActionOutput(port_no, 0)]
+        dp.send_packet_out(in_port=inport, actions=actions, data=packet)
+
+    @handler.set_ev_cls(ofp_event.EventOFPPacketIn)
+    def packet_in_handler(self, ev):
+        msg = ev.msg
+        print "trace"
+        if ofproto_v1_2.OFPR_INVALID_TTL == msg.reason:
+            print "zero ttl packet"
+            self.handle_traceroute(msg)
+            return
+
+    @handler.set_ev_cls(dpset.EventDP)
+    def handler_datapath(self, ev):
+        print "join"
+        if ev.enter:
+            print "dp joined"
+            dp = ev.dp
+            m = dp.ofproto_parser.OFPSetConfig(dp, 1 << 2, miss_send_len=1600)
+            dp.send_msg(m)
+

ryu/controller/handler.py

@@ -46,6 +46,13 @@ def set_ev_handler(ev_cls, dispatchers=None):
     return _set_ev_cls_dec
 
 
+def observe_all_events(source):
+    def _set_ev_cls_dec(handler):
+        handler.observer_all = source
+        return handler
+    return _set_ev_cls_dec
+
+
 def _is_ev_cls(meth):
     return hasattr(meth, 'ev_cls')
 
@@ -63,3 +70,5 @@ def register_instance(i):
         # LOG.debug('instance %s k %s m %s', i, _k, m)
         if _is_ev_cls(m):
             i.register_handler(m.ev_cls, m)
+        elif hasattr(m, 'observer_all'):
+            i.register_handler(None, m)

ryu/lib/packet/arp.py

@@ -16,6 +16,8 @@
 import struct
 
 from ryu.ofproto import ether
+from ryu.lib import ip
+from ryu.lib import mac
 from . import packet_base
 
 ARP_HW_TYPE_ETHERNET = 1  # ethernet hardware type
@@ -52,8 +54,12 @@ class arp(packet_base.PacketBase):
     _PACK_STR = '!HHBBH6sI6sI'
     _MIN_LEN = struct.calcsize(_PACK_STR)
 
-    def __init__(self, hwtype, proto, hlen, plen, opcode,
+    def __init__(self, hwtype=ARP_HW_TYPE_ETHERNET, proto=ether.ETH_TYPE_IP,
-                 src_mac, src_ip, dst_mac, dst_ip):
+                 hlen=6, plen=4, opcode=ARP_REQUEST,
+                 src_mac=mac.haddr_to_bin('ff:ff:ff:ff:ff:ff'),
+                 src_ip=ip.ipv4_to_bin('0.0.0.0'),
+                 dst_mac=mac.haddr_to_bin('ff:ff:ff:ff:ff:ff'),
+                 dst_ip=ip.ipv4_to_bin('0.0.0.0')):
         super(arp, self).__init__()
         self.hwtype = hwtype
         self.proto = proto

ryu/lib/packet/icmp.py

@@ -24,6 +24,7 @@ ICMP_DEST_UNREACH = 3
 ICMP_SRC_QUENCH = 4
 ICMP_REDIRECT = 5
 ICMP_ECHO_REQUEST = 8
+ICMP_TIME_EXCEEDED = 11
 
 
 class icmp(packet_base.PacketBase):
@@ -146,3 +147,20 @@ class echo(object):
             hdr += self.data
 
         return hdr
+
+
+@icmp.register_icmp_type(ICMP_TIME_EXCEEDED)
+class TimeExceeded(object):
+    _PACK_STR = '!4x'
+    _MIN_LEN = struct.calcsize(_PACK_STR)
+
+    def __init__(self, data=None):
+        self.data = data
+
+    def serialize(self):
+        hdr = bytearray(TimeExceeded._MIN_LEN)
+
+        if self.data is not None:
+            hdr += self.data
+
+        return hdr

ryu/lib/packet/ipv4.py

@@ -21,6 +21,7 @@ from . import icmp
 from . import udp
 from . import tcp
 from ryu.ofproto import inet
+from ryu.lib import ip
 
 
 IPV4_ADDRESS_PACK_STR = '!I'
@@ -64,9 +65,12 @@ class ipv4(packet_base.PacketBase):
     _PACK_STR = '!BBHHHBBHII'
     _MIN_LEN = struct.calcsize(_PACK_STR)
 
-    def __init__(self, version, header_length, tos, total_length,
+    def __init__(self, version=4, header_length=5, tos=0,
-                 identification, flags, offset, ttl, proto, csum,
+                 total_length=0, identification=0, flags=0,
-                 src, dst, option=None):
+                 offset=0, ttl=255, proto=0, csum=0,
+                 src=ip.ipv4_to_bin('0.0.0.0'),
+                 dst=ip.ipv4_to_bin('0.0.0.0'),
+                 option=None):
         super(ipv4, self).__init__()
         self.version = version
         self.header_length = header_length

ryu/lib/packet/mpls.py

@@ -44,7 +44,7 @@ class mpls(packet_base.PacketBase):
     _PACK_STR = '!I'
     _MIN_LEN = struct.calcsize(_PACK_STR)
 
-    def __init__(self, label, exp, bsb, ttl):
+    def __init__(self, label=0, exp=0, bsb=1, ttl=255):
         super(mpls, self).__init__()
         self.label = label
         self.exp = exp

ryu/lib/packet/packet.py

@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import inspect
+
 from . import packet_base
 from . import ethernet
 
@@ -31,15 +33,17 @@ class Packet(object):
     *data* should be omitted when encoding a packet.
     """
 
-    def __init__(self, data=None):
+    def __init__(self, data=None, protocols=None, parse_cls=ethernet.ethernet):
         super(Packet, self).__init__()
         self.data = data
-        self.protocols = []
+        if protocols is None:
+            self.protocols = []
+        else:
+            self.protocols = protocols
         self.protocol_idx = 0
         self.parsed_bytes = 0
         if self.data:
-            # Do we need to handle non ethernet?
+            self._parser(parse_cls)
-            self._parser(ethernet.ethernet)
 
     def _parser(self, cls):
         while cls:
@@ -83,6 +87,14 @@ class Packet(object):
 
         self.protocols.append(proto)
 
+    def get_protocols(self, protocol):
+        """Returns a list of protocols that matches to the specified protocol.
+        """
+        if isinstance(protocol, packet_base.PacketBase):
+            protocol = protocol.__class__
+        assert issubclass(protocol, packet_base.PacketBase)
+        return [p for p in self.protocols if isinstance(p, protocol)]
+
     def next(self):
         try:
             p = self.protocols[self.protocol_idx]
@@ -95,3 +107,21 @@ class Packet(object):
 
     def __iter__(self):
         return self
+
+    def __getitem__(self, idx):
+        return self.protocols[idx]
+
+    def __setitem__(self, idx, item):
+        self.protocols[idx] = item
+
+    def __delitem__(self, idx):
+        del self.protocols[idx]
+
+    def __len__(self):
+        return len(self.protocols)
+
+    def __contains__(self, protocol):
+        if (inspect.isclass(protocol) and
+                issubclass(protocol, packet_base.PacketBase)):
+            return protocol in [p.__class__ for p in self.protocols]
+        return protocol in self.protocols

ryu/lib/packet/vlan.py

@@ -43,7 +43,7 @@ class vlan(packet_base.PacketBase):
     _PACK_STR = "!HH"
     _MIN_LEN = struct.calcsize(_PACK_STR)
 
-    def __init__(self, pcp, cfi, vid, ethertype):
+    def __init__(self, pcp=0, cfi=0, vid=0, ethertype=ether.ETH_TYPE_IP):
         super(vlan, self).__init__()
         self.pcp = pcp
         self.cfi = cfi

ryu/lib/rpc.py

@@ -0,0 +1,37 @@
+import msgpack
+
+
+class RpcMessage(object):
+    REQUEST = 0
+    RESPONSE = 1
+    NOTIFY = 2
+
+
+class RpcSession(object):
+    def __init__(self):
+        super(RpcSession, self).__init__()
+        self._packer = msgpack.Packer()
+        self._unpacker = msgpack.Unpacker()
+        self._next_msgid = 0
+
+    def _create_msgid(self):
+        this_id = self._next_msgid
+        self._next_msgid += 1
+        return this_id
+
+    def create_request(self, method, params):
+        msgid = self._create_msgid()
+        return self._packer.pack([RpcMessage.REQUEST, msgid, method, params])
+
+    def create_response(self, msgid, error, result):
+        return self._packer.pack([RpcMessage.RESPONSE, msgid, error, result])
+
+    def create_notification(self, method, params):
+        return self._packer.pack([RpcMessage.NOTIFY, method, params])
+
+    def get_messages(self, data):
+        self._unpacker.feed(data)
+        messages = []
+        for msg in self._unpacker:
+            messages.append(msg)
+        return messages

ryu/ofproto/ofproto_parser.py

@@ -147,14 +147,18 @@ def msg_pack_into(fmt, buf, offset, *args):
     struct.pack_into(fmt, buf, offset, *args)
 
 
+def ofp_attr(msg):
+    exclude = ['_attributes']
+    try:
+        exclude += msg._attributes
+    except AttributeError:
+        pass
+    return set(dir(msg)) - set(exclude)
+
+
 def msg_str_attr(msg, buf, attr_list=None):
     if attr_list is None:
-        exclude = ['_attributes']
+        attr_list = ofp_attr(msg)
-        try:
-            exclude += msg._attributes
-        except AttributeError:
-            pass
-        attr_list = set(dir(msg)) - set(exclude)
     for attr in attr_list:
         val = getattr(msg, attr, None)
         if val is not None:

ryu/ofproto/ofproto_v1_2_parser.py

@@ -2104,16 +2104,6 @@ class MTVlanVid(OFPMatchField):
         self.value = value
         self.mask = mask
 
-    @classmethod
-    def field_parser(cls, header, buf, offset):
-        m = super(MTVlanVid, cls).field_parser(header, buf, offset)
-        m.value &= ~ofproto_v1_2.OFPVID_PRESENT
-        return m
-
-    def serialize(self, buf, offset):
-        self.value |= ofproto_v1_2.OFPVID_PRESENT
-        super(MTVlanVid, self).serialize(buf, offset)
-
 
 @OFPMatchField.register_field_header([ofproto_v1_2.OXM_OF_VLAN_PCP])
 class MTVlanPcp(OFPMatchField):