src/emuvim/dcemulator/net.py

   1 """
   2 Copyright (c) 2015 SONATA-NFV and Paderborn University
   3 ALL RIGHTS RESERVED.
   4
   5 Licensed under the Apache License, Version 2.0 (the "License");
   6 you may not use this file except in compliance with the License.
   7 You may obtain a copy of the License at
   8
   9     http://www.apache.org/licenses/LICENSE-2.0
  10
  11 Unless required by applicable law or agreed to in writing, software
  12 distributed under the License is distributed on an "AS IS" BASIS,
  13 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 See the License for the specific language governing permissions and
  15 limitations under the License.
  16
  17 Neither the name of the SONATA-NFV [, ANY ADDITIONAL AFFILIATION]
  18 nor the names of its contributors may be used to endorse or promote
  19 products derived from this software without specific prior written
  20 permission.
  21
  22 This work has been performed in the framework of the SONATA project,
  23 funded by the European Commission under Grant number 671517 through
  24 the Horizon 2020 and 5G-PPP programmes. The authors would like to
  25 acknowledge the contributions of their colleagues of the SONATA
  26 partner consortium (www.sonata-nfv.eu).
  27 """
  28 import logging
  29
  30 import site
  31 import time
  32 from subprocess import Popen
  33 import os
  34 import re
  35 import urllib2
  36 from functools import partial
  37
  38 from mininet.net import Containernet
  39 from mininet.node import Controller, DefaultController, OVSSwitch, OVSKernelSwitch, Docker, RemoteController
  40 from mininet.cli import CLI
  41 from mininet.link import TCLink
  42 from mininet.clean import cleanup
  43 import networkx as nx
  44 from emuvim.dcemulator.monitoring import DCNetworkMonitor
  45 from emuvim.dcemulator.node import Datacenter, EmulatorCompute
  46 from emuvim.dcemulator.resourcemodel import ResourceModelRegistrar
  47
  48 LOG = logging.getLogger("dcemulator.net")
  49 LOG.setLevel(logging.DEBUG)
  50
  51 class DCNetwork(Containernet):
  52     """
  53     Wraps the original Mininet/Containernet class and provides
  54     methods to add data centers, switches, etc.
  55
  56     This class is used by topology definition scripts.
  57     """
  58
  59     def __init__(self, controller=RemoteController, monitor=False,
  60                  enable_learning = True,   # in case of RemoteController (Ryu), learning switch behavior can be turned off/on
  61                  dc_emulation_max_cpu=1.0,  # fraction of overall CPU time for emulation
  62                  dc_emulation_max_mem=512,  # emulation max mem in MB
  63                  **kwargs):
  64         """
  65         Create an extended version of a Containernet network
  66         :param dc_emulation_max_cpu: max. CPU time used by containers in data centers
  67         :param kwargs: path through for Mininet parameters
  68         :return:
  69         """
  70         # members
  71         self.dcs = {}
  72         self.ryu_process = None
  73
  74         # always cleanup environment before we start the emulator
  75         self.killRyu()
  76         cleanup()
  77
  78         # call original Docker.__init__ and setup default controller
  79         Containernet.__init__(
  80             self, switch=OVSKernelSwitch, controller=controller, **kwargs)
  81
  82         # Ryu management
  83         if controller == RemoteController:
  84             # start Ryu controller
  85             self.startRyu(learning_switch=enable_learning)
  86
  87         # add the specified controller
  88         self.addController('c0', controller=controller)
  89
  90         # graph of the complete DC network
  91         self.DCNetwork_graph = nx.MultiDiGraph()
  92
  93         # initialize pool of vlan tags to setup the SDN paths
  94         self.vlans = range(4096)[::-1]
  95
  96         # link to Ryu REST_API
  97         ryu_ip = '0.0.0.0'
  98         ryu_port = '8080'
  99         self.ryu_REST_api = 'http://{0}:{1}'.format(ryu_ip, ryu_port)
 100
 101         # monitoring agent
 102         if monitor:
 103             self.monitor_agent = DCNetworkMonitor(self)
 104         else:
 105             self.monitor_agent = None
 106
 107         # initialize resource model registrar
 108         self.rm_registrar = ResourceModelRegistrar(
 109             dc_emulation_max_cpu, dc_emulation_max_mem)
 110
 111     def addDatacenter(self, label, metadata={}, resource_log_path=None):
 112         """
 113         Create and add a logical cloud data center to the network.
 114         """
 115         if label in self.dcs:
 116             raise Exception("Data center label already exists: %s" % label)
 117         dc = Datacenter(label, metadata=metadata, resource_log_path=resource_log_path)
 118         dc.net = self  # set reference to network
 119         self.dcs[label] = dc
 120         dc.create()  # finally create the data center in our Mininet instance
 121         LOG.info("added data center: %s" % label)
 122         return dc
 123
 124     def addLink(self, node1, node2, **params):
 125         """
 126         Able to handle Datacenter objects as link
 127         end points.
 128         """
 129         assert node1 is not None
 130         assert node2 is not None
 131         LOG.debug("addLink: n1=%s n2=%s" % (str(node1), str(node2)))
 132         # ensure type of node1
 133         if isinstance( node1, basestring ):
 134             if node1 in self.dcs:
 135                 node1 = self.dcs[node1].switch
 136         if isinstance( node1, Datacenter ):
 137             node1 = node1.switch
 138         # ensure type of node2
 139         if isinstance( node2, basestring ):
 140             if node2 in self.dcs:
 141                 node2 = self.dcs[node2].switch
 142         if isinstance( node2, Datacenter ):
 143             node2 = node2.switch
 144         # try to give containers a default IP
 145         if isinstance( node1, Docker ):
 146             if "params1" not in params:
 147                 params["params1"] = {}
 148             if "ip" not in params["params1"]:
 149                 params["params1"]["ip"] = self.getNextIp()
 150         if isinstance( node2, Docker ):
 151             if "params2" not in params:
 152                 params["params2"] = {}
 153             if "ip" not in params["params2"]:
 154                 params["params2"]["ip"] = self.getNextIp()
 155         # ensure that we allow TCLinks between data centers
 156         # TODO this is not optimal, we use cls=Link for containers and TCLink for data centers
 157         # see Containernet issue: https://github.com/mpeuster/containernet/issues/3
 158         if "cls" not in params:
 159             params["cls"] = TCLink
 160
 161         link = Containernet.addLink(self, node1, node2, **params)
 162
 163         # try to give container interfaces a default id
 164         node1_port_id = node1.ports[link.intf1]
 165         if isinstance(node1, Docker):
 166             if "id" in params["params1"]:
 167                 node1_port_id = params["params1"]["id"]
 168         node1_port_name = link.intf1.name
 169
 170         node2_port_id = node2.ports[link.intf2]
 171         if isinstance(node2, Docker):
 172             if "id" in params["params2"]:
 173                 node2_port_id = params["params2"]["id"]
 174         node2_port_name = link.intf2.name
 175
 176
 177         # add edge and assigned port number to graph in both directions between node1 and node2
 178         # port_id: id given in descriptor (if available, otherwise same as port)
 179         # port: portnumber assigned by Containernet
 180
 181         attr_dict = {}
 182         # possible weight metrics allowed by TClink class:
 183         weight_metrics = ['bw', 'delay', 'jitter', 'loss']
 184         edge_attributes = [p for p in params if p in weight_metrics]
 185         for attr in edge_attributes:
 186             # if delay: strip ms (need number as weight in graph)
 187             match = re.search('([0-9]*\.?[0-9]+)', params[attr])
 188             if match:
 189                 attr_number = match.group(1)
 190             else:
 191                 attr_number = None
 192             attr_dict[attr] = attr_number
 193
 194
 195         attr_dict2 = {'src_port_id': node1_port_id, 'src_port_nr': node1.ports[link.intf1],
 196                       'src_port_name': node1_port_name,
 197                      'dst_port_id': node2_port_id, 'dst_port_nr': node2.ports[link.intf2],
 198                       'dst_port_name': node2_port_name}
 199         attr_dict2.update(attr_dict)
 200         self.DCNetwork_graph.add_edge(node1.name, node2.name, attr_dict=attr_dict2)
 201
 202         attr_dict2 = {'src_port_id': node2_port_id, 'src_port_nr': node2.ports[link.intf2],
 203                       'src_port_name': node2_port_name,
 204                      'dst_port_id': node1_port_id, 'dst_port_nr': node1.ports[link.intf1],
 205                       'dst_port_name': node1_port_name}
 206         attr_dict2.update(attr_dict)
 207         self.DCNetwork_graph.add_edge(node2.name, node1.name, attr_dict=attr_dict2)
 208
 209         return link
 210
 211     def addDocker( self, label, **params ):
 212         """
 213         Wrapper for addDocker method to use custom container class.
 214         """
 215         self.DCNetwork_graph.add_node(label)
 216         return Containernet.addDocker(self, label, cls=EmulatorCompute, **params)
 217
 218     def removeDocker( self, label, **params ):
 219         """
 220         Wrapper for removeDocker method to update graph.
 221         """
 222         self.DCNetwork_graph.remove_node(label)
 223         return Containernet.removeDocker(self, label, **params)
 224
 225     def addSwitch( self, name, add_to_graph=True, **params ):
 226         """
 227         Wrapper for addSwitch method to store switch also in graph.
 228         """
 229         if add_to_graph:
 230             self.DCNetwork_graph.add_node(name)
 231         return Containernet.addSwitch(self, name, protocols='OpenFlow10,OpenFlow12,OpenFlow13', **params)
 232
 233     def getAllContainers(self):
 234         """
 235         Returns a list with all containers within all data centers.
 236         """
 237         all_containers = []
 238         for dc in self.dcs.itervalues():
 239             all_containers += dc.listCompute()
 240         return all_containers
 241
 242     def start(self):
 243         # start
 244         for dc in self.dcs.itervalues():
 245             dc.start()
 246         Containernet.start(self)
 247
 248     def stop(self):
 249
 250         # stop the monitor agent
 251         if self.monitor_agent is not None:
 252             self.monitor_agent.stop()
 253
 254         # stop emulator net
 255         Containernet.stop(self)
 256
 257         # stop Ryu controller
 258         self.killRyu()
 259
 260
 261     def CLI(self):
 262         CLI(self)
 263
 264     # to remove chain do setChain( src, dst, cmd='del-flows')
 265     def setChain(self, vnf_src_name, vnf_dst_name, vnf_src_interface=None, vnf_dst_interface=None, **kwargs):
 266         cmd = kwargs.get('cmd')
 267         if cmd == 'add-flow':
 268             ret = self._chainAddFlow(vnf_src_name, vnf_dst_name, vnf_src_interface, vnf_dst_interface, **kwargs)
 269             if kwargs.get('bidirectional'):
 270                 ret = ret +'\n' + self._chainAddFlow(vnf_dst_name, vnf_src_name, vnf_dst_interface, vnf_src_interface, **kwargs)
 271
 272         elif cmd == 'del-flows':
 273             ret = self._chainAddFlow(vnf_src_name, vnf_dst_name, vnf_src_interface, vnf_dst_interface, **kwargs)
 274             if kwargs.get('bidirectional'):
 275                 ret = ret + '\n' + self._chainAddFlow(vnf_dst_name, vnf_src_name, vnf_dst_interface, vnf_src_interface, **kwargs)
 276
 277         else:
 278             ret = "Command unknown"
 279
 280         return ret
 281
 282
 283     def _chainAddFlow(self, vnf_src_name, vnf_dst_name, vnf_src_interface=None, vnf_dst_interface=None, **kwargs):
 284
 285         src_sw = None
 286         dst_sw = None
 287         src_sw_inport_nr = 0
 288         dst_sw_outport_nr = 0
 289
 290         LOG.debug("call chainAddFlow vnf_src_name=%r, vnf_src_interface=%r, vnf_dst_name=%r, vnf_dst_interface=%r",
 291                   vnf_src_name, vnf_src_interface, vnf_dst_name, vnf_dst_interface)
 292
 293         #check if port is specified (vnf:port)
 294         if vnf_src_interface is None:
 295             # take first interface by default
 296             connected_sw = self.DCNetwork_graph.neighbors(vnf_src_name)[0]
 297             link_dict = self.DCNetwork_graph[vnf_src_name][connected_sw]
 298             vnf_src_interface = link_dict[0]['src_port_id']
 299
 300         for connected_sw in self.DCNetwork_graph.neighbors(vnf_src_name):
 301             link_dict = self.DCNetwork_graph[vnf_src_name][connected_sw]
 302             for link in link_dict:
 303                 if (link_dict[link]['src_port_id'] == vnf_src_interface or
 304                         link_dict[link]['src_port_name'] == vnf_src_interface):  # Fix: we might also get interface names, e.g, from a son-emu-cli call
 305                     # found the right link and connected switch
 306                     src_sw = connected_sw
 307                     src_sw_inport_nr = link_dict[link]['dst_port_nr']
 308                     break
 309
 310         if vnf_dst_interface is None:
 311             # take first interface by default
 312             connected_sw = self.DCNetwork_graph.neighbors(vnf_dst_name)[0]
 313             link_dict = self.DCNetwork_graph[connected_sw][vnf_dst_name]
 314             vnf_dst_interface = link_dict[0]['dst_port_id']
 315
 316         vnf_dst_name = vnf_dst_name.split(':')[0]
 317         for connected_sw in self.DCNetwork_graph.neighbors(vnf_dst_name):
 318             link_dict = self.DCNetwork_graph[connected_sw][vnf_dst_name]
 319             for link in link_dict:
 320                 if link_dict[link]['dst_port_id'] == vnf_dst_interface or \
 321                         link_dict[link]['dst_port_name'] == vnf_dst_interface:  # Fix: we might also get interface names, e.g, from a son-emu-cli call
 322                     # found the right link and connected switch
 323                     dst_sw = connected_sw
 324                     dst_sw_outport_nr = link_dict[link]['src_port_nr']
 325                     break
 326
 327
 328         # get shortest path
 329         try:
 330             # returns the first found shortest path
 331             # if all shortest paths are wanted, use: all_shortest_paths
 332             path = nx.shortest_path(self.DCNetwork_graph, src_sw, dst_sw, weight=kwargs.get('weight'))
 333         except:
 334             LOG.exception("No path could be found between {0} and {1} using src_sw={2} and dst_sw={3}".format(
 335                 vnf_src_name, vnf_dst_name, src_sw, dst_sw))
 336             LOG.debug("Graph nodes: %r" % self.DCNetwork_graph.nodes())
 337             LOG.debug("Graph edges: %r" % self.DCNetwork_graph.edges())
 338             for e, v in self.DCNetwork_graph.edges():
 339                 LOG.debug("%r" % self.DCNetwork_graph[e][v])
 340             return "No path could be found between {0} and {1}".format(vnf_src_name, vnf_dst_name)
 341
 342         LOG.info("Path between {0} and {1}: {2}".format(vnf_src_name, vnf_dst_name, path))
 343
 344         current_hop = src_sw
 345         switch_inport_nr = src_sw_inport_nr
 346
 347         # choose free vlan if path contains more than 1 switch
 348         cmd = kwargs.get('cmd')
 349         vlan = None
 350         if cmd == 'add-flow':
 351             if len(path) > 1:
 352                 vlan = self.vlans.pop()
 353
 354         for i in range(0,len(path)):
 355             current_node = self.getNodeByName(current_hop)
 356
 357             if path.index(current_hop) < len(path)-1:
 358                 next_hop = path[path.index(current_hop)+1]
 359             else:
 360                 #last switch reached
 361                 next_hop = vnf_dst_name
 362
 363             next_node = self.getNodeByName(next_hop)
 364
 365             if next_hop == vnf_dst_name:
 366                 switch_outport_nr = dst_sw_outport_nr
 367                 LOG.info("end node reached: {0}".format(vnf_dst_name))
 368             elif not isinstance( next_node, OVSSwitch ):
 369                 LOG.info("Next node: {0} is not a switch".format(next_hop))
 370                 return "Next node: {0} is not a switch".format(next_hop)
 371             else:
 372                 # take first link between switches by default
 373                 index_edge_out = 0
 374                 switch_outport_nr = self.DCNetwork_graph[current_hop][next_hop][index_edge_out]['src_port_nr']
 375
 376
 377            # set of entry via ovs-ofctl
 378             if isinstance( current_node, OVSSwitch ):
 379                 kwargs['vlan'] = vlan
 380                 kwargs['path'] = path
 381                 kwargs['current_hop'] = current_hop
 382
 383                 if self.controller == RemoteController:
 384                     ## set flow entry via ryu rest api
 385                     self._set_flow_entry_ryu_rest(current_node, switch_inport_nr, switch_outport_nr, **kwargs)
 386                 else:
 387                     ## set flow entry via ovs-ofctl
 388                     self._set_flow_entry_dpctl(current_node, switch_inport_nr, switch_outport_nr, **kwargs)
 389
 390
 391
 392             # take first link between switches by default
 393             if isinstance( next_node, OVSSwitch ):
 394                 switch_inport_nr = self.DCNetwork_graph[current_hop][next_hop][0]['dst_port_nr']
 395                 current_hop = next_hop
 396
 397         return "path {2} between {0} and {1}".format(vnf_src_name, vnf_dst_name, cmd)
 398
 399     def _set_flow_entry_ryu_rest(self, node, switch_inport_nr, switch_outport_nr, **kwargs):
 400         match = 'in_port=%s' % switch_inport_nr
 401
 402         cookie = kwargs.get('cookie')
 403         match_input = kwargs.get('match')
 404         cmd = kwargs.get('cmd')
 405         path = kwargs.get('path')
 406         current_hop = kwargs.get('current_hop')
 407         vlan = kwargs.get('vlan')
 408
 409         s = ','
 410         if match_input:
 411             match = s.join([match, match_input])
 412
 413         flow = {}
 414         flow['dpid'] = int(node.dpid, 16)
 415
 416         if cookie:
 417             flow['cookie'] = int(cookie)
 418
 419
 420         flow['actions'] = []
 421
 422         # possible Ryu actions, match fields:
 423         # http://ryu.readthedocs.io/en/latest/app/ofctl_rest.html#add-a-flow-entry
 424         if cmd == 'add-flow':
 425             prefix = 'stats/flowentry/add'
 426             if vlan != None:
 427                 if path.index(current_hop) == 0:  # first node
 428                     action = {}
 429                     action['type'] = 'PUSH_VLAN'  # Push a new VLAN tag if a input frame is non-VLAN-tagged
 430                     action['ethertype'] = 33024   # Ethertype 0x8100(=33024): IEEE 802.1Q VLAN-tagged frame
 431                     flow['actions'].append(action)
 432                     action = {}
 433                     action['type'] = 'SET_FIELD'
 434                     action['field'] = 'vlan_vid'
 435                     action['value'] = vlan
 436                     flow['actions'].append(action)
 437                 elif path.index(current_hop) == len(path) - 1:  # last node
 438                     match += ',dl_vlan=%s' % vlan
 439                     action = {}
 440                     action['type'] = 'POP_VLAN'
 441                     flow['actions'].append(action)
 442                 else:  # middle nodes
 443                     match += ',dl_vlan=%s' % vlan
 444             # output action must come last
 445             action = {}
 446             action['type'] = 'OUTPUT'
 447             action['port'] = switch_outport_nr
 448             flow['actions'].append(action)
 449
 450         elif cmd == 'del-flows':
 451             prefix = 'stats/flowentry/delete'
 452
 453             if cookie:
 454                 # TODO: add cookie_mask as argument
 455                 flow['cookie_mask'] = int('0xffffffffffffffff', 16)  # need full mask to match complete cookie
 456
 457             action = {}
 458             action['type'] = 'OUTPUT'
 459             action['port'] = switch_outport_nr
 460             flow['actions'].append(action)
 461
 462         flow['match'] = self._parse_match(match)
 463         self.ryu_REST(prefix, data=flow)
 464
 465     def _set_flow_entry_dpctl(self, node, switch_inport_nr, switch_outport_nr, **kwargs):
 466         match = 'in_port=%s' % switch_inport_nr
 467
 468         cookie = kwargs.get('cookie')
 469         match_input = kwargs.get('match')
 470         cmd = kwargs.get('cmd')
 471         path = kwargs.get('path')
 472         current_hop = kwargs.get('current_hop')
 473         vlan = kwargs.get('vlan')
 474
 475         s = ','
 476         if cookie:
 477             cookie = 'cookie=%s' % cookie
 478             match = s.join([cookie, match])
 479         if match_input:
 480             match = s.join([match, match_input])
 481         if cmd == 'add-flow':
 482             action = 'action=%s' % switch_outport_nr
 483             if vlan != None:
 484                 if path.index(current_hop) == 0:  # first node
 485                     action = ('action=mod_vlan_vid:%s' % vlan) + (',output=%s' % switch_outport_nr)
 486                     match = '-O OpenFlow13 ' + match
 487                 elif path.index(current_hop) == len(path) - 1:  # last node
 488                     match += ',dl_vlan=%s' % vlan
 489                     action = 'action=strip_vlan,output=%s' % switch_outport_nr
 490                 else:  # middle nodes
 491                     match += ',dl_vlan=%s' % vlan
 492             ofcmd = s.join([match, action])
 493         elif cmd == 'del-flows':
 494             ofcmd = match
 495         else:
 496             ofcmd = ''
 497
 498         node.dpctl(cmd, ofcmd)
 499         LOG.info("{3} in switch: {0} in_port: {1} out_port: {2}".format(node.name, switch_inport_nr,
 500                                                                                  switch_outport_nr, cmd))
 501
 502     # start Ryu Openflow controller as Remote Controller for the DCNetwork
 503     def startRyu(self, learning_switch=True):
 504         # start Ryu controller with rest-API
 505         python_install_path = site.getsitepackages()[0]
 506         ryu_path = python_install_path + '/ryu/app/simple_switch_13.py'
 507         ryu_path2 = python_install_path + '/ryu/app/ofctl_rest.py'
 508         # change the default Openflow controller port to 6653 (official IANA-assigned port number), as used by Mininet
 509         # Ryu still uses 6633 as default
 510         ryu_option = '--ofp-tcp-listen-port'
 511         ryu_of_port = '6653'
 512         ryu_cmd = 'ryu-manager'
 513         FNULL = open("/tmp/ryu.log", 'w')
 514         if learning_switch:
 515             self.ryu_process = Popen([ryu_cmd, ryu_path, ryu_path2, ryu_option, ryu_of_port], stdout=FNULL, stderr=FNULL)
 516         else:
 517             # no learning switch, but with rest api
 518             self.ryu_process = Popen([ryu_cmd, ryu_path2, ryu_option, ryu_of_port], stdout=FNULL, stderr=FNULL)
 519         time.sleep(1)
 520
 521     def killRyu(self):
 522         """
 523         Stop the Ryu controller that might be started by son-emu.
 524         :return:
 525         """
 526         # try it nicely
 527         if self.ryu_process is not None:
 528             self.ryu_process.terminate()
 529             self.ryu_process.kill()
 530         # ensure its death ;-)
 531         Popen(['pkill', '-f', 'ryu-manager'])
 532
 533     def ryu_REST(self, prefix, dpid=None, data=None):
 534         try:
 535             if dpid:
 536                 url = self.ryu_REST_api + '/' + str(prefix) + '/' + str(dpid)
 537             else:
 538                 url = self.ryu_REST_api + '/' + str(prefix)
 539             if data:
 540                 #LOG.info('POST: {0}'.format(str(data)))
 541                 req = urllib2.Request(url, str(data))
 542             else:
 543                 req = urllib2.Request(url)
 544
 545             ret = urllib2.urlopen(req).read()
 546             return ret
 547         except:
 548             LOG.info('error url: {0}'.format(str(url)))
 549             if data: LOG.info('error POST: {0}'.format(str(data)))
 550
 551     # need to respect that some match fields must be integers
 552     # http://ryu.readthedocs.io/en/latest/app/ofctl_rest.html#description-of-match-and-actions
 553     def _parse_match(self, match):
 554         matches = match.split(',')
 555         dict = {}
 556         for m in matches:
 557             match = m.split('=')
 558             if len(match) == 2:
 559                 try:
 560                     m2 = int(match[1], 0)
 561                 except:
 562                     m2 = match[1]
 563
 564                 dict.update({match[0]:m2})
 565         return dict
 566