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.info("No path could be found between {0} and {1}".format(vnf_src_name, vnf_dst_name))
 318             return "No path could be found between {0} and {1}".format(vnf_src_name, vnf_dst_name)
 319
 320         logging.info("Path between {0} and {1}: {2}".format(vnf_src_name, vnf_dst_name, path))
 321
 322         current_hop = src_sw
 323         switch_inport_nr = src_sw_inport_nr
 324
 325         # choose free vlan if path contains more than 1 switch
 326         cmd = kwargs.get('cmd')
 327         vlan = None
 328         if cmd == 'add-flow':
 329             if len(path) > 1:
 330                 vlan = self.vlans.pop()
 331
 332         for i in range(0,len(path)):
 333             current_node = self.getNodeByName(current_hop)
 334
 335             if path.index(current_hop) < len(path)-1:
 336                 next_hop = path[path.index(current_hop)+1]
 337             else:
 338                 #last switch reached
 339                 next_hop = vnf_dst_name
 340
 341             next_node = self.getNodeByName(next_hop)
 342
 343             if next_hop == vnf_dst_name:
 344                 switch_outport_nr = dst_sw_outport_nr
 345                 logging.info("end node reached: {0}".format(vnf_dst_name))
 346             elif not isinstance( next_node, OVSSwitch ):
 347                 logging.info("Next node: {0} is not a switch".format(next_hop))
 348                 return "Next node: {0} is not a switch".format(next_hop)
 349             else:
 350                 # take first link between switches by default
 351                 index_edge_out = 0
 352                 switch_outport_nr = self.DCNetwork_graph[current_hop][next_hop][index_edge_out]['src_port_nr']
 353
 354
 355            # set of entry via ovs-ofctl
 356             if isinstance( current_node, OVSSwitch ):
 357                 kwargs['vlan'] = vlan
 358                 kwargs['path'] = path
 359                 kwargs['current_hop'] = current_hop
 360
 361                 if self.controller == RemoteController:
 362                     ## set flow entry via ryu rest api
 363                     self._set_flow_entry_ryu_rest(current_node, switch_inport_nr, switch_outport_nr, **kwargs)
 364                 else:
 365                     ## set flow entry via ovs-ofctl
 366                     self._set_flow_entry_dpctl(current_node, switch_inport_nr, switch_outport_nr, **kwargs)
 367
 368
 369
 370             # take first link between switches by default
 371             if isinstance( next_node, OVSSwitch ):
 372                 switch_inport_nr = self.DCNetwork_graph[current_hop][next_hop][0]['dst_port_nr']
 373                 current_hop = next_hop
 374
 375         return "path {2} between {0} and {1}".format(vnf_src_name, vnf_dst_name, cmd)
 376
 377     def _set_flow_entry_ryu_rest(self, node, switch_inport_nr, switch_outport_nr, **kwargs):
 378         match = 'in_port=%s' % switch_inport_nr
 379
 380         cookie = kwargs.get('cookie')
 381         match_input = kwargs.get('match')
 382         cmd = kwargs.get('cmd')
 383         path = kwargs.get('path')
 384         current_hop = kwargs.get('current_hop')
 385         vlan = kwargs.get('vlan')
 386
 387         s = ','
 388         if match_input:
 389             match = s.join([match, match_input])
 390
 391         flow = {}
 392         flow['dpid'] = int(node.dpid, 16)
 393
 394         if cookie:
 395             flow['cookie'] = int(cookie)
 396
 397
 398         flow['actions'] = []
 399
 400         # possible Ryu actions, match fields:
 401         # http://ryu.readthedocs.io/en/latest/app/ofctl_rest.html#add-a-flow-entry
 402         if cmd == 'add-flow':
 403             prefix = 'stats/flowentry/add'
 404             if vlan != None:
 405                 if path.index(current_hop) == 0:  # first node
 406                     action = {}
 407                     action['type'] = 'PUSH_VLAN'  # Push a new VLAN tag if a input frame is non-VLAN-tagged
 408                     action['ethertype'] = 33024   # Ethertype 0x8100(=33024): IEEE 802.1Q VLAN-tagged frame
 409                     flow['actions'].append(action)
 410                     action = {}
 411                     action['type'] = 'SET_FIELD'
 412                     action['field'] = 'vlan_vid'
 413                     action['value'] = vlan
 414                     flow['actions'].append(action)
 415                 elif path.index(current_hop) == len(path) - 1:  # last node
 416                     match += ',dl_vlan=%s' % vlan
 417                     action = {}
 418                     action['type'] = 'POP_VLAN'
 419                     flow['actions'].append(action)
 420                 else:  # middle nodes
 421                     match += ',dl_vlan=%s' % vlan
 422             # output action must come last
 423             action = {}
 424             action['type'] = 'OUTPUT'
 425             action['port'] = switch_outport_nr
 426             flow['actions'].append(action)
 427
 428         elif cmd == 'del-flows':
 429             prefix = 'stats/flowentry/delete'
 430
 431             if cookie:
 432                 # TODO: add cookie_mask as argument
 433                 flow['cookie_mask'] = int('0xffffffffffffffff', 16)  # need full mask to match complete cookie
 434
 435             action = {}
 436             action['type'] = 'OUTPUT'
 437             action['port'] = switch_outport_nr
 438             flow['actions'].append(action)
 439
 440         flow['match'] = self._parse_match(match)
 441         self.ryu_REST(prefix, data=flow)
 442
 443     def _set_flow_entry_dpctl(self, node, switch_inport_nr, switch_outport_nr, **kwargs):
 444         match = 'in_port=%s' % switch_inport_nr
 445
 446         cookie = kwargs.get('cookie')
 447         match_input = kwargs.get('match')
 448         cmd = kwargs.get('cmd')
 449         path = kwargs.get('path')
 450         current_hop = kwargs.get('current_hop')
 451         vlan = kwargs.get('vlan')
 452
 453         s = ','
 454         if cookie:
 455             cookie = 'cookie=%s' % cookie
 456             match = s.join([cookie, match])
 457         if match_input:
 458             match = s.join([match, match_input])
 459         if cmd == 'add-flow':
 460             action = 'action=%s' % switch_outport_nr
 461             if vlan != None:
 462                 if path.index(current_hop) == 0:  # first node
 463                     action = ('action=mod_vlan_vid:%s' % vlan) + (',output=%s' % switch_outport_nr)
 464                     match = '-O OpenFlow13 ' + match
 465                 elif path.index(current_hop) == len(path) - 1:  # last node
 466                     match += ',dl_vlan=%s' % vlan
 467                     action = 'action=strip_vlan,output=%s' % switch_outport_nr
 468                 else:  # middle nodes
 469                     match += ',dl_vlan=%s' % vlan
 470             ofcmd = s.join([match, action])
 471         elif cmd == 'del-flows':
 472             ofcmd = match
 473         else:
 474             ofcmd = ''
 475
 476         node.dpctl(cmd, ofcmd)
 477         logging.info("{3} in switch: {0} in_port: {1} out_port: {2}".format(node.name, switch_inport_nr,
 478                                                                                  switch_outport_nr, cmd))
 479
 480     # start Ryu Openflow controller as Remote Controller for the DCNetwork
 481     def startRyu(self, learning_switch=True):
 482         # start Ryu controller with rest-API
 483         python_install_path = site.getsitepackages()[0]
 484         ryu_path = python_install_path + '/ryu/app/simple_switch_13.py'
 485         ryu_path2 = python_install_path + '/ryu/app/ofctl_rest.py'
 486         # change the default Openflow controller port to 6653 (official IANA-assigned port number), as used by Mininet
 487         # Ryu still uses 6633 as default
 488         ryu_option = '--ofp-tcp-listen-port'
 489         ryu_of_port = '6653'
 490         ryu_cmd = 'ryu-manager'
 491         FNULL = open("/tmp/ryu.log", 'w')
 492         if learning_switch:
 493             self.ryu_process = Popen([ryu_cmd, ryu_path, ryu_path2, ryu_option, ryu_of_port], stdout=FNULL, stderr=FNULL)
 494         else:
 495             # no learning switch, but with rest api
 496             self.ryu_process = Popen([ryu_cmd, ryu_path2, ryu_option, ryu_of_port], stdout=FNULL, stderr=FNULL)
 497         time.sleep(1)
 498
 499     def stopRyu(self):
 500         if self.ryu_process is not None:
 501             self.ryu_process.terminate()
 502             self.ryu_process.kill()
 503
 504     def ryu_REST(self, prefix, dpid=None, data=None):
 505         try:
 506             if dpid:
 507                 url = self.ryu_REST_api + '/' + str(prefix) + '/' + str(dpid)
 508             else:
 509                 url = self.ryu_REST_api + '/' + str(prefix)
 510             if data:
 511                 #logging.info('POST: {0}'.format(str(data)))
 512                 req = urllib2.Request(url, str(data))
 513             else:
 514                 req = urllib2.Request(url)
 515
 516             ret = urllib2.urlopen(req).read()
 517             return ret
 518         except:
 519             logging.info('error url: {0}'.format(str(url)))
 520             if data: logging.info('error POST: {0}'.format(str(data)))
 521
 522     # need to respect that some match fields must be integers
 523     # http://ryu.readthedocs.io/en/latest/app/ofctl_rest.html#description-of-match-and-actions
 524     def _parse_match(self, match):
 525         matches = match.split(',')
 526         dict = {}
 527         for m in matches:
 528             match = m.split('=')
 529             if len(match) == 2:
 530                 try:
 531                     m2 = int(match[1], 0)
 532                 except:
 533                     m2 = match[1]
 534
 535                 dict.update({match[0]:m2})
 536         return dict
 537