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 mininet
.net
import Containernet
37 from mininet
.node
import Controller
, DefaultController
, OVSSwitch
, OVSKernelSwitch
, Docker
, RemoteController
38 from mininet
.cli
import CLI
39 from mininet
.link
import TCLink
40 from mininet
.clean
import cleanup
42 from emuvim
.dcemulator
.monitoring
import DCNetworkMonitor
43 from emuvim
.dcemulator
.node
import Datacenter
, EmulatorCompute
44 from emuvim
.dcemulator
.resourcemodel
import ResourceModelRegistrar
46 LOG
= logging
.getLogger("dcemulator.net")
47 LOG
.setLevel(logging
.DEBUG
)
49 class DCNetwork(Containernet
):
51 Wraps the original Mininet/Containernet class and provides
52 methods to add data centers, switches, etc.
54 This class is used by topology definition scripts.
57 def __init__(self
, controller
=RemoteController
, monitor
=False,
58 enable_learning
= True, # in case of RemoteController (Ryu), learning switch behavior can be turned off/on
59 dc_emulation_max_cpu
=1.0, # fraction of overall CPU time for emulation
60 dc_emulation_max_mem
=512, # emulation max mem in MB
63 Create an extended version of a Containernet network
64 :param dc_emulation_max_cpu: max. CPU time used by containers in data centers
65 :param kwargs: path through for Mininet parameters
70 self
.ryu_process
= None
72 # always cleanup environment before we start the emulator
76 # call original Docker.__init__ and setup default controller
77 Containernet
.__init
__(
78 self
, switch
=OVSKernelSwitch
, controller
=controller
, **kwargs
)
81 if controller
== RemoteController
:
82 # start Ryu controller
83 self
.startRyu(learning_switch
=enable_learning
)
85 # add the specified controller
86 self
.addController('c0', controller
=controller
)
88 # graph of the complete DC network
89 self
.DCNetwork_graph
= nx
.MultiDiGraph()
91 # initialize pool of vlan tags to setup the SDN paths
92 self
.vlans
= range(4096)[::-1]
94 # link to Ryu REST_API
97 self
.ryu_REST_api
= 'http://{0}:{1}'.format(ryu_ip
, ryu_port
)
98 self
.RyuSession
= requests
.Session()
102 self
.monitor_agent
= DCNetworkMonitor(self
)
104 self
.monitor_agent
= None
106 # initialize resource model registrar
107 self
.rm_registrar
= ResourceModelRegistrar(
108 dc_emulation_max_cpu
, dc_emulation_max_mem
)
110 def addDatacenter(self
, label
, metadata
={}, resource_log_path
=None):
112 Create and add a logical cloud data center to the network.
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
119 dc
.create() # finally create the data center in our Mininet instance
120 LOG
.info("added data center: %s" % label
)
123 def addLink(self
, node1
, node2
, **params
):
125 Able to handle Datacenter objects as link
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
):
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
):
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
160 link
= Containernet
.addLink(self
, node1
, node2
, **params
)
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
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
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
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
])
188 attr_number
= match
.group(1)
191 attr_dict
[attr
] = attr_number
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
)
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
)
210 def addDocker( self
, label
, **params
):
212 Wrapper for addDocker method to use custom container class.
214 self
.DCNetwork_graph
.add_node(label
)
215 return Containernet
.addDocker(self
, label
, cls
=EmulatorCompute
, **params
)
217 def removeDocker( self
, label
, **params
):
219 Wrapper for removeDocker method to update graph.
221 self
.DCNetwork_graph
.remove_node(label
)
222 return Containernet
.removeDocker(self
, label
, **params
)
224 def addSwitch( self
, name
, add_to_graph
=True, **params
):
226 Wrapper for addSwitch method to store switch also in graph.
229 self
.DCNetwork_graph
.add_node(name
)
230 return Containernet
.addSwitch(self
, name
, protocols
='OpenFlow10,OpenFlow12,OpenFlow13', **params
)
232 def getAllContainers(self
):
234 Returns a list with all containers within all data centers.
237 for dc
in self
.dcs
.itervalues():
238 all_containers
+= dc
.listCompute()
239 return all_containers
243 for dc
in self
.dcs
.itervalues():
245 Containernet
.start(self
)
249 # stop the monitor agent
250 if self
.monitor_agent
is not None:
251 self
.monitor_agent
.stop()
254 Containernet
.stop(self
)
256 # stop Ryu controller
263 def setChain(self
, vnf_src_name
, vnf_dst_name
, vnf_src_interface
=None, vnf_dst_interface
=None, **kwargs
):
265 Chain 2 vnf interfaces together by installing the flowrules in the switches along their path.
266 Currently the path is found using the default networkx shortest path function.
267 Each chain gets a unique vlan id , so different chains wil not interfere.
269 :param vnf_src_name: vnf name (string)
270 :param vnf_dst_name: vnf name (string)
271 :param vnf_src_interface: source interface name (string)
272 :param vnf_dst_interface: destination interface name (string)
273 :param cmd: 'add-flow' (default) to add a chain, 'del-flows' to remove a chain
274 :param cookie: cookie for the installed flowrules (can be used later as identifier for a set of installed chains)
275 :param match: custom match entry to be added to the flowrules (default: only in_port and vlan tag)
276 :param priority: custom flowrule priority
277 :return: output log string
279 cmd
= kwargs
.get('cmd')
280 if cmd
== 'add-flow':
281 ret
= self
._chainAddFlow
(vnf_src_name
, vnf_dst_name
, vnf_src_interface
, vnf_dst_interface
, **kwargs
)
282 if kwargs
.get('bidirectional'):
283 ret
= ret
+'\n' + self
._chainAddFlow
(vnf_dst_name
, vnf_src_name
, vnf_dst_interface
, vnf_src_interface
, **kwargs
)
285 elif cmd
== 'del-flows':
286 ret
= self
._chainAddFlow
(vnf_src_name
, vnf_dst_name
, vnf_src_interface
, vnf_dst_interface
, **kwargs
)
287 if kwargs
.get('bidirectional'):
288 ret
= ret
+ '\n' + self
._chainAddFlow
(vnf_dst_name
, vnf_src_name
, vnf_dst_interface
, vnf_src_interface
, **kwargs
)
291 ret
= "Command unknown"
296 def _chainAddFlow(self
, vnf_src_name
, vnf_dst_name
, vnf_src_interface
=None, vnf_dst_interface
=None, **kwargs
):
301 dst_sw_outport_nr
= 0
303 LOG
.debug("call chainAddFlow vnf_src_name=%r, vnf_src_interface=%r, vnf_dst_name=%r, vnf_dst_interface=%r",
304 vnf_src_name
, vnf_src_interface
, vnf_dst_name
, vnf_dst_interface
)
306 #check if port is specified (vnf:port)
307 if vnf_src_interface
is None:
308 # take first interface by default
309 connected_sw
= self
.DCNetwork_graph
.neighbors(vnf_src_name
)[0]
310 link_dict
= self
.DCNetwork_graph
[vnf_src_name
][connected_sw
]
311 vnf_src_interface
= link_dict
[0]['src_port_id']
313 for connected_sw
in self
.DCNetwork_graph
.neighbors(vnf_src_name
):
314 link_dict
= self
.DCNetwork_graph
[vnf_src_name
][connected_sw
]
315 for link
in link_dict
:
316 if (link_dict
[link
]['src_port_id'] == vnf_src_interface
or
317 link_dict
[link
]['src_port_name'] == vnf_src_interface
): # Fix: we might also get interface names, e.g, from a son-emu-cli call
318 # found the right link and connected switch
319 src_sw
= connected_sw
320 src_sw_inport_nr
= link_dict
[link
]['dst_port_nr']
323 if vnf_dst_interface
is None:
324 # take first interface by default
325 connected_sw
= self
.DCNetwork_graph
.neighbors(vnf_dst_name
)[0]
326 link_dict
= self
.DCNetwork_graph
[connected_sw
][vnf_dst_name
]
327 vnf_dst_interface
= link_dict
[0]['dst_port_id']
329 vnf_dst_name
= vnf_dst_name
.split(':')[0]
330 for connected_sw
in self
.DCNetwork_graph
.neighbors(vnf_dst_name
):
331 link_dict
= self
.DCNetwork_graph
[connected_sw
][vnf_dst_name
]
332 for link
in link_dict
:
333 if link_dict
[link
]['dst_port_id'] == vnf_dst_interface
or \
334 link_dict
[link
]['dst_port_name'] == vnf_dst_interface
: # Fix: we might also get interface names, e.g, from a son-emu-cli call
335 # found the right link and connected switch
336 dst_sw
= connected_sw
337 dst_sw_outport_nr
= link_dict
[link
]['src_port_nr']
343 # returns the first found shortest path
344 # if all shortest paths are wanted, use: all_shortest_paths
345 path
= nx
.shortest_path(self
.DCNetwork_graph
, src_sw
, dst_sw
, weight
=kwargs
.get('weight'))
347 LOG
.exception("No path could be found between {0} and {1} using src_sw={2} and dst_sw={3}".format(
348 vnf_src_name
, vnf_dst_name
, src_sw
, dst_sw
))
349 LOG
.debug("Graph nodes: %r" % self
.DCNetwork_graph
.nodes())
350 LOG
.debug("Graph edges: %r" % self
.DCNetwork_graph
.edges())
351 for e
, v
in self
.DCNetwork_graph
.edges():
352 LOG
.debug("%r" % self
.DCNetwork_graph
[e
][v
])
353 return "No path could be found between {0} and {1}".format(vnf_src_name
, vnf_dst_name
)
355 LOG
.info("Path between {0} and {1}: {2}".format(vnf_src_name
, vnf_dst_name
, path
))
358 switch_inport_nr
= src_sw_inport_nr
360 # choose free vlan if path contains more than 1 switch
361 cmd
= kwargs
.get('cmd')
363 if cmd
== 'add-flow':
365 vlan
= self
.vlans
.pop()
367 for i
in range(0,len(path
)):
368 current_node
= self
.getNodeByName(current_hop
)
370 if path
.index(current_hop
) < len(path
)-1:
371 next_hop
= path
[path
.index(current_hop
)+1]
374 next_hop
= vnf_dst_name
376 next_node
= self
.getNodeByName(next_hop
)
378 if next_hop
== vnf_dst_name
:
379 switch_outport_nr
= dst_sw_outport_nr
380 LOG
.info("end node reached: {0}".format(vnf_dst_name
))
381 elif not isinstance( next_node
, OVSSwitch
):
382 LOG
.info("Next node: {0} is not a switch".format(next_hop
))
383 return "Next node: {0} is not a switch".format(next_hop
)
385 # take first link between switches by default
387 switch_outport_nr
= self
.DCNetwork_graph
[current_hop
][next_hop
][index_edge_out
]['src_port_nr']
390 # set of entry via ovs-ofctl
391 if isinstance( current_node
, OVSSwitch
):
392 kwargs
['vlan'] = vlan
393 kwargs
['path'] = path
394 kwargs
['current_hop'] = current_hop
396 if self
.controller
== RemoteController
:
397 ## set flow entry via ryu rest api
398 self
._set
_flow
_entry
_ryu
_rest
(current_node
, switch_inport_nr
, switch_outport_nr
, **kwargs
)
400 ## set flow entry via ovs-ofctl
401 self
._set
_flow
_entry
_dpctl
(current_node
, switch_inport_nr
, switch_outport_nr
, **kwargs
)
405 # take first link between switches by default
406 if isinstance( next_node
, OVSSwitch
):
407 switch_inport_nr
= self
.DCNetwork_graph
[current_hop
][next_hop
][0]['dst_port_nr']
408 current_hop
= next_hop
410 return "path {2} between {0} and {1}".format(vnf_src_name
, vnf_dst_name
, cmd
)
412 def _set_flow_entry_ryu_rest(self
, node
, switch_inport_nr
, switch_outport_nr
, **kwargs
):
413 match
= 'in_port=%s' % switch_inport_nr
415 cookie
= kwargs
.get('cookie')
416 match_input
= kwargs
.get('match')
417 cmd
= kwargs
.get('cmd')
418 path
= kwargs
.get('path')
419 current_hop
= kwargs
.get('current_hop')
420 vlan
= kwargs
.get('vlan')
421 priority
= kwargs
.get('priority')
425 match
= s
.join([match
, match_input
])
428 flow
['dpid'] = int(node
.dpid
, 16)
431 flow
['cookie'] = int(cookie
)
433 flow
['priority'] = int(priority
)
437 # possible Ryu actions, match fields:
438 # http://ryu.readthedocs.io/en/latest/app/ofctl_rest.html#add-a-flow-entry
439 if cmd
== 'add-flow':
440 prefix
= 'stats/flowentry/add'
442 if path
.index(current_hop
) == 0: # first node
444 action
['type'] = 'PUSH_VLAN' # Push a new VLAN tag if a input frame is non-VLAN-tagged
445 action
['ethertype'] = 33024 # Ethertype 0x8100(=33024): IEEE 802.1Q VLAN-tagged frame
446 flow
['actions'].append(action
)
448 action
['type'] = 'SET_FIELD'
449 action
['field'] = 'vlan_vid'
450 # ryu expects the field to be masked
451 action
['value'] = vlan |
0x1000
452 flow
['actions'].append(action
)
453 elif path
.index(current_hop
) == len(path
) - 1: # last node
454 match
+= ',dl_vlan=%s' % vlan
456 action
['type'] = 'POP_VLAN'
457 flow
['actions'].append(action
)
459 match
+= ',dl_vlan=%s' % vlan
460 # output action must come last
462 action
['type'] = 'OUTPUT'
463 action
['port'] = switch_outport_nr
464 flow
['actions'].append(action
)
466 elif cmd
== 'del-flows':
467 prefix
= 'stats/flowentry/delete'
470 # TODO: add cookie_mask as argument
471 flow
['cookie_mask'] = int('0xffffffffffffffff', 16) # need full mask to match complete cookie
474 action
['type'] = 'OUTPUT'
475 action
['port'] = switch_outport_nr
476 flow
['actions'].append(action
)
478 flow
['match'] = self
._parse
_match
(match
)
479 self
.ryu_REST(prefix
, data
=flow
)
481 def _set_flow_entry_dpctl(self
, node
, switch_inport_nr
, switch_outport_nr
, **kwargs
):
482 match
= 'in_port=%s' % switch_inport_nr
484 cookie
= kwargs
.get('cookie')
485 match_input
= kwargs
.get('match')
486 cmd
= kwargs
.get('cmd')
487 path
= kwargs
.get('path')
488 current_hop
= kwargs
.get('current_hop')
489 vlan
= kwargs
.get('vlan')
493 cookie
= 'cookie=%s' % cookie
494 match
= s
.join([cookie
, match
])
496 match
= s
.join([match
, match_input
])
497 if cmd
== 'add-flow':
498 action
= 'action=%s' % switch_outport_nr
500 if path
.index(current_hop
) == 0: # first node
501 action
= ('action=mod_vlan_vid:%s' % vlan
) + (',output=%s' % switch_outport_nr
)
502 match
= '-O OpenFlow13 ' + match
503 elif path
.index(current_hop
) == len(path
) - 1: # last node
504 match
+= ',dl_vlan=%s' % vlan
505 action
= 'action=strip_vlan,output=%s' % switch_outport_nr
507 match
+= ',dl_vlan=%s' % vlan
508 ofcmd
= s
.join([match
, action
])
509 elif cmd
== 'del-flows':
514 node
.dpctl(cmd
, ofcmd
)
515 LOG
.info("{3} in switch: {0} in_port: {1} out_port: {2}".format(node
.name
, switch_inport_nr
,
516 switch_outport_nr
, cmd
))
518 # start Ryu Openflow controller as Remote Controller for the DCNetwork
519 def startRyu(self
, learning_switch
=True):
520 # start Ryu controller with rest-API
521 python_install_path
= site
.getsitepackages()[0]
522 ryu_path
= python_install_path
+ '/ryu/app/simple_switch_13.py'
523 ryu_path2
= python_install_path
+ '/ryu/app/ofctl_rest.py'
524 # change the default Openflow controller port to 6653 (official IANA-assigned port number), as used by Mininet
525 # Ryu still uses 6633 as default
526 ryu_option
= '--ofp-tcp-listen-port'
528 ryu_cmd
= 'ryu-manager'
529 FNULL
= open("/tmp/ryu.log", 'w')
531 self
.ryu_process
= Popen([ryu_cmd
, ryu_path
, ryu_path2
, ryu_option
, ryu_of_port
], stdout
=FNULL
, stderr
=FNULL
)
533 # no learning switch, but with rest api
534 self
.ryu_process
= Popen([ryu_cmd
, ryu_path2
, ryu_option
, ryu_of_port
], stdout
=FNULL
, stderr
=FNULL
)
539 Stop the Ryu controller that might be started by son-emu.
543 if self
.ryu_process
is not None:
544 self
.ryu_process
.terminate()
545 self
.ryu_process
.kill()
546 # ensure its death ;-)
547 Popen(['pkill', '-f', 'ryu-manager'])
549 def ryu_REST(self
, prefix
, dpid
=None, data
=None):
552 url
= self
.ryu_REST_api
+ '/' + str(prefix
) + '/' + str(dpid
)
554 url
= self
.ryu_REST_api
+ '/' + str(prefix
)
556 req
= self
.RyuSession
.post(url
, json
=data
)
558 req
= self
.RyuSession
.get(url
)
561 # do extra logging if status code is not 200 (OK)
562 if req
.status_code
is not requests
.codes
.ok
:
564 'type {0} encoding: {1} text: {2} headers: {3} history: {4}'.format(req
.headers
['content-type'],
565 req
.encoding
, req
.text
,
566 req
.headers
, req
.history
))
567 LOG
.info('url: {0}'.format(str(url
)))
568 if data
: LOG
.info('POST: {0}'.format(str(data
)))
569 LOG
.info('status: {0} reason: {1}'.format(req
.status_code
, req
.reason
))
572 if 'json' in req
.headers
['content-type']:
576 ret
= req
.text
.rstrip()
580 # need to respect that some match fields must be integers
581 # http://ryu.readthedocs.io/en/latest/app/ofctl_rest.html#description-of-match-and-actions
582 def _parse_match(self
, match
):
583 matches
= match
.split(',')
589 m2
= int(match
[1], 0)
593 dict.update({match
[0]:m2
})