1 # -*- coding: utf-8 -*-
3 from osm_common
import dbmongo
4 from osm_common
import dbmemory
5 from osm_common
import fslocal
6 from osm_common
import msglocal
7 from osm_common
import msgkafka
12 from uuid
import uuid4
13 from hashlib
import sha256
, md5
14 from osm_common
.dbbase
import DbException
, deep_update
15 from osm_common
.fsbase
import FsException
16 from osm_common
.msgbase
import MsgException
17 from http
import HTTPStatus
20 from validation
import validate_input
, ValidationError
22 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
25 class EngineException(Exception):
27 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
28 self
.http_code
= http_code
29 Exception.__init
__(self
, message
)
32 def get_iterable(input_var
):
34 Returns an iterable, in case input_var is None it just returns an empty tuple
35 :param input_var: can be a list, tuple or None
36 :return: input_var or () if it is None
51 self
.logger
= logging
.getLogger("nbi.engine")
53 def start(self
, config
):
55 Connect to database, filesystem storage, and messaging
56 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
62 if config
["database"]["driver"] == "mongo":
63 self
.db
= dbmongo
.DbMongo()
64 self
.db
.db_connect(config
["database"])
65 elif config
["database"]["driver"] == "memory":
66 self
.db
= dbmemory
.DbMemory()
67 self
.db
.db_connect(config
["database"])
69 raise EngineException("Invalid configuration param '{}' at '[database]':'driver'".format(
70 config
["database"]["driver"]))
72 if config
["storage"]["driver"] == "local":
73 self
.fs
= fslocal
.FsLocal()
74 self
.fs
.fs_connect(config
["storage"])
76 raise EngineException("Invalid configuration param '{}' at '[storage]':'driver'".format(
77 config
["storage"]["driver"]))
79 if config
["message"]["driver"] == "local":
80 self
.msg
= msglocal
.MsgLocal()
81 self
.msg
.connect(config
["message"])
82 elif config
["message"]["driver"] == "kafka":
83 self
.msg
= msgkafka
.MsgKafka()
84 self
.msg
.connect(config
["message"])
86 raise EngineException("Invalid configuration param '{}' at '[message]':'driver'".format(
87 config
["storage"]["driver"]))
88 except (DbException
, FsException
, MsgException
) as e
:
89 raise EngineException(str(e
), http_code
=e
.http_code
)
94 self
.db
.db_disconnect()
96 self
.fs
.fs_disconnect()
98 self
.fs
.fs_disconnect()
99 except (DbException
, FsException
, MsgException
) as e
:
100 raise EngineException(str(e
), http_code
=e
.http_code
)
103 def _remove_envelop(item
, indata
=None):
105 Obtain the useful data removing the envelop. It goes throw the vnfd or nsd catalog and returns the
107 :param item: can be vnfds, nsds, users, projects, userDefinedData (initial content of a vnfds, nsds
108 :param indata: Content to be inspected
109 :return: the useful part of indata
111 clean_indata
= indata
115 if clean_indata
.get('vnfd:vnfd-catalog'):
116 clean_indata
= clean_indata
['vnfd:vnfd-catalog']
117 elif clean_indata
.get('vnfd-catalog'):
118 clean_indata
= clean_indata
['vnfd-catalog']
119 if clean_indata
.get('vnfd'):
120 if not isinstance(clean_indata
['vnfd'], list) or len(clean_indata
['vnfd']) != 1:
121 raise EngineException("'vnfd' must be a list only one element")
122 clean_indata
= clean_indata
['vnfd'][0]
124 if clean_indata
.get('nsd:nsd-catalog'):
125 clean_indata
= clean_indata
['nsd:nsd-catalog']
126 elif clean_indata
.get('nsd-catalog'):
127 clean_indata
= clean_indata
['nsd-catalog']
128 if clean_indata
.get('nsd'):
129 if not isinstance(clean_indata
['nsd'], list) or len(clean_indata
['nsd']) != 1:
130 raise EngineException("'nsd' must be a list only one element")
131 clean_indata
= clean_indata
['nsd'][0]
132 elif item
== "userDefinedData":
133 if "userDefinedData" in indata
:
134 clean_indata
= clean_indata
['userDefinedData']
137 def _check_project_dependencies(self
, project_id
):
139 Check if a project can be deleted
143 # TODO Is it needed to check descriptors _admin.project_read/project_write??
144 _filter
= {"projects": project_id
}
145 if self
.db
.get_list("users", _filter
):
146 raise EngineException("There are users that uses this project", http_code
=HTTPStatus
.CONFLICT
)
148 def _check_pdus_usage(self
, pdu_id
):
150 Check if a pdu can be deleted
154 # TODO Is it needed to check descriptors _admin.project_read/project_write??
155 _filter
= {"vdur.pdu-id": pdu_id
}
156 if self
.db
.get_list("vnfrs", _filter
):
157 raise EngineException("There are ns that uses this pdu", http_code
=HTTPStatus
.CONFLICT
)
159 def _check_dependencies_on_descriptor(self
, session
, item
, descriptor_id
, _id
):
161 Check that the descriptor to be deleded is not a dependency of others
162 :param session: client session information
163 :param item: can be vnfds, nsds
164 :param descriptor_id: id (provided by client) of descriptor to be deleted
165 :param _id: internal id of descriptor to be deleted
166 :return: None or raises exception
169 _filter
= {"constituent-vnfd.ANYINDEX.vnfd-id-ref": descriptor_id
}
170 if self
.get_item_list(session
, "nsds", _filter
):
171 raise EngineException("There are nsd that depends on this VNFD", http_code
=HTTPStatus
.CONFLICT
)
172 if self
.get_item_list(session
, "vnfrs", {"vnfd-id": _id
}):
173 raise EngineException("There are vnfr that depends on this VNFD", http_code
=HTTPStatus
.CONFLICT
)
175 _filter
= {"nsdId": _id
}
176 if self
.get_item_list(session
, "nsrs", _filter
):
177 raise EngineException("There are nsr that depends on this NSD", http_code
=HTTPStatus
.CONFLICT
)
179 def _check_descriptor_dependencies(self
, session
, item
, descriptor
):
181 Check that the dependent descriptors exist on a new descriptor or edition
182 :param session: client session information
183 :param item: can be nsds, nsrs
184 :param descriptor: descriptor to be inserted or edit
185 :return: None or raises exception
188 if not descriptor
.get("constituent-vnfd"):
190 for vnf
in descriptor
["constituent-vnfd"]:
191 vnfd_id
= vnf
["vnfd-id-ref"]
192 if not self
.get_item_list(session
, "vnfds", {"id": vnfd_id
}):
193 raise EngineException("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non "
194 "existing vnfd".format(vnfd_id
), http_code
=HTTPStatus
.CONFLICT
)
196 if not descriptor
.get("nsdId"):
198 nsd_id
= descriptor
["nsdId"]
199 if not self
.get_item_list(session
, "nsds", {"id": nsd_id
}):
200 raise EngineException("Descriptor error at nsdId='{}' references a non exist nsd".format(nsd_id
),
201 http_code
=HTTPStatus
.CONFLICT
)
203 def _check_edition(self
, session
, item
, indata
, id, force
=False):
205 if indata
.get("projects"):
206 if not session
["admin"]:
207 raise EngineException("Needed admin privileges to edit user projects", HTTPStatus
.UNAUTHORIZED
)
208 if indata
.get("password"):
209 # regenerate salt and encrypt password
211 indata
["_admin"] = {"salt": salt
}
212 indata
["password"] = sha256(indata
["password"].encode('utf-8') + salt
.encode('utf-8')).hexdigest()
214 def _validate_new_data(self
, session
, item
, indata
, id=None, force
=False):
216 # check username not exists
217 if not id and self
.db
.get_one(item
, {"username": indata
.get("username")}, fail_on_empty
=False,
219 raise EngineException("username '{}' exists".format(indata
["username"]), HTTPStatus
.CONFLICT
)
222 for p
in indata
["projects"]:
225 if not self
.db
.get_one("projects", {"_id": p
}, fail_on_empty
=False, fail_on_more
=False):
226 raise EngineException("project '{}' does not exists".format(p
), HTTPStatus
.CONFLICT
)
227 elif item
== "projects":
228 if not indata
.get("name"):
229 raise EngineException("missing 'name'")
230 # check name not exists
231 if not id and self
.db
.get_one(item
, {"name": indata
.get("name")}, fail_on_empty
=False, fail_on_more
=False):
232 raise EngineException("name '{}' exists".format(indata
["name"]), HTTPStatus
.CONFLICT
)
233 elif item
in ("vnfds", "nsds"):
234 filter = {"id": indata
["id"]}
236 filter["_id.neq"] = id
237 # TODO add admin to filter, validate rights
238 self
._add
_read
_filter
(session
, item
, filter)
239 if self
.db
.get_one(item
, filter, fail_on_empty
=False):
240 raise EngineException("{} with id '{}' already exists for this tenant".format(item
[:-1], indata
["id"]),
242 # TODO validate with pyangbind. Load and dumps to convert data types
244 # transform constituent-vnfd:member-vnf-index to string
245 if indata
.get("constituent-vnfd"):
246 for constituent_vnfd
in indata
["constituent-vnfd"]:
247 if "member-vnf-index" in constituent_vnfd
:
248 constituent_vnfd
["member-vnf-index"] = str(constituent_vnfd
["member-vnf-index"])
250 if item
== "nsds" and not force
:
251 self
._check
_descriptor
_dependencies
(session
, "nsds", indata
)
252 elif item
== "userDefinedData":
253 # TODO validate userDefinedData is a keypair values
258 elif item
== "vim_accounts" or item
== "sdns":
259 filter = {"name": indata
.get("name")}
261 filter["_id.neq"] = id
262 if self
.db
.get_one(item
, filter, fail_on_empty
=False, fail_on_more
=False):
263 raise EngineException("name '{}' already exists for {}".format(indata
["name"], item
),
266 def _check_ns_operation(self
, session
, nsr
, operation
, indata
):
268 Check that user has enter right parameters for the operation
270 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
271 :param indata: descriptor with the parameters of the operation
278 def check_valid_vnf_member_index(member_vnf_index
):
279 for vnf
in nsd
["constituent-vnfd"]:
280 if member_vnf_index
== vnf
["member-vnf-index"]:
281 vnfd_id
= vnf
["vnfd-id-ref"]
282 if vnfd_id
not in vnfds
:
283 vnfds
[vnfd_id
] = self
.db
.get_one("vnfds", {"id": vnfd_id
})
284 return vnfds
[vnfd_id
]
286 raise EngineException("Invalid parameter member_vnf_index='{}' is not one of the "
287 "nsd:constituent-vnfd".format(member_vnf_index
))
289 def check_valid_vim_account(vim_account
):
290 if vim_account
in vim_accounts
:
293 self
.db
.get_one("vim_accounts", {"_id": vim_account
})
295 raise EngineException("Invalid vimAccountId='{}' not present".format(vim_account
))
296 vim_accounts
.append(vim_account
)
298 if operation
== "action":
299 # check vnf_member_index
300 if indata
.get("vnf_member_index"):
301 indata
["member_vnf_index"] = indata
.pop("vnf_member_index") # for backward compatibility
302 if not indata
.get("member_vnf_index"):
303 raise EngineException("Missing 'member_vnf_index' parameter")
304 vnfd
= check_valid_vnf_member_index(indata
["member_vnf_index"])
306 for config_primitive
in get_iterable(vnfd
.get("vnf-configuration", {}).get("config-primitive")):
307 if indata
["primitive"] == config_primitive
["name"]:
308 # check needed primitive_params are provided
309 if indata
.get("primitive_params"):
310 in_primitive_params_copy
= copy(indata
["primitive_params"])
312 in_primitive_params_copy
= {}
313 for paramd
in get_iterable(config_primitive
.get("parameter")):
314 if paramd
["name"] in in_primitive_params_copy
:
315 del in_primitive_params_copy
[paramd
["name"]]
316 elif not paramd
.get("default-value"):
317 raise EngineException("Needed parameter {} not provided for primitive '{}'".format(
318 paramd
["name"], indata
["primitive"]))
319 # check no extra primitive params are provided
320 if in_primitive_params_copy
:
321 raise EngineException("parameter/s '{}' not present at vnfd for primitive '{}'".format(
322 list(in_primitive_params_copy
.keys()), indata
["primitive"]))
325 raise EngineException("Invalid primitive '{}' is not present at vnfd".format(indata
["primitive"]))
326 if operation
== "scale":
327 vnfd
= check_valid_vnf_member_index(indata
["scaleVnfData"]["scaleByStepData"]["member-vnf-index"])
328 for scaling_group
in get_iterable(vnfd
.get("scaling-group-descriptor")):
329 if indata
["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] == scaling_group
["name"]:
332 raise EngineException("Invalid scaleVnfData:scaleByStepData:scaling-group-descriptor '{}' is not "
333 "present at vnfd:scaling-group-descriptor".format(
334 indata
["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]))
335 if operation
== "instantiate":
337 check_valid_vim_account(indata
["vimAccountId"])
338 for in_vnf
in get_iterable(indata
.get("vnf")):
339 vnfd
= check_valid_vnf_member_index(in_vnf
["member-vnf-index"])
340 if in_vnf
.get("vimAccountId"):
341 check_valid_vim_account(in_vnf
["vimAccountId"])
342 for in_vdu
in get_iterable(in_vnf
.get("vdu")):
343 for vdud
in get_iterable(vnfd
.get("vdu")):
344 if vdud
["id"] == in_vdu
["id"]:
345 for volume
in get_iterable(in_vdu
.get("volume")):
346 for volumed
in get_iterable(vdud
.get("volumes")):
347 if volumed
["name"] == volume
["name"]:
350 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu[id='{}']:"
351 "volume:name='{}' is not present at vnfd:vdu:volumes list".
352 format(in_vnf
["member-vnf-index"], in_vdu
["id"],
356 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu:id='{}' is not "
357 "present at vnfd".format(in_vnf
["member-vnf-index"], in_vdu
["id"]))
359 for in_internal_vld
in get_iterable(in_vnf
.get("internal-vld")):
360 for internal_vldd
in get_iterable(vnfd
.get("internal-vld")):
361 if in_internal_vld
["name"] == internal_vldd
["name"] or \
362 in_internal_vld
["name"] == internal_vldd
["id"]:
365 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:internal-vld:name='{}'"
366 " is not present at vnfd '{}'".format(in_vnf
["member-vnf-index"],
367 in_internal_vld
["name"],
369 for in_vld
in get_iterable(indata
.get("vld")):
370 for vldd
in get_iterable(nsd
.get("vld")):
371 if in_vld
["name"] == vldd
["name"] or in_vld
["name"] == vldd
["id"]:
374 raise EngineException("Invalid parameter vld:name='{}' is not present at nsd:vld".format(
377 def _format_new_data(self
, session
, item
, indata
):
379 if "_admin" not in indata
:
380 indata
["_admin"] = {}
381 indata
["_admin"]["created"] = now
382 indata
["_admin"]["modified"] = now
384 indata
["_id"] = indata
["username"]
386 indata
["_admin"]["salt"] = salt
387 indata
["password"] = sha256(indata
["password"].encode('utf-8') + salt
.encode('utf-8')).hexdigest()
388 elif item
== "projects":
389 indata
["_id"] = indata
["name"]
391 if not indata
.get("_id"):
392 indata
["_id"] = str(uuid4())
393 if item
in ("vnfds", "nsds", "nsrs", "vnfrs", "pdus"):
394 if not indata
["_admin"].get("projects_read"):
395 indata
["_admin"]["projects_read"] = [session
["project_id"]]
396 if not indata
["_admin"].get("projects_write"):
397 indata
["_admin"]["projects_write"] = [session
["project_id"]]
398 if item
in ("vnfds", "nsds", "pdus"):
399 indata
["_admin"]["onboardingState"] = "CREATED"
400 indata
["_admin"]["operationalState"] = "DISABLED"
401 indata
["_admin"]["usageSate"] = "NOT_IN_USE"
403 indata
["_admin"]["nsState"] = "NOT_INSTANTIATED"
404 if item
in ("vim_accounts", "sdns"):
405 indata
["_admin"]["operationalState"] = "PROCESSING"
407 def upload_content(self
, session
, item
, _id
, indata
, kwargs
, headers
):
409 Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract)
410 :param session: session
411 :param item: can be nsds or vnfds
412 :param _id : the nsd,vnfd is already created, this is the id
413 :param indata: http body request
414 :param kwargs: user query string to override parameters. NOT USED
415 :param headers: http request headers
416 :return: True package has is completely uploaded or False if partial content has been uplodaed.
417 Raise exception on error
419 # Check that _id exists and it is valid
420 current_desc
= self
.get_item(session
, item
, _id
)
422 content_range_text
= headers
.get("Content-Range")
423 expected_md5
= headers
.get("Content-File-MD5")
425 content_type
= headers
.get("Content-Type")
426 if content_type
and "application/gzip" in content_type
or "application/x-gzip" in content_type
or \
427 "application/zip" in content_type
:
429 filename
= headers
.get("Content-Filename")
431 filename
= "package.tar.gz" if compressed
else "package"
432 # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266
436 if content_range_text
:
437 content_range
= content_range_text
.replace("-", " ").replace("/", " ").split()
438 if content_range
[0] != "bytes": # TODO check x<y not negative < total....
440 start
= int(content_range
[1])
441 end
= int(content_range
[2]) + 1
442 total
= int(content_range
[3])
447 if not self
.fs
.file_exists(_id
, 'dir'):
448 raise EngineException("invalid Transaction-Id header", HTTPStatus
.NOT_FOUND
)
450 self
.fs
.file_delete(_id
, ignore_non_exist
=True)
453 storage
= self
.fs
.get_params()
454 storage
["folder"] = _id
456 file_path
= (_id
, filename
)
457 if self
.fs
.file_exists(file_path
, 'file'):
458 file_size
= self
.fs
.file_size(file_path
)
461 if file_size
!= start
:
462 raise EngineException("invalid Content-Range start sequence, expected '{}' but received '{}'".format(
463 file_size
, start
), HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
464 file_pkg
= self
.fs
.file_open(file_path
, 'a+b')
465 if isinstance(indata
, dict):
466 indata_text
= yaml
.safe_dump(indata
, indent
=4, default_flow_style
=False)
467 file_pkg
.write(indata_text
.encode(encoding
="utf-8"))
471 indata_text
= indata
.read(4096)
472 indata_len
+= len(indata_text
)
475 file_pkg
.write(indata_text
)
476 if content_range_text
:
477 if indata_len
!= end
-start
:
478 raise EngineException("Mismatch between Content-Range header {}-{} and body length of {}".format(
479 start
, end
-1, indata_len
), HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
481 # TODO update to UPLOADING
488 chunk_data
= file_pkg
.read(1024)
490 file_md5
.update(chunk_data
)
491 chunk_data
= file_pkg
.read(1024)
492 if expected_md5
!= file_md5
.hexdigest():
493 raise EngineException("Error, MD5 mismatch", HTTPStatus
.CONFLICT
)
495 if compressed
== "gzip":
496 tar
= tarfile
.open(mode
='r', fileobj
=file_pkg
)
497 descriptor_file_name
= None
499 tarname
= tarinfo
.name
500 tarname_path
= tarname
.split("/")
501 if not tarname_path
[0] or ".." in tarname_path
: # if start with "/" means absolute path
502 raise EngineException("Absolute path or '..' are not allowed for package descriptor tar.gz")
503 if len(tarname_path
) == 1 and not tarinfo
.isdir():
504 raise EngineException("All files must be inside a dir for package descriptor tar.gz")
505 if tarname
.endswith(".yaml") or tarname
.endswith(".json") or tarname
.endswith(".yml"):
506 storage
["pkg-dir"] = tarname_path
[0]
507 if len(tarname_path
) == 2:
508 if descriptor_file_name
:
509 raise EngineException(
510 "Found more than one descriptor file at package descriptor tar.gz")
511 descriptor_file_name
= tarname
512 if not descriptor_file_name
:
513 raise EngineException("Not found any descriptor file at package descriptor tar.gz")
514 storage
["descriptor"] = descriptor_file_name
515 storage
["zipfile"] = filename
516 self
.fs
.file_extract(tar
, _id
)
517 with self
.fs
.file_open((_id
, descriptor_file_name
), "r") as descriptor_file
:
518 content
= descriptor_file
.read()
520 content
= file_pkg
.read()
521 storage
["descriptor"] = descriptor_file_name
= filename
523 if descriptor_file_name
.endswith(".json"):
524 error_text
= "Invalid json format "
525 indata
= json
.load(content
)
527 error_text
= "Invalid yaml format "
528 indata
= yaml
.load(content
)
530 current_desc
["_admin"]["storage"] = storage
531 current_desc
["_admin"]["onboardingState"] = "ONBOARDED"
532 current_desc
["_admin"]["operationalState"] = "ENABLED"
534 self
._edit
_item
(session
, item
, _id
, current_desc
, indata
, kwargs
)
535 # TODO if descriptor has changed because kwargs update content and remove cached zip
536 # TODO if zip is not present creates one
539 except EngineException
:
542 raise EngineException("invalid Content-Range header format. Expected 'bytes start-end/total'",
543 HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
545 raise EngineException("invalid upload transaction sequence: '{}'".format(e
), HTTPStatus
.BAD_REQUEST
)
546 except tarfile
.ReadError
as e
:
547 raise EngineException("invalid file content {}".format(e
), HTTPStatus
.BAD_REQUEST
)
548 except (ValueError, yaml
.YAMLError
) as e
:
549 raise EngineException(error_text
+ str(e
))
554 def new_nsr(self
, rollback
, session
, ns_request
):
556 Creates a new nsr into database. It also creates needed vnfrs
557 :param rollback: list where this method appends created items at database in case a rollback may to be done
558 :param session: contains the used login username and working project
559 :param ns_request: params to be used for the nsr
560 :return: the _id of nsr descriptor stored at database
562 rollback_index
= len(rollback
)
566 step
= "getting nsd id='{}' from database".format(ns_request
.get("nsdId"))
567 nsd
= self
.get_item(session
, "nsds", ns_request
["nsdId"])
568 nsr_id
= str(uuid4())
570 step
= "filling nsr from input data"
572 "name": ns_request
["nsName"],
573 "name-ref": ns_request
["nsName"],
574 "short-name": ns_request
["nsName"],
575 "admin-status": "ENABLED",
577 "datacenter": ns_request
["vimAccountId"],
578 "resource-orchestrator": "osmopenmano",
579 "description": ns_request
.get("nsDescription", ""),
580 "constituent-vnfr-ref": [],
582 "operational-status": "init", # typedef ns-operational-
583 "config-status": "init", # typedef config-states
584 "detailed-status": "scheduled",
586 "orchestration-progress": {},
587 # {"networks": {"active": 0, "total": 0}, "vms": {"active": 0, "total": 0}},
590 "nsd-name-ref": nsd
["name"],
591 "operational-events": [], # "id", "timestamp", "description", "event",
592 "nsd-ref": nsd
["id"],
594 "instantiate_params": ns_request
,
595 "ns-instance-config-ref": nsr_id
,
598 # "input-parameter": xpath, value,
599 "ssh-authorized-key": ns_request
.get("key-pair-ref"),
601 ns_request
["nsr_id"] = nsr_id
605 for member_vnf
in nsd
["constituent-vnfd"]:
606 vnfd_id
= member_vnf
["vnfd-id-ref"]
607 step
= "getting vnfd id='{}' constituent-vnfd='{}' from database".format(
608 member_vnf
["vnfd-id-ref"], member_vnf
["member-vnf-index"])
609 if vnfd_id
not in needed_vnfds
:
611 vnf_filter
= {"id": vnfd_id
}
612 self
._add
_read
_filter
(session
, "vnfds", vnf_filter
)
613 vnfd
= self
.db
.get_one("vnfds", vnf_filter
)
615 needed_vnfds
[vnfd_id
] = vnfd
617 vnfd
= needed_vnfds
[vnfd_id
]
618 step
= "filling vnfr vnfd-id='{}' constituent-vnfd='{}'".format(
619 member_vnf
["vnfd-id-ref"], member_vnf
["member-vnf-index"])
620 vnfr_id
= str(uuid4())
624 "nsr-id-ref": nsr_id
,
625 "member-vnf-index-ref": member_vnf
["member-vnf-index"],
627 # "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise
629 "vnfd-id": vnfd
["_id"], # not at OSM model, but useful
630 "vim-account-id": None,
632 "connection-point": [],
633 "ip-address": None, # mgmt-interface filled by LCM
635 for cp
in vnfd
.get("connection-point", ()):
638 "connection-point-id": cp
.get("id"),
640 # "ip-address", "mac-address" # filled by LCM
641 # vim-id # TODO it would be nice having a vim port id
643 vnfr_descriptor
["connection-point"].append(vnf_cp
)
644 for vdu
in vnfd
["vdu"]:
645 vdur_id
= str(uuid4())
648 "vdu-id-ref": vdu
["id"],
649 # TODO "name": "" Name of the VDU in the VIM
650 "ip-address": None, # mgmt-interface filled by LCM
651 # "vim-id", "flavor-id", "image-id", "management-ip" # filled by LCM
652 "internal-connection-point": [],
655 if vnfd
.get("pdu-type"):
656 vdur
["pdu-type"] = vnfd
["pdu-type"]
657 # TODO volumes: name, volume-id
658 for icp
in vdu
.get("internal-connection-point", ()):
661 "connection-point-id": icp
["id"],
662 "name": icp
.get("name"),
663 # "ip-address", "mac-address" # filled by LCM
664 # vim-id # TODO it would be nice having a vim port id
666 vdur
["internal-connection-point"].append(vdu_icp
)
667 for iface
in vdu
.get("interface", ()):
669 "name": iface
.get("name"),
670 # "ip-address", "mac-address" # filled by LCM
671 # vim-id # TODO it would be nice having a vim port id
673 vdur
["interfaces"].append(vdu_iface
)
674 vnfr_descriptor
["vdur"].append(vdur
)
676 step
= "creating vnfr vnfd-id='{}' constituent-vnfd='{}' at database".format(
677 member_vnf
["vnfd-id-ref"], member_vnf
["member-vnf-index"])
678 self
._format
_new
_data
(session
, "vnfrs", vnfr_descriptor
)
679 self
.db
.create("vnfrs", vnfr_descriptor
)
680 rollback
.insert(0, {"item": "vnfrs", "_id": vnfr_id
})
681 nsr_descriptor
["constituent-vnfr-ref"].append(vnfr_id
)
683 step
= "creating nsr at database"
684 self
._format
_new
_data
(session
, "nsrs", nsr_descriptor
)
685 self
.db
.create("nsrs", nsr_descriptor
)
686 rollback
.insert(rollback_index
, {"item": "nsrs", "_id": nsr_id
})
688 except Exception as e
:
689 raise EngineException("Error {}: {}".format(step
, e
))
692 def _update_descriptor(desc
, kwargs
):
694 Update descriptor with the kwargs. It contains dot separated keys
695 :param desc: dictionary to be updated
696 :param kwargs: plain dictionary to be used for updating.
702 for k
, v
in kwargs
.items():
703 update_content
= desc
707 if kitem_old
is not None:
708 update_content
= update_content
[kitem_old
]
709 if isinstance(update_content
, dict):
711 elif isinstance(update_content
, list):
712 kitem_old
= int(kitem
)
714 raise EngineException(
715 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k
, kitem
))
716 update_content
[kitem_old
] = v
718 raise EngineException(
719 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k
, kitem_old
))
721 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
724 raise EngineException(
725 "Invalid query string '{}'. Index '{}' out of range".format(k
, kitem_old
))
727 def new_item(self
, rollback
, session
, item
, indata
={}, kwargs
=None, headers
={}, force
=False):
729 Creates a new entry into database. For nsds and vnfds it creates an almost empty DISABLED entry,
730 that must be completed with a call to method upload_content
731 :param rollback: list where this method appends created items at database in case a rollback may to be done
732 :param session: contains the used login username and working project
733 :param item: it can be: users, projects, vim_accounts, sdns, nsrs, nsds, vnfds
734 :param indata: data to be inserted
735 :param kwargs: used to override the indata descriptor
736 :param headers: http request headers
737 :param force: If True avoid some dependence checks
738 :return: _id: identity of the inserted data.
741 if not session
["admin"] and item
in ("users", "projects"):
742 raise EngineException("Needed admin privileges to perform this operation", HTTPStatus
.UNAUTHORIZED
)
746 if item
in ("nsds", "vnfds"):
747 item_envelop
= "userDefinedData"
748 content
= self
._remove
_envelop
(item_envelop
, indata
)
750 # Override descriptor with query string kwargs
751 self
._update
_descriptor
(content
, kwargs
)
752 if not content
and item
not in ("nsds", "vnfds"):
753 raise EngineException("Empty payload")
755 validate_input(content
, item
, new
=True)
758 # in this case the input descriptor is not the data to be stored
759 return self
.new_nsr(rollback
, session
, ns_request
=content
)
761 self
._validate
_new
_data
(session
, item_envelop
, content
, force
=force
)
762 if item
in ("nsds", "vnfds"):
763 content
= {"_admin": {"userDefinedData": content
}}
764 self
._format
_new
_data
(session
, item
, content
)
765 _id
= self
.db
.create(item
, content
)
766 rollback
.insert(0, {"item": item
, "_id": _id
})
768 if item
== "vim_accounts":
769 msg_data
= self
.db
.get_one(item
, {"_id": _id
})
770 msg_data
.pop("_admin", None)
771 self
.msg
.write("vim_account", "create", msg_data
)
773 msg_data
= self
.db
.get_one(item
, {"_id": _id
})
774 msg_data
.pop("_admin", None)
775 self
.msg
.write("sdn", "create", msg_data
)
777 except ValidationError
as e
:
778 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
780 def new_nslcmop(self
, session
, nsInstanceId
, operation
, params
):
786 "operationState": "PROCESSING", # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
787 "statusEnteredTime": now
,
788 "nsInstanceId": nsInstanceId
,
789 "lcmOperationType": operation
,
791 "isAutomaticInvocation": False,
792 "operationParams": params
,
793 "isCancelPending": False,
795 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
796 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsInstanceId
,
801 def ns_operation(self
, rollback
, session
, nsInstanceId
, operation
, indata
, kwargs
=None):
803 Performs a new operation over a ns
804 :param rollback: list where this method appends created items at database in case a rollback may to be done
805 :param session: contains the used login username and working project
806 :param nsInstanceId: _id of the nsr to perform the operation
807 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
808 :param indata: descriptor with the parameters of the operation
809 :param kwargs: used to override the indata descriptor
810 :return: id of the nslcmops
813 # Override descriptor with query string kwargs
814 self
._update
_descriptor
(indata
, kwargs
)
815 validate_input(indata
, "ns_" + operation
, new
=True)
817 nsr
= self
.get_item(session
, "nsrs", nsInstanceId
)
818 if not nsr
["_admin"].get("nsState") or nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
819 if operation
== "terminate" and indata
.get("autoremove"):
820 # NSR must be deleted
821 return self
.del_item(session
, "nsrs", nsInstanceId
)
822 if operation
!= "instantiate":
823 raise EngineException("ns_instance '{}' cannot be '{}' because it is not instantiated".format(
824 nsInstanceId
, operation
), HTTPStatus
.CONFLICT
)
826 if operation
== "instantiate" and not indata
.get("force"):
827 raise EngineException("ns_instance '{}' cannot be '{}' because it is already instantiated".format(
828 nsInstanceId
, operation
), HTTPStatus
.CONFLICT
)
829 indata
["nsInstanceId"] = nsInstanceId
830 self
._check
_ns
_operation
(session
, nsr
, operation
, indata
)
831 nslcmop
= self
.new_nslcmop(session
, nsInstanceId
, operation
, indata
)
832 self
._format
_new
_data
(session
, "nslcmops", nslcmop
)
833 _id
= self
.db
.create("nslcmops", nslcmop
)
834 rollback
.insert(0, {"item": "nslcmops", "_id": _id
})
836 self
.msg
.write("ns", operation
, nslcmop
)
838 except ValidationError
as e
:
839 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
840 # except DbException as e:
841 # raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND)
843 def _add_read_filter(self
, session
, item
, filter):
844 if session
["admin"]: # allows all
847 filter["username"] = session
["username"]
848 elif item
in ("vnfds", "nsds", "nsrs", "vnfrs", "pdus"):
849 filter["_admin.projects_read.cont"] = ["ANY", session
["project_id"]]
851 def _add_delete_filter(self
, session
, item
, filter):
852 if not session
["admin"] and item
in ("users", "projects"):
853 raise EngineException("Only admin users can perform this task", http_code
=HTTPStatus
.FORBIDDEN
)
855 if filter.get("_id") == session
["username"] or filter.get("username") == session
["username"]:
856 raise EngineException("You cannot delete your own user", http_code
=HTTPStatus
.CONFLICT
)
857 elif item
== "project":
858 if filter.get("_id") == session
["project_id"]:
859 raise EngineException("You cannot delete your own project", http_code
=HTTPStatus
.CONFLICT
)
860 elif item
in ("vnfds", "nsds", "pdus") and not session
["admin"]:
861 filter["_admin.projects_write.cont"] = ["ANY", session
["project_id"]]
863 def get_file(self
, session
, item
, _id
, path
=None, accept_header
=None):
865 Return the file content of a vnfd or nsd
866 :param session: contains the used login username and working project
867 :param item: it can be vnfds or nsds
868 :param _id: Identity of the vnfd, ndsd
869 :param path: artifact path or "$DESCRIPTOR" or None
870 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
871 :return: opened file or raises an exception
873 accept_text
= accept_zip
= False
875 if 'text/plain' in accept_header
or '*/*' in accept_header
:
877 if 'application/zip' in accept_header
or '*/*' in accept_header
:
879 if not accept_text
and not accept_zip
:
880 raise EngineException("provide request header 'Accept' with 'application/zip' or 'text/plain'",
881 http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
883 content
= self
.get_item(session
, item
, _id
)
884 if content
["_admin"]["onboardingState"] != "ONBOARDED":
885 raise EngineException("Cannot get content because this resource is not at 'ONBOARDED' state. "
886 "onboardingState is {}".format(content
["_admin"]["onboardingState"]),
887 http_code
=HTTPStatus
.CONFLICT
)
888 storage
= content
["_admin"]["storage"]
889 if path
is not None and path
!= "$DESCRIPTOR": # artifacts
890 if not storage
.get('pkg-dir'):
891 raise EngineException("Packages does not contains artifacts", http_code
=HTTPStatus
.BAD_REQUEST
)
892 if self
.fs
.file_exists((storage
['folder'], storage
['pkg-dir'], *path
), 'dir'):
893 folder_content
= self
.fs
.dir_ls((storage
['folder'], storage
['pkg-dir'], *path
))
894 return folder_content
, "text/plain"
895 # TODO manage folders in http
897 return self
.fs
.file_open((storage
['folder'], storage
['pkg-dir'], *path
), "rb"),\
898 "application/octet-stream"
900 # pkgtype accept ZIP TEXT -> result
901 # manyfiles yes X -> zip
903 # onefile yes no -> zip
906 if accept_text
and (not storage
.get('pkg-dir') or path
== "$DESCRIPTOR"):
907 return self
.fs
.file_open((storage
['folder'], storage
['descriptor']), "r"), "text/plain"
908 elif storage
.get('pkg-dir') and not accept_zip
:
909 raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'"
910 "Accept header", http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
912 if not storage
.get('zipfile'):
913 # TODO generate zipfile if not present
914 raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in "
915 "future versions", http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
916 return self
.fs
.file_open((storage
['folder'], storage
['zipfile']), "rb"), "application/zip"
918 def get_item_list(self
, session
, item
, filter={}):
921 :param session: contains the used login username and working project
922 :param item: it can be: users, projects, vnfds, nsds, ...
923 :param filter: filter of data to be applied
924 :return: The list, it can be empty if no one match the filter.
926 # TODO add admin to filter, validate rights
927 # TODO transform data for SOL005 URL requests. Transform filtering
928 # TODO implement "field-type" query string SOL005
930 self
._add
_read
_filter
(session
, item
, filter)
931 return self
.db
.get_list(item
, filter)
933 def get_item(self
, session
, item
, _id
):
935 Get complete information on an items
936 :param session: contains the used login username and working project
937 :param item: it can be: users, projects, vnfds, nsds,
938 :param _id: server id of the item
939 :return: dictionary, raise exception if not found.
941 filter = {"_id": _id
}
942 # TODO add admin to filter, validate rights
943 # TODO transform data for SOL005 URL requests
944 self
._add
_read
_filter
(session
, item
, filter)
945 return self
.db
.get_one(item
, filter)
947 def del_item_list(self
, session
, item
, filter={}):
949 Delete a list of items
950 :param session: contains the used login username and working project
951 :param item: it can be: users, projects, vnfds, nsds, ...
952 :param filter: filter of data to be applied
953 :return: The deleted list, it can be empty if no one match the filter.
955 # TODO add admin to filter, validate rights
956 self
._add
_read
_filter
(session
, item
, filter)
957 return self
.db
.del_list(item
, filter)
959 def del_item(self
, session
, item
, _id
, force
=False):
961 Delete item by its internal id
962 :param session: contains the used login username and working project
963 :param item: it can be: users, projects, vnfds, nsds, ...
964 :param _id: server id of the item
965 :param force: indicates if deletion must be forced in case of conflict
966 :return: dictionary with deleted item _id. It raises exception if not found.
968 # TODO add admin to filter, validate rights
969 # data = self.get_item(item, _id)
970 filter = {"_id": _id
}
971 self
._add
_delete
_filter
(session
, item
, filter)
972 if item
in ("vnfds", "nsds") and not force
:
973 descriptor
= self
.get_item(session
, item
, _id
)
974 descriptor_id
= descriptor
.get("id")
976 self
._check
_dependencies
_on
_descriptor
(session
, item
, descriptor_id
, _id
)
977 elif item
== "projects":
979 self
._check
_project
_dependencies
(_id
)
982 self
._check
_pdus
_usage
(_id
)
985 nsr
= self
.db
.get_one(item
, filter)
986 if nsr
["_admin"].get("nsState") == "INSTANTIATED" and not force
:
987 raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. "
988 "Launch 'terminate' operation first; or force deletion".format(_id
),
989 http_code
=HTTPStatus
.CONFLICT
)
990 v
= self
.db
.del_one(item
, {"_id": _id
})
991 self
.db
.del_list("nslcmops", {"nsInstanceId": _id
})
992 self
.db
.del_list("vnfrs", {"nsr-id-ref": _id
})
993 # set all used pdus as free
994 self
.db
.set_list("pdus", {"_admin.usage.nsr_id": _id
},
995 {"_admin.usageSate": "NOT_IN_USE", "_admin.usage": None})
996 self
.msg
.write("ns", "deleted", {"_id": _id
})
998 if item
in ("vim_accounts", "sdns") and not force
:
999 self
.db
.set_one(item
, {"_id": _id
}, {"_admin.to_delete": True}) # TODO change status
1000 if item
== "vim_accounts":
1001 self
.msg
.write("vim_account", "delete", {"_id": _id
})
1002 elif item
== "sdns":
1003 self
.msg
.write("sdn", "delete", {"_id": _id
})
1004 return {"deleted": 1} # TODO indicate an offline operation to return 202 ACCEPTED
1006 v
= self
.db
.del_one(item
, filter)
1007 if item
in ("vnfds", "nsds"):
1008 self
.fs
.file_delete(_id
, ignore_non_exist
=True)
1009 if item
in ("vim_accounts", "sdns", "vnfds", "nsds"):
1010 self
.msg
.write(item
[:-1], "deleted", {"_id": _id
})
1015 Prune database not needed content
1018 return self
.db
.del_list("nsrs", {"_admin.to_delete": True})
1020 def create_admin(self
):
1022 Creates a new user admin/admin into database if database is empty. Useful for initialization
1023 :return: _id identity of the inserted data, or None
1025 users
= self
.db
.get_one("users", fail_on_empty
=False, fail_on_more
=False)
1028 # raise EngineException("Unauthorized. Database users is not empty", HTTPStatus.UNAUTHORIZED)
1029 indata
= {"username": "admin", "password": "admin", "projects": ["admin"]}
1030 fake_session
= {"project_id": "admin", "username": "admin"}
1031 self
._format
_new
_data
(fake_session
, "users", indata
)
1032 _id
= self
.db
.create("users", indata
)
1035 def init_db(self
, target_version
='1.0'):
1037 Init database if empty. If not empty it checks that database version is ok.
1038 If empty, it creates a new user admin/admin at 'users' and a new entry at 'version'
1039 :return: None if ok, exception if error or if the version is different.
1041 version
= self
.db
.get_one("version", fail_on_empty
=False, fail_on_more
=False)
1045 # create database version
1047 "_id": '1.0', # version text
1048 "version": 1000, # version number
1049 "date": "2018-04-12", # version date
1050 "description": "initial design", # changes in this version
1051 'status': 'ENABLED' # ENABLED, DISABLED (migration in process), ERROR,
1053 self
.db
.create("version", version_data
)
1054 elif version
["_id"] != target_version
:
1055 # TODO implement migration process
1056 raise EngineException("Wrong database version '{}'. Expected '{}'".format(
1057 version
["_id"], target_version
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
1058 elif version
["status"] != 'ENABLED':
1059 raise EngineException("Wrong database status '{}'".format(
1060 version
["status"]), HTTPStatus
.INTERNAL_SERVER_ERROR
)
1063 def _edit_item(self
, session
, item
, id, content
, indata
={}, kwargs
=None, force
=False):
1065 indata
= self
._remove
_envelop
(item
, indata
)
1067 # Override descriptor with query string kwargs
1070 for k
, v
in kwargs
.items():
1071 update_content
= indata
1073 klist
= k
.split(".")
1075 if kitem_old
is not None:
1076 update_content
= update_content
[kitem_old
]
1077 if isinstance(update_content
, dict):
1079 elif isinstance(update_content
, list):
1080 kitem_old
= int(kitem
)
1082 raise EngineException(
1083 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k
, kitem
))
1084 update_content
[kitem_old
] = v
1086 raise EngineException(
1087 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k
, kitem_old
))
1089 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
1092 raise EngineException(
1093 "Invalid query string '{}'. Index '{}' out of range".format(k
, kitem_old
))
1095 validate_input(indata
, item
, new
=False)
1096 except ValidationError
as e
:
1097 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
1099 self
._check
_edition
(session
, item
, indata
, id, force
)
1100 deep_update(content
, indata
)
1101 self
._validate
_new
_data
(session
, item
, content
, id=id, force
=force
)
1102 # self._format_new_data(session, item, content)
1103 self
.db
.replace(item
, id, content
)
1104 if item
in ("vim_accounts", "sdns"):
1105 indata
.pop("_admin", None)
1107 if item
== "vim_accounts":
1108 self
.msg
.write("vim_account", "edit", indata
)
1109 elif item
== "sdns":
1110 self
.msg
.write("sdn", "edit", indata
)
1113 def edit_item(self
, session
, item
, _id
, indata
={}, kwargs
=None, force
=False):
1115 Update an existing entry at database
1116 :param session: contains the used login username and working project
1117 :param item: it can be: users, projects, vnfds, nsds, ...
1118 :param _id: identifier to be updated
1119 :param indata: data to be inserted
1120 :param kwargs: used to override the indata descriptor
1121 :param force: If True avoid some dependence checks
1122 :return: dictionary, raise exception if not found.
1124 if not session
["admin"] and item
== "projects":
1125 raise EngineException("Needed admin privileges to perform this operation", HTTPStatus
.UNAUTHORIZED
)
1127 content
= self
.get_item(session
, item
, _id
)
1128 return self
._edit
_item
(session
, item
, _id
, content
, indata
, kwargs
, force
)