5e5d0de1f36cebdf92e41368111c88fdf22ea275
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).
29 Distributed Cloud Emulator (dcemulator)
30 (c) 2015 by Manuel Peuster <manuel.peuster@upb.de>
42 logging
.basicConfig(level
=logging
.INFO
)
45 class ZeroRpcApiEndpoint(object):
47 Simple API endpoint that offers a zerorpc-based
48 interface. This interface will be used by the
49 default command line client.
50 It can be used as a reference to implement
51 REST interfaces providing the same semantics,
52 like e.g. OpenStack compute API.
55 def __init__(self
, listenip
, port
):
59 logging
.debug("Created API endpoint %s(%s:%d)" % (
60 self
.__class
__.__name
__, self
.ip
, self
.port
))
62 def connectDatacenter(self
, dc
):
63 self
.dcs
[dc
.label
] = dc
64 logging
.info("Connected DC(%s) to API endpoint %s(%s:%d)" % (
65 dc
.label
, self
.__class
__.__name
__, self
.ip
, self
.port
))
68 thread
= threading
.Thread(target
=self
._api
_server
_thread
, args
=())
71 logging
.debug("Started API endpoint %s(%s:%d)" % (
72 self
.__class
__.__name
__, self
.ip
, self
.port
))
74 def _api_server_thread(self
):
75 s
= zerorpc
.Server(MultiDatacenterApi(self
.dcs
))
76 s
.bind("tcp://%s:%d" % (self
.ip
, self
.port
))
80 class MultiDatacenterApi(object):
82 Just pass through the corresponding request to the
83 selected data center. Do not implement provisioning
84 logic here because will will have multiple API
85 endpoint implementations at the end.
88 def __init__(self
, dcs
):
91 def compute_action_start(self
, dc_label
, compute_name
, image
, network
, command
):
93 Start a new compute instance: A docker container (note: zerorpc does not support keyword arguments)
94 :param dc_label: name of the DC
95 :param compute_name: compute container name
96 :param image: image name
97 :param command: command to execute
98 :param network: list of all interface of the vnf, with their parameters (id=id1,ip=x.x.x.x/x),...
99 :return: networks list({"id":"input","ip": "10.0.0.254/8"}, {"id":"output","ip": "11.0.0.254/24"})
101 # TODO what to return UUID / given name / internal name ?
102 logging
.debug("RPC CALL: compute start")
104 c
= self
.dcs
.get(dc_label
).startCompute(
105 compute_name
, image
=image
, command
=command
, network
=network
)
107 # return docker inspect dict
109 except Exception as ex
:
110 logging
.exception("RPC error.")
113 def compute_action_stop(self
, dc_label
, compute_name
):
114 logging
.debug("RPC CALL: compute stop")
116 return self
.dcs
.get(dc_label
).stopCompute(compute_name
)
117 except Exception as ex
:
118 logging
.exception("RPC error.")
121 def compute_list(self
, dc_label
):
122 logging
.debug("RPC CALL: compute list")
125 # return list with all compute nodes in all DCs
127 for dc
in self
.dcs
.itervalues():
128 all_containers
+= dc
.listCompute()
129 return [(c
.name
, c
.getStatus())
130 for c
in all_containers
]
132 # return list of compute nodes for specified DC
133 return [(c
.name
, c
.getStatus())
134 for c
in self
.dcs
.get(dc_label
).listCompute()]
135 except Exception as ex
:
136 logging
.exception("RPC error.")
139 def compute_status(self
, dc_label
, compute_name
):
140 logging
.debug("RPC CALL: compute status")
143 dc_label
).containers
.get(compute_name
).getStatus()
144 except Exception as ex
:
145 logging
.exception("RPC error.")
149 def compute_profile(self
, dc_label
, compute_name
, kwargs
):
150 # note: zerorpc does not support keyword arguments
152 ## VIM/dummy gatekeeper's tasks:
154 vnf_status
= self
.compute_action_start( dc_label
, compute_name
,
156 kwargs
.get('network'),
157 kwargs
.get('command'))
158 # start traffic source (with fixed ip addres, no use for now...)
159 psrc_status
= self
.compute_action_start( dc_label
, 'psrc', 'profile_source', [{'id':'output'}], None)
160 # start traffic sink (with fixed ip addres)
161 psink_status
= self
.compute_action_start(dc_label
, 'psink', 'profile_sink', [{'id': 'input'}], None)
162 # link vnf to traffic source
163 DCNetwork
= self
.dcs
.get(dc_label
).net
164 DCNetwork
.setChain('psrc', compute_name
,
165 vnf_src_interface
='output',
166 vnf_dst_interface
=kwargs
.get('input'),
167 cmd
='add-flow', weight
=None, bidirectional
=True)
168 DCNetwork
.setChain('psrc', compute_name
,
169 vnf_src_interface
='output',
170 vnf_dst_interface
=kwargs
.get('input'),
171 cmd
='add-flow', weight
=None,
172 match
='dl_type=0x0800,nw_proto=17,udp_dst=5001',
174 DCNetwork
.setChain( compute_name
, 'psink',
175 vnf_src_interface
='output',
176 vnf_dst_interface
=kwargs
.get('input'),
177 cmd
='add-flow', weight
=None, bidirectional
=True)
178 DCNetwork
.setChain(compute_name
, 'psink',
179 vnf_src_interface
='output',
180 vnf_dst_interface
=kwargs
.get('input'),
181 cmd
='add-flow', weight
=None,
182 match
='dl_type=0x0800,nw_proto=17,udp_dst=5001',
186 # start traffic generation
188 for nw in psrc_status.get('network'):
189 if nw.get('intf_name') == 'output':
190 psrc_output_ip = unicode(nw['ip'])
192 dummy_iperf_server_ip = ipaddress.IPv4Address(psrc_output_ip) + 1
194 for nw
in psink_status
.get('network'):
195 if nw
.get('intf_name') == 'input':
196 psink_input_ip
= nw
['ip']
200 # get monitor data and analyze
201 vnf_uuid
= vnf_status
['id']
202 psrc_mgmt_ip
= psrc_status
['docker_network']
206 #need to wait a bit before containers are fully up?
210 for rate
in [0, 1, 2, 3]:
211 #logging.info('query:{0}'.format(query_cpu))
213 output_line
= DCNetwork
.monitor_agent
.profile(psrc_mgmt_ip
, rate
, psink_input_ip
, vnf_uuid
)
222 ## VIM/dummy gatekeeper's tasks:
223 # remove vnfs and chain
224 DCNetwork
.setChain('psrc', compute_name
,
225 vnf_src_interface
='output',
226 vnf_dst_interface
=kwargs
.get('input'),
227 cmd
='del-flows', weight
=None, bidirectional
=True)
228 DCNetwork
.setChain('psrc', compute_name
,
229 vnf_src_interface
='output',
230 vnf_dst_interface
=kwargs
.get('input'),
231 cmd
='del-flows', weight
=None,
232 match
='dl_type=0x0800,nw_proto=17,udp_dst=5001',
234 DCNetwork
.setChain(compute_name
, 'psink',
235 vnf_src_interface
='output',
236 vnf_dst_interface
=kwargs
.get('input'),
237 cmd
='del-flows', weight
=None, bidirectional
=True)
238 DCNetwork
.setChain(compute_name
, 'psink',
239 vnf_src_interface
='output',
240 vnf_dst_interface
=kwargs
.get('input'),
241 cmd
='del-flows', weight
=None,
242 match
='dl_type=0x0800,nw_proto=17,udp_dst=5001',
244 self
.compute_action_stop(dc_label
, compute_name
)
245 self
.compute_action_stop(dc_label
, 'psink')
246 self
.compute_action_stop(dc_label
, 'psrc')
250 def datacenter_list(self
):
251 logging
.debug("RPC CALL: datacenter list")
253 return [d
.getStatus() for d
in self
.dcs
.itervalues()]
254 except Exception as ex
:
255 logging
.exception("RPC error.")
258 def datacenter_status(self
, dc_label
):
259 logging
.debug("RPC CALL: datacenter status")
261 return self
.dcs
.get(dc_label
).getStatus()
262 except Exception as ex
:
263 logging
.exception("RPC error.")