2 Copyright (c) 2015 SONATA-NFV and Paderborn University
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
9 http://www.apache.org/licenses/LICENSE-2.0
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.
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
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).
32 from subprocess
import Popen
36 from functools
import partial
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
43 from emuvim
.dcemulator
.monitoring
import DCNetworkMonitor
44 from emuvim
.dcemulator
.node
import Datacenter
, EmulatorCompute
45 from emuvim
.dcemulator
.resourcemodel
import ResourceModelRegistrar
47 class DCNetwork(Containernet
):
49 Wraps the original Mininet/Containernet class and provides
50 methods to add data centers, switches, etc.
52 This class is used by topology definition scripts.
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
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
68 # call original Docker.__init__ and setup default controller
69 Containernet
.__init
__(
70 self
, switch
=OVSKernelSwitch
, controller
=controller
, **kwargs
)
74 self
.ryu_process
= None
75 if controller
== RemoteController
:
76 # start Ryu controller
77 self
.startRyu(learning_switch
=enable_learning
)
79 # add the specified controller
80 self
.addController('c0', controller
=controller
)
82 # graph of the complete DC network
83 self
.DCNetwork_graph
= nx
.MultiDiGraph()
85 # initialize pool of vlan tags to setup the SDN paths
86 self
.vlans
= range(4096)[::-1]
88 # link to Ryu REST_API
91 self
.ryu_REST_api
= 'http://{0}:{1}'.format(ryu_ip
, ryu_port
)
95 self
.monitor_agent
= DCNetworkMonitor(self
)
97 self
.monitor_agent
= None
99 # initialize resource model registrar
100 self
.rm_registrar
= ResourceModelRegistrar(
101 dc_emulation_max_cpu
, dc_emulation_max_mem
)
103 def addDatacenter(self
, label
, metadata
={}, resource_log_path
=None):
105 Create and add a logical cloud data center to the network.
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
112 dc
.create() # finally create the data center in our Mininet instance
113 logging
.info("added data center: %s" % label
)
116 def addLink(self
, node1
, node2
, **params
):
118 Able to handle Datacenter objects as link
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
):
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
):
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
153 link
= Containernet
.addLink(self
, node1
, node2
, **params
)
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
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
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
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
])
181 attr_number
= match
.group(1)
184 attr_dict
[attr
] = attr_number
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
)
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
)
203 def addDocker( self
, label
, **params
):
205 Wrapper for addDocker method to use custom container class.
207 self
.DCNetwork_graph
.add_node(label
)
208 return Containernet
.addDocker(self
, label
, cls
=EmulatorCompute
, **params
)
210 def removeDocker( self
, label
, **params
):
212 Wrapper for removeDocker method to update graph.
214 self
.DCNetwork_graph
.remove_node(label
)
215 return Containernet
.removeDocker(self
, label
, **params
)
217 def addSwitch( self
, name
, add_to_graph
=True, **params
):
219 Wrapper for addSwitch method to store switch also in graph.
222 self
.DCNetwork_graph
.add_node(name
)
223 return Containernet
.addSwitch(self
, name
, protocols
='OpenFlow10,OpenFlow12,OpenFlow13', **params
)
225 def getAllContainers(self
):
227 Returns a list with all containers within all data centers.
230 for dc
in self
.dcs
.itervalues():
231 all_containers
+= dc
.listCompute()
232 return all_containers
236 for dc
in self
.dcs
.itervalues():
238 Containernet
.start(self
)
242 # stop the monitor agent
243 if self
.monitor_agent
is not None:
244 self
.monitor_agent
.stop()
247 Containernet
.stop(self
)
249 # stop Ryu controller
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
)
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
)
270 ret
= "Command unknown"
275 def _chainAddFlow(self
, vnf_src_name
, vnf_dst_name
, vnf_src_interface
=None, vnf_dst_interface
=None, **kwargs
):
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']
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
292 src_sw_inport_nr
= link_dict
[link
]['dst_port_nr']
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']
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']
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'))
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
)
321 logging
.info("Path between {0} and {1}: {2}".format(vnf_src_name
, vnf_dst_name
, path
))
324 switch_inport_nr
= src_sw_inport_nr
326 # choose free vlan if path contains more than 1 switch
327 cmd
= kwargs
.get('cmd')
329 if cmd
== 'add-flow':
331 vlan
= self
.vlans
.pop()
333 for i
in range(0,len(path
)):
334 current_node
= self
.getNodeByName(current_hop
)
336 if path
.index(current_hop
) < len(path
)-1:
337 next_hop
= path
[path
.index(current_hop
)+1]
340 next_hop
= vnf_dst_name
342 next_node
= self
.getNodeByName(next_hop
)
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
)
351 # take first link between switches by default
353 switch_outport_nr
= self
.DCNetwork_graph
[current_hop
][next_hop
][index_edge_out
]['src_port_nr']
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
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
)
366 ## set flow entry via ovs-ofctl
367 self
._set
_flow
_entry
_dpctl
(current_node
, switch_inport_nr
, switch_outport_nr
, **kwargs
)
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
376 return "path {2} between {0} and {1}".format(vnf_src_name
, vnf_dst_name
, cmd
)
378 def _set_flow_entry_ryu_rest(self
, node
, switch_inport_nr
, switch_outport_nr
, **kwargs
):
379 match
= 'in_port=%s' % switch_inport_nr
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')
390 match
= s
.join([match
, match_input
])
393 flow
['dpid'] = int(node
.dpid
, 16)
396 flow
['cookie'] = int(cookie
)
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'
406 if path
.index(current_hop
) == 0: # first node
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
)
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
419 action
['type'] = 'POP_VLAN'
420 flow
['actions'].append(action
)
422 match
+= ',dl_vlan=%s' % vlan
423 # output action must come last
425 action
['type'] = 'OUTPUT'
426 action
['port'] = switch_outport_nr
427 flow
['actions'].append(action
)
429 elif cmd
== 'del-flows':
430 prefix
= 'stats/flowentry/delete'
433 # TODO: add cookie_mask as argument
434 flow
['cookie_mask'] = int('0xffffffffffffffff', 16) # need full mask to match complete cookie
437 action
['type'] = 'OUTPUT'
438 action
['port'] = switch_outport_nr
439 flow
['actions'].append(action
)
441 flow
['match'] = self
._parse
_match
(match
)
442 self
.ryu_REST(prefix
, data
=flow
)
444 def _set_flow_entry_dpctl(self
, node
, switch_inport_nr
, switch_outport_nr
, **kwargs
):
445 match
= 'in_port=%s' % switch_inport_nr
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')
456 cookie
= 'cookie=%s' % cookie
457 match
= s
.join([cookie
, match
])
459 match
= s
.join([match
, match_input
])
460 if cmd
== 'add-flow':
461 action
= 'action=%s' % switch_outport_nr
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
470 match
+= ',dl_vlan=%s' % vlan
471 ofcmd
= s
.join([match
, action
])
472 elif cmd
== 'del-flows':
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
))
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'
491 ryu_cmd
= 'ryu-manager'
492 FNULL
= open("/tmp/ryu.log", 'w')
494 self
.ryu_process
= Popen([ryu_cmd
, ryu_path
, ryu_path2
, ryu_option
, ryu_of_port
], stdout
=FNULL
, stderr
=FNULL
)
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
)
501 if self
.ryu_process
is not None:
502 self
.ryu_process
.terminate()
503 self
.ryu_process
.kill()
505 def ryu_REST(self
, prefix
, dpid
=None, data
=None):
508 url
= self
.ryu_REST_api
+ '/' + str(prefix
) + '/' + str(dpid
)
510 url
= self
.ryu_REST_api
+ '/' + str(prefix
)
512 #logging.info('POST: {0}'.format(str(data)))
513 req
= urllib2
.Request(url
, str(data
))
515 req
= urllib2
.Request(url
)
517 ret
= urllib2
.urlopen(req
).read()
520 logging
.info('error url: {0}'.format(str(url
)))
521 if data
: logging
.info('error POST: {0}'.format(str(data
)))
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(',')
532 m2
= int(match
[1], 0)
536 dict.update({match
[0]:m2
})