import StringIO import urllib import base64 import thread import time import os from webob import Response from ryu.topology import switches, event from ryu.base import app_manager from ryu.controller import ofp_event from ryu.controller import dpset from ryu.controller.handler import MAIN_DISPATCHER from ryu.controller.handler import set_ev_cls from ryu.ofproto import ofproto_v1_0 from ryu.app.wsgi import ControllerBase, WSGIApplication from ryu.lib.packet.packet import Packet from ryu.lib.ofctl_v1_0 import get_flow_stats import networkx as nx import matplotlib.pyplot as plt # REST API # draw the topology # GET /castflow/topology def get_as_uri(G): """Export graph as an html png. Arguments: G -- networkx.Graph -- the graph that will be exported Return: it returns a string containing the html representation """ #clear the figure, in case another one was already drawn. plt.clf() #draw the graph on the figure. nx.draw(G, nx.fruchterman_reingold_layout(G)) #save the figure on a stream. imgdata = StringIO.StringIO() plt.savefig(imgdata, format='png') #return to the beginning of the stream. imgdata.seek(0) #convert to 64 bit representation. buf64 = base64.b64encode(imgdata.buf) uri = 'data:image/png;base64,' + urllib.quote(buf64) #return the html code for the representation. return '' % uri def getPeriodicStats(dp): """show periodic stats on screen :param dp: the datapath from which the stats shall be shown """ waiters = {} while True: get_flow_stats(dp, waiters) #print stats time.sleep(1) class CastflowController(ControllerBase): """Contain the methods called on the REST interface. GETs supported: ip:port/castflow/topology show_topology: Show the current topology. """ def __init__(self, req, link, data, **config): """The parameters are the standard parameter for the REST interface. """ super(CastflowController, self).__init__(req, link, data, **config) self.castflow_api_app = data['castflow_api_app'] def show_topology(self, req, **kwargs): """Show the topology in PNG format. """ body = get_as_uri(self.castflow_api_app.topo) return Response(content_type='text/html', body=body) class Castflow(app_manager.RyuApp): """Main class, do switching and REST interface Based on SimpleSwitch class from the sample, this class does the switching, using many resources from RYU. Also added REST api interface, that can be accessed in: ip:port/castflow/topology (usually port is 8080). """ OFP_VERSIONS = [ofproto_v1_0.OFP_VERSION] #The contexts are necessary for some events and resources. _CONTEXTS = {'dpset': dpset.DPSet, # used for EventDP 'switches': switches.Switches, # used for EventLinkAdd 'wsgi': WSGIApplication} # used for REST stuff def __init__(self, *args, **kwargs): super(Castflow, self).__init__(*args, **kwargs) #dictionary used to remember which mac is in which port self.mac_to_port = {} #the topology self.topo = nx.Graph() wsgi = kwargs['wsgi'] mapper = wsgi.mapper #the controller for the REST API controller = CastflowController #an instance of this class will be passed to the controller wsgi.registory[controller.__name__] = {'castflow_api_app': self} route_name = 'castflow' #the path to access the function uri = '/castflow/topology' #the action is the name of the function that will be called mapper.connect(route_name, uri, controller=controller, action='show_topology', conditions=dict(method=['GET'])) def add_flow(self, datapath, in_port, dst, actions): """Add a flow to the datapath. datapath -- the datapath in which the flow will be installed in_port -- the port that reported the packet_in dst -- the mac to which the packet was sent actions -- the actions (mostly OFPActionOutput) Basicly, the datapath received a packet from in_port to be sent to dst and wants to do actions to it. """ ofproto = datapath.ofproto #the actions are applied to the packets that match this "match" match = datapath.ofproto_parser.OFPMatch(in_port=in_port, dl_dst=dst) #the flowmod message to install the rule on the datapath mod = datapath.ofproto_parser.OFPFlowMod( datapath=datapath, match=match, cookie=0, command=ofproto.OFPFC_ADD, idle_timeout=0, hard_timeout=0, priority=ofproto.OFP_DEFAULT_PRIORITY, flags=ofproto.OFPFF_SEND_FLOW_REM, actions=actions) datapath.send_msg(mod) @set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER) def _packet_in_handler(self, ev): """Handle packet_in event, installing the necessary flows. ev -- the event and all of it's fields. """ msg = ev.msg datapath = msg.datapath ofproto = datapath.ofproto #the packet that was sent pkt = Packet(msg.data) #the ethernet, another next would lead to ip and so on... eth = pkt.next() #eth has all the ethernet fields, including dst = eth.dst # destination MAC src = eth.src # source MAC _eth_type = eth.ethertype # packet type #the dpid that sent the message dpid = datapath.id self.mac_to_port.setdefault(dpid, {}) # learn a mac address to avoid FLOOD next time. self.mac_to_port[dpid][src] = msg.in_port if dst in self.mac_to_port[dpid]: #if destination is know send it to the right port out_port = self.mac_to_port[dpid][dst] else: #otherwise, flood out_port = ofproto.OFPP_FLOOD actions = [datapath.ofproto_parser.OFPActionOutput(out_port)] # install a flow to avoid packet_in next time if out_port != ofproto.OFPP_FLOOD: self.add_flow(datapath, msg.in_port, dst, actions) #flowmod message to be sent to the switch (datapath) out = datapath.ofproto_parser.OFPPacketOut( datapath=datapath, buffer_id=msg.buffer_id, in_port=msg.in_port, actions=actions) datapath.send_msg(out) @set_ev_cls(ofp_event.EventOFPPortStatus, MAIN_DISPATCHER) def _port_status_handler(self, ev): """Report port changed, if port was deleted, remove from topology.. ev -- PortStatus event, with all of it's fields. """ msg = ev.msg reason = msg.reason port_no = msg.desc.port_no #msg.datapath.id contains the dpid where the port was added #msg.desc.port_no contains which port ofproto = msg.datapath.ofproto if reason == ofproto.OFPPR_ADD: self.logger.info("port added %s", port_no) elif reason == ofproto.OFPPR_DELETE: self.logger.info("port deleted %s", port_no) elif reason == ofproto.OFPPR_MODIFY: self.logger.info("port modified %s", port_no) else: self.logger.info("Illeagal port state %s %s", port_no, reason) @set_ev_cls(dpset.EventDP, MAIN_DISPATCHER) def dp_handler(self, ev): """Update topology graph when a switch enters or leaves. ev -- datapath event and all of it's fields important fields: ev.dp.id: dpid that is joining or leaving ev.enter is true if the datapath is entering and false if it is leaving """ if ev.dp.id is None: return if ev.enter: self.topo.add_node(ev.dp.id) thread.start_new(getPeriodicStats, (ev.dp,)) self.logger.info('Switch %s added to the topology', str(ev.dp.id)) else: self.topo.remove_node(ev.dp.id) self.logger.info('Switch %s removed from the topology', str(ev.dp.id)) @set_ev_cls(event.EventLinkAdd, MAIN_DISPATCHER) def link_handler(self, ev): """Add new links to the topology graph ev -- LinkAdd event and all of it's fields. important fields: ev.link.src: tuple (srcdpid, srcport) ev.link.dst: tuple (dstdpid, dstport) """ src = ev.link.src dst = ev.link.dst if src.dpid is None or dst.dpid is None: return self.topo.add_edge(src.dpid, dst.dpid) self.topo[src.dpid][dst.dpid][src.dpid] = src.port_no self.topo[src.dpid][dst.dpid][dst.dpid] = dst.port_no self.logger.info('Link Detected, topology so far: %s', str(self.topo.edges())) @set_ev_cls(ofp_event.EventOFPFlowStatsReply, MAIN_DISPATCHER) def stats_reply_handler(self, ev): msg = ev.msg dp = msg.datapath print "Flows on datapath ", dp.id flows = msg.body i = 1 for flow in flows: print "Flow %d" % i print "Hard timeout: %d, Idle timeout: %d" % (flow.hard_timeout, flow.idle_timeout) print "Flow packet count: %d" % flow.packet_count print "Flow duration: %d" % flow.duration_sec print "Flow match:", flow.match print "Actions:", flow.actions i += 1 print "\n"