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