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