feature 5956. Split engine in several files
[osm/NBI.git] / osm_nbi / instance_topics.py
1 # -*- coding: utf-8 -*-
2
3 # import logging
4 from uuid import uuid4
5 from http import HTTPStatus
6 from time import time
7 from copy import copy
8 from validation import validate_input, ValidationError, ns_instantiate, ns_action, ns_scale
9 from base_topic import BaseTopic, EngineException, get_iterable
10 from descriptor_topics import DescriptorTopic
11
12 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
13
14
15 class NsrTopic(BaseTopic):
16 topic = "nsrs"
17 topic_msg = "ns"
18
19 def __init__(self, db, fs, msg):
20 BaseTopic.__init__(self, db, fs, msg)
21
22 def _check_descriptor_dependencies(self, session, descriptor):
23 """
24 Check that the dependent descriptors exist on a new descriptor or edition
25 :param session: client session information
26 :param descriptor: descriptor to be inserted or edit
27 :return: None or raises exception
28 """
29 if not descriptor.get("nsdId"):
30 return
31 nsd_id = descriptor["nsdId"]
32 if not self.get_item_list(session, "nsds", {"id": nsd_id}):
33 raise EngineException("Descriptor error at nsdId='{}' references a non exist nsd".format(nsd_id),
34 http_code=HTTPStatus.CONFLICT)
35
36 @staticmethod
37 def format_on_new(content, project_id=None, make_public=False):
38 BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
39 content["_admin"]["nsState"] = "NOT_INSTANTIATED"
40
41 def check_conflict_on_del(self, session, _id, force=False):
42 if force:
43 return
44 nsr = self.db.get_one("nsrs", {"_id": _id})
45 if nsr["_admin"].get("nsState") == "INSTANTIATED":
46 raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. "
47 "Launch 'terminate' operation first; or force deletion".format(_id),
48 http_code=HTTPStatus.CONFLICT)
49
50 def delete(self, session, _id, force=False, dry_run=False):
51 """
52 Delete item by its internal _id
53 :param session: contains the used login username, working project, and admin rights
54 :param _id: server internal id
55 :param force: indicates if deletion must be forced in case of conflict
56 :param dry_run: make checking but do not delete
57 :return: dictionary with deleted item _id. It raises EngineException on error: not found, conflict, ...
58 """
59 # TODO add admin to filter, validate rights
60 BaseTopic.delete(self, session, _id, force, dry_run=True)
61 if dry_run:
62 return
63
64 v = self.db.del_one("nsrs", {"_id": _id})
65 self.db.del_list("nslcmops", {"nsInstanceId": _id})
66 self.db.del_list("vnfrs", {"nsr-id-ref": _id})
67 # set all used pdus as free
68 self.db.set_list("pdus", {"_admin.usage.nsr_id": _id},
69 {"_admin.usageSate": "NOT_IN_USE", "_admin.usage": None})
70 self._send_msg("deleted", {"_id": _id})
71 return v
72
73 def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False):
74 """
75 Creates a new nsr into database. It also creates needed vnfrs
76 :param rollback: list to append the created items at database in case a rollback must be done
77 :param session: contains the used login username and working project
78 :param indata: params to be used for the nsr
79 :param kwargs: used to override the indata descriptor
80 :param headers: http request headers
81 :param force: If True avoid some dependence checks
82 :param make_public: Make the created item public to all projects
83 :return: the _id of nsr descriptor created at database
84 """
85
86 try:
87 ns_request = self._remove_envelop(indata)
88 # Override descriptor with query string kwargs
89 self._update_input_with_kwargs(ns_request, kwargs)
90 self._validate_input_new(ns_request, force)
91
92 step = ""
93 # look for nsr
94 step = "getting nsd id='{}' from database".format(ns_request.get("nsdId"))
95 _filter = {"_id": ns_request["nsdId"]}
96 _filter.update(BaseTopic._get_project_filter(session, write=False, show_all=True))
97 nsd = self.db.get_one("nsds", _filter)
98
99 nsr_id = str(uuid4())
100 now = time()
101 step = "filling nsr from input data"
102 nsr_descriptor = {
103 "name": ns_request["nsName"],
104 "name-ref": ns_request["nsName"],
105 "short-name": ns_request["nsName"],
106 "admin-status": "ENABLED",
107 "nsd": nsd,
108 "datacenter": ns_request["vimAccountId"],
109 "resource-orchestrator": "osmopenmano",
110 "description": ns_request.get("nsDescription", ""),
111 "constituent-vnfr-ref": [],
112
113 "operational-status": "init", # typedef ns-operational-
114 "config-status": "init", # typedef config-states
115 "detailed-status": "scheduled",
116
117 "orchestration-progress": {},
118 # {"networks": {"active": 0, "total": 0}, "vms": {"active": 0, "total": 0}},
119
120 "crete-time": now,
121 "nsd-name-ref": nsd["name"],
122 "operational-events": [], # "id", "timestamp", "description", "event",
123 "nsd-ref": nsd["id"],
124 "instantiate_params": ns_request,
125 "ns-instance-config-ref": nsr_id,
126 "id": nsr_id,
127 "_id": nsr_id,
128 # "input-parameter": xpath, value,
129 "ssh-authorized-key": ns_request.get("key-pair-ref"),
130 }
131 ns_request["nsr_id"] = nsr_id
132
133 # Create VNFR
134 needed_vnfds = {}
135 for member_vnf in nsd["constituent-vnfd"]:
136 vnfd_id = member_vnf["vnfd-id-ref"]
137 step = "getting vnfd id='{}' constituent-vnfd='{}' from database".format(
138 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
139 if vnfd_id not in needed_vnfds:
140 # Obtain vnfd
141 vnfd = DescriptorTopic.get_one_by_id(self.db, session, "vnfds", vnfd_id)
142 vnfd.pop("_admin")
143 needed_vnfds[vnfd_id] = vnfd
144 else:
145 vnfd = needed_vnfds[vnfd_id]
146 step = "filling vnfr vnfd-id='{}' constituent-vnfd='{}'".format(
147 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
148 vnfr_id = str(uuid4())
149 vnfr_descriptor = {
150 "id": vnfr_id,
151 "_id": vnfr_id,
152 "nsr-id-ref": nsr_id,
153 "member-vnf-index-ref": member_vnf["member-vnf-index"],
154 "created-time": now,
155 # "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise
156 "vnfd-ref": vnfd_id,
157 "vnfd-id": vnfd["_id"], # not at OSM model, but useful
158 "vim-account-id": None,
159 "vdur": [],
160 "connection-point": [],
161 "ip-address": None, # mgmt-interface filled by LCM
162 }
163 for cp in vnfd.get("connection-point", ()):
164 vnf_cp = {
165 "name": cp["name"],
166 "connection-point-id": cp.get("id"),
167 "id": cp.get("id"),
168 # "ip-address", "mac-address" # filled by LCM
169 # vim-id # TODO it would be nice having a vim port id
170 }
171 vnfr_descriptor["connection-point"].append(vnf_cp)
172 for vdu in vnfd["vdu"]:
173 vdur_id = str(uuid4())
174 vdur = {
175 "id": vdur_id,
176 "vdu-id-ref": vdu["id"],
177 # TODO "name": "" Name of the VDU in the VIM
178 "ip-address": None, # mgmt-interface filled by LCM
179 # "vim-id", "flavor-id", "image-id", "management-ip" # filled by LCM
180 "internal-connection-point": [],
181 "interfaces": [],
182 }
183 # TODO volumes: name, volume-id
184 for icp in vdu.get("internal-connection-point", ()):
185 vdu_icp = {
186 "id": icp["id"],
187 "connection-point-id": icp["id"],
188 "name": icp.get("name"),
189 # "ip-address", "mac-address" # filled by LCM
190 # vim-id # TODO it would be nice having a vim port id
191 }
192 vdur["internal-connection-point"].append(vdu_icp)
193 for iface in vdu.get("interface", ()):
194 vdu_iface = {
195 "name": iface.get("name"),
196 # "ip-address", "mac-address" # filled by LCM
197 # vim-id # TODO it would be nice having a vim port id
198 }
199 vdur["interfaces"].append(vdu_iface)
200 vnfr_descriptor["vdur"].append(vdur)
201
202 step = "creating vnfr vnfd-id='{}' constituent-vnfd='{}' at database".format(
203 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
204
205 # add at database
206 BaseTopic.format_on_new(vnfr_descriptor, session["project_id"], make_public=make_public)
207 self.db.create("vnfrs", vnfr_descriptor)
208 rollback.append({"topic": "vnfrs", "_id": vnfr_id})
209 nsr_descriptor["constituent-vnfr-ref"].append(vnfr_id)
210
211 step = "creating nsr at database"
212 self.format_on_new(nsr_descriptor, session["project_id"], make_public=make_public)
213 self.db.create("nsrs", nsr_descriptor)
214 rollback.append({"topic": "nsrs", "_id": nsr_id})
215 return nsr_id
216 except Exception as e:
217 self.logger.exception("Exception {} at NsrTopic.new()".format(e), exc_info=True)
218 raise EngineException("Error {}: {}".format(step, e))
219 except ValidationError as e:
220 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
221
222 def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None):
223 raise EngineException("Method edit called directly", HTTPStatus.INTERNAL_SERVER_ERROR)
224
225
226 class VnfrTopic(BaseTopic):
227 topic = "vnfrs"
228 topic_msg = None
229
230 def __init__(self, db, fs, msg):
231 BaseTopic.__init__(self, db, fs, msg)
232
233 def delete(self, session, _id, force=False, dry_run=False):
234 raise EngineException("Method delete called directly", HTTPStatus.INTERNAL_SERVER_ERROR)
235
236 def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None):
237 raise EngineException("Method edit called directly", HTTPStatus.INTERNAL_SERVER_ERROR)
238
239 def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False):
240 # Not used because vnfrs are created and deleted by NsrTopic class directly
241 raise EngineException("Method new called directly", HTTPStatus.INTERNAL_SERVER_ERROR)
242
243
244 class NsLcmOpTopic(BaseTopic):
245 topic = "nslcmops"
246 topic_msg = "ns"
247 operation_schema = { # mapping between operation and jsonschema to validate
248 "instantiate": ns_instantiate,
249 "action": ns_action,
250 "scale": ns_scale,
251 "terminate": None,
252 }
253
254 def __init__(self, db, fs, msg):
255 BaseTopic.__init__(self, db, fs, msg)
256
257 def _validate_input_new(self, input, force=False):
258 """
259 Validates input user content for a new entry. It uses jsonschema for each type or operation.
260 :param input: user input content for the new topic
261 :param force: may be used for being more tolerant
262 :return: The same input content, or a changed version of it.
263 """
264 if self.schema_new:
265 validate_input(input, self.schema_new)
266 return input
267
268 def _check_ns_operation(self, session, nsr, operation, indata):
269 """
270 Check that user has enter right parameters for the operation
271 :param session:
272 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
273 :param indata: descriptor with the parameters of the operation
274 :return: None
275 """
276 vnfds = {}
277 vim_accounts = []
278 nsd = nsr["nsd"]
279
280 def check_valid_vnf_member_index(member_vnf_index):
281 # TODO change to vnfR
282 for vnf in nsd["constituent-vnfd"]:
283 if member_vnf_index == vnf["member-vnf-index"]:
284 vnfd_id = vnf["vnfd-id-ref"]
285 if vnfd_id not in vnfds:
286 vnfds[vnfd_id] = self.db.get_one("vnfds", {"id": vnfd_id})
287 return vnfds[vnfd_id]
288 else:
289 raise EngineException("Invalid parameter member_vnf_index='{}' is not one of the "
290 "nsd:constituent-vnfd".format(member_vnf_index))
291
292 def check_valid_vim_account(vim_account):
293 if vim_account in vim_accounts:
294 return
295 try:
296 # TODO add _get_project_filter
297 self.db.get_one("vim_accounts", {"_id": vim_account})
298 except Exception:
299 raise EngineException("Invalid vimAccountId='{}' not present".format(vim_account))
300 vim_accounts.append(vim_account)
301
302 if operation == "action":
303 # check vnf_member_index
304 if indata.get("vnf_member_index"):
305 indata["member_vnf_index"] = indata.pop("vnf_member_index") # for backward compatibility
306 if not indata.get("member_vnf_index"):
307 raise EngineException("Missing 'member_vnf_index' parameter")
308 vnfd = check_valid_vnf_member_index(indata["member_vnf_index"])
309 # check primitive
310 for config_primitive in get_iterable(vnfd.get("vnf-configuration", {}).get("config-primitive")):
311 if indata["primitive"] == config_primitive["name"]:
312 # check needed primitive_params are provided
313 if indata.get("primitive_params"):
314 in_primitive_params_copy = copy(indata["primitive_params"])
315 else:
316 in_primitive_params_copy = {}
317 for paramd in get_iterable(config_primitive.get("parameter")):
318 if paramd["name"] in in_primitive_params_copy:
319 del in_primitive_params_copy[paramd["name"]]
320 elif not paramd.get("default-value"):
321 raise EngineException("Needed parameter {} not provided for primitive '{}'".format(
322 paramd["name"], indata["primitive"]))
323 # check no extra primitive params are provided
324 if in_primitive_params_copy:
325 raise EngineException("parameter/s '{}' not present at vnfd for primitive '{}'".format(
326 list(in_primitive_params_copy.keys()), indata["primitive"]))
327 break
328 else:
329 raise EngineException("Invalid primitive '{}' is not present at vnfd".format(indata["primitive"]))
330 if operation == "scale":
331 vnfd = check_valid_vnf_member_index(indata["scaleVnfData"]["scaleByStepData"]["member-vnf-index"])
332 for scaling_group in get_iterable(vnfd.get("scaling-group-descriptor")):
333 if indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] == scaling_group["name"]:
334 break
335 else:
336 raise EngineException("Invalid scaleVnfData:scaleByStepData:scaling-group-descriptor '{}' is not "
337 "present at vnfd:scaling-group-descriptor".format(
338 indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]))
339 if operation == "instantiate":
340 # check vim_account
341 check_valid_vim_account(indata["vimAccountId"])
342 for in_vnf in get_iterable(indata.get("vnf")):
343 vnfd = check_valid_vnf_member_index(in_vnf["member-vnf-index"])
344 if in_vnf.get("vimAccountId"):
345 check_valid_vim_account(in_vnf["vimAccountId"])
346 for in_vdu in get_iterable(in_vnf.get("vdu")):
347 for vdud in get_iterable(vnfd.get("vdu")):
348 if vdud["id"] == in_vdu["id"]:
349 for volume in get_iterable(in_vdu.get("volume")):
350 for volumed in get_iterable(vdud.get("volumes")):
351 if volumed["name"] == volume["name"]:
352 break
353 else:
354 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu[id='{}']:"
355 "volume:name='{}' is not present at vnfd:vdu:volumes list".
356 format(in_vnf["member-vnf-index"], in_vdu["id"],
357 volume["name"]))
358 break
359 else:
360 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu:id='{}' is not "
361 "present at vnfd".format(in_vnf["member-vnf-index"], in_vdu["id"]))
362
363 for in_internal_vld in get_iterable(in_vnf.get("internal-vld")):
364 for internal_vldd in get_iterable(vnfd.get("internal-vld")):
365 if in_internal_vld["name"] == internal_vldd["name"] or \
366 in_internal_vld["name"] == internal_vldd["id"]:
367 break
368 else:
369 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:internal-vld:name='{}'"
370 " is not present at vnfd '{}'".format(in_vnf["member-vnf-index"],
371 in_internal_vld["name"],
372 vnfd["id"]))
373 for in_vld in get_iterable(indata.get("vld")):
374 for vldd in get_iterable(nsd.get("vld")):
375 if in_vld["name"] == vldd["name"] or in_vld["name"] == vldd["id"]:
376 break
377 else:
378 raise EngineException("Invalid parameter vld:name='{}' is not present at nsd:vld".format(
379 in_vld["name"]))
380
381 def _create_nslcmop(self, session, nsInstanceId, operation, params):
382 now = time()
383 _id = str(uuid4())
384 nslcmop = {
385 "id": _id,
386 "_id": _id,
387 "operationState": "PROCESSING", # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
388 "statusEnteredTime": now,
389 "nsInstanceId": nsInstanceId,
390 "lcmOperationType": operation,
391 "startTime": now,
392 "isAutomaticInvocation": False,
393 "operationParams": params,
394 "isCancelPending": False,
395 "links": {
396 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id,
397 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsInstanceId,
398 }
399 }
400 return nslcmop
401
402 def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False):
403 """
404 Performs a new operation over a ns
405 :param rollback: list to append created items at database in case a rollback must to be done
406 :param session: contains the used login username and working project
407 :param indata: descriptor with the parameters of the operation. It must contains among others
408 nsInstanceId: _id of the nsr to perform the operation
409 operation: it can be: instantiate, terminate, action, TODO: update, heal
410 :param kwargs: used to override the indata descriptor
411 :param headers: http request headers
412 :param force: If True avoid some dependence checks
413 :param make_public: Make the created item public to all projects
414 :return: id of the nslcmops
415 """
416 try:
417 # Override descriptor with query string kwargs
418 self._update_input_with_kwargs(indata, kwargs)
419 operation = indata["lcmOperationType"]
420 nsInstanceId = indata["nsInstanceId"]
421
422 validate_input(indata, self.operation_schema[operation])
423 # get ns from nsr_id
424 _filter = BaseTopic._get_project_filter(session, write=True, show_all=False)
425 _filter["_id"] = nsInstanceId
426 nsr = self.db.get_one("nsrs", _filter)
427
428 # initial checking
429 if not nsr["_admin"].get("nsState") or nsr["_admin"]["nsState"] == "NOT_INSTANTIATED":
430 if operation == "terminate" and indata.get("autoremove"):
431 # NSR must be deleted
432 return self.delete(session, nsInstanceId)
433 if operation != "instantiate":
434 raise EngineException("ns_instance '{}' cannot be '{}' because it is not instantiated".format(
435 nsInstanceId, operation), HTTPStatus.CONFLICT)
436 else:
437 if operation == "instantiate" and not indata.get("force"):
438 raise EngineException("ns_instance '{}' cannot be '{}' because it is already instantiated".format(
439 nsInstanceId, operation), HTTPStatus.CONFLICT)
440 self._check_ns_operation(session, nsr, operation, indata)
441 nslcmop_desc = self._create_nslcmop(session, nsInstanceId, operation, indata)
442 self.format_on_new(nslcmop_desc, session["project_id"], make_public=make_public)
443 _id = self.db.create("nslcmops", nslcmop_desc)
444 rollback.append({"topic": "nslcmops", "_id": _id})
445 self.msg.write("ns", operation, nslcmop_desc)
446 return _id
447 except ValidationError as e:
448 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
449 # except DbException as e:
450 # raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND)
451
452 def delete(self, session, _id, force=False, dry_run=False):
453 raise EngineException("Method delete called directly", HTTPStatus.INTERNAL_SERVER_ERROR)
454
455 def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None):
456 raise EngineException("Method edit called directly", HTTPStatus.INTERNAL_SERVER_ERROR)