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