projects / osm / RO.git / blob
? search:


35502c283cf90323eb3af6d1f17fa50f7e826d84
[osm/RO.git] / NG-RO / osm_ng_ro / ns.py
1 # -*- coding: utf-8 -*-
2
3 ##
4 # Copyright 2020 Telefonica Investigacion y Desarrollo, S.A.U.
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
8 #
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #
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
14 # implied.
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
17 ##
18
19 # import yaml
20 import logging
21 from typing import Any, Dict
22 from traceback import format_exc as traceback_format_exc
23 from osm_ng_ro.ns_thread import NsWorker, NsWorkerException, deep_get
24 from osm_ng_ro.validation import validate_input, deploy_schema
25 from osm_common import (
26 dbmongo,
27 dbmemory,
28 fslocal,
29 fsmongo,
30 msglocal,
31 msgkafka,
32 version as common_version,
33 )
34 from osm_common.dbbase import DbException
35 from osm_common.fsbase import FsException
36 from osm_common.msgbase import MsgException
37 from http import HTTPStatus
38 from uuid import uuid4
39 from threading import Lock
40 from random import choice as random_choice
41 from time import time
42 from jinja2 import (
43 Environment,
44 TemplateError,
45 TemplateNotFound,
46 StrictUndefined,
47 UndefinedError,
48 )
49 from cryptography.hazmat.primitives import serialization as crypto_serialization
50 from cryptography.hazmat.primitives.asymmetric import rsa
51 from cryptography.hazmat.backends import default_backend as crypto_default_backend
52
53 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
54 min_common_version = "0.1.16"
55
56
57 class NsException(Exception):
58 def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
59 self.http_code = http_code
60 super(Exception, self).__init__(message)
61
62
63 def get_process_id():
64 """
65 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
66 will provide a random one
67 :return: Obtained ID
68 """
69 # Try getting docker id. If fails, get pid
70 try:
71 with open("/proc/self/cgroup", "r") as f:
72 text_id_ = f.readline()
73 _, _, text_id = text_id_.rpartition("/")
74 text_id = text_id.replace("\n", "")[:12]
75
76 if text_id:
77 return text_id
78 except Exception:
79 pass
80
81 # Return a random id
82 return "".join(random_choice("0123456789abcdef") for _ in range(12))
83
84
85 def versiontuple(v):
86 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
87 filled = []
88
89 for point in v.split("."):
90 filled.append(point.zfill(8))
91
92 return tuple(filled)
93
94
95 class Ns(object):
96 def __init__(self):
97 self.db = None
98 self.fs = None
99 self.msg = None
100 self.config = None
101 # self.operations = None
102 self.logger = None
103 # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
104 # If done now it will not be linked to parent not getting its handler and level
105 self.map_topic = {}
106 self.write_lock = None
107 self.vims_assigned = {}
108 self.next_worker = 0
109 self.plugins = {}
110 self.workers = []
111
112 def init_db(self, target_version):
113 pass
114
115 def start(self, config):
116 """
117 Connect to database, filesystem storage, and messaging
118 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
119 :param config: Configuration of db, storage, etc
120 :return: None
121 """
122 self.config = config
123 self.config["process_id"] = get_process_id() # used for HA identity
124 self.logger = logging.getLogger("ro.ns")
125
126 # check right version of common
127 if versiontuple(common_version) < versiontuple(min_common_version):
128 raise NsException(
129 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
130 common_version, min_common_version
131 )
132 )
133
134 try:
135 if not self.db:
136 if config["database"]["driver"] == "mongo":
137 self.db = dbmongo.DbMongo()
138 self.db.db_connect(config["database"])
139 elif config["database"]["driver"] == "memory":
140 self.db = dbmemory.DbMemory()
141 self.db.db_connect(config["database"])
142 else:
143 raise NsException(
144 "Invalid configuration param '{}' at '[database]':'driver'".format(
145 config["database"]["driver"]
146 )
147 )
148
149 if not self.fs:
150 if config["storage"]["driver"] == "local":
151 self.fs = fslocal.FsLocal()
152 self.fs.fs_connect(config["storage"])
153 elif config["storage"]["driver"] == "mongo":
154 self.fs = fsmongo.FsMongo()
155 self.fs.fs_connect(config["storage"])
156 elif config["storage"]["driver"] is None:
157 pass
158 else:
159 raise NsException(
160 "Invalid configuration param '{}' at '[storage]':'driver'".format(
161 config["storage"]["driver"]
162 )
163 )
164
165 if not self.msg:
166 if config["message"]["driver"] == "local":
167 self.msg = msglocal.MsgLocal()
168 self.msg.connect(config["message"])
169 elif config["message"]["driver"] == "kafka":
170 self.msg = msgkafka.MsgKafka()
171 self.msg.connect(config["message"])
172 else:
173 raise NsException(
174 "Invalid configuration param '{}' at '[message]':'driver'".format(
175 config["message"]["driver"]
176 )
177 )
178
179 # TODO load workers to deal with exising database tasks
180
181 self.write_lock = Lock()
182 except (DbException, FsException, MsgException) as e:
183 raise NsException(str(e), http_code=e.http_code)
184
185 def get_assigned_vims(self):
186 return list(self.vims_assigned.keys())
187
188 def stop(self):
189 try:
190 if self.db:
191 self.db.db_disconnect()
192
193 if self.fs:
194 self.fs.fs_disconnect()
195
196 if self.msg:
197 self.msg.disconnect()
198
199 self.write_lock = None
200 except (DbException, FsException, MsgException) as e:
201 raise NsException(str(e), http_code=e.http_code)
202
203 for worker in self.workers:
204 worker.insert_task(("terminate",))
205
206 def _create_worker(self):
207 """
208 Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
209 limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
210 return the index of the assigned worker thread. Worker threads are storead at self.workers
211 """
212 # Look for a thread in idle status
213 worker_id = next(
214 (
215 i
216 for i in range(len(self.workers))
217 if self.workers[i] and self.workers[i].idle
218 ),
219 None,
220 )
221
222 if worker_id is not None:
223 # unset idle status to avoid race conditions
224 self.workers[worker_id].idle = False
225 else:
226 worker_id = len(self.workers)
227
228 if worker_id < self.config["global"]["server.ns_threads"]:
229 # create a new worker
230 self.workers.append(
231 NsWorker(worker_id, self.config, self.plugins, self.db)
232 )
233 self.workers[worker_id].start()
234 else:
235 # reached maximum number of threads, assign VIM to an existing one
236 worker_id = self.next_worker
237 self.next_worker = (self.next_worker + 1) % self.config["global"][
238 "server.ns_threads"
239 ]
240
241 return worker_id
242
243 def assign_vim(self, target_id):
244 with self.write_lock:
245 return self._assign_vim(target_id)
246
247 def _assign_vim(self, target_id):
248 if target_id not in self.vims_assigned:
249 worker_id = self.vims_assigned[target_id] = self._create_worker()
250 self.workers[worker_id].insert_task(("load_vim", target_id))
251
252 def reload_vim(self, target_id):
253 # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
254 # this is because database VIM information is cached for threads working with SDN
255 with self.write_lock:
256 for worker in self.workers:
257 if worker and not worker.idle:
258 worker.insert_task(("reload_vim", target_id))
259
260 def unload_vim(self, target_id):
261 with self.write_lock:
262 return self._unload_vim(target_id)
263
264 def _unload_vim(self, target_id):
265 if target_id in self.vims_assigned:
266 worker_id = self.vims_assigned[target_id]
267 self.workers[worker_id].insert_task(("unload_vim", target_id))
268 del self.vims_assigned[target_id]
269
270 def check_vim(self, target_id):
271 with self.write_lock:
272 if target_id in self.vims_assigned:
273 worker_id = self.vims_assigned[target_id]
274 else:
275 worker_id = self._create_worker()
276
277 worker = self.workers[worker_id]
278 worker.insert_task(("check_vim", target_id))
279
280 def unload_unused_vims(self):
281 with self.write_lock:
282 vims_to_unload = []
283
284 for target_id in self.vims_assigned:
285 if not self.db.get_one(
286 "ro_tasks",
287 q_filter={
288 "target_id": target_id,
289 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
290 },
291 fail_on_empty=False,
292 ):
293 vims_to_unload.append(target_id)
294
295 for target_id in vims_to_unload:
296 self._unload_vim(target_id)
297
298 def _get_cloud_init(self, where):
299 """
300 Not used as cloud init content is provided in the http body. This method reads cloud init from a file
301 :param where: can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
302 :return:
303 """
304 vnfd_id, _, other = where.partition(":")
305 _type, _, name = other.partition(":")
306 vnfd = self.db.get_one("vnfds", {"_id": vnfd_id})
307
308 if _type == "file":
309 base_folder = vnfd["_admin"]["storage"]
310 cloud_init_file = "{}/{}/cloud_init/{}".format(
311 base_folder["folder"], base_folder["pkg-dir"], name
312 )
313
314 if not self.fs:
315 raise NsException(
316 "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
317 cloud_init_file
318 )
319 )
320
321 with self.fs.file_open(cloud_init_file, "r") as ci_file:
322 cloud_init_content = ci_file.read()
323 elif _type == "vdu":
324 cloud_init_content = vnfd["vdu"][int(name)]["cloud-init"]
325 else:
326 raise NsException("Mismatch descriptor for cloud init: {}".format(where))
327
328 return cloud_init_content
329
330 def _parse_jinja2(self, cloud_init_content, params, context):
331 try:
332 env = Environment(undefined=StrictUndefined)
333 template = env.from_string(cloud_init_content)
334
335 return template.render(params or {})
336 except UndefinedError as e:
337 raise NsException(
338 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
339 "inside the 'additionalParamsForVnf' block".format(e, context)
340 )
341 except (TemplateError, TemplateNotFound) as e:
342 raise NsException(
343 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
344 context, e
345 )
346 )
347
348 def _create_db_ro_nsrs(self, nsr_id, now):
349 try:
350 key = rsa.generate_private_key(
351 backend=crypto_default_backend(), public_exponent=65537, key_size=2048
352 )
353 private_key = key.private_bytes(
354 crypto_serialization.Encoding.PEM,
355 crypto_serialization.PrivateFormat.PKCS8,
356 crypto_serialization.NoEncryption(),
357 )
358 public_key = key.public_key().public_bytes(
359 crypto_serialization.Encoding.OpenSSH,
360 crypto_serialization.PublicFormat.OpenSSH,
361 )
362 private_key = private_key.decode("utf8")
363 # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
364 i = private_key.find("\n")
365 private_key = "-----BEGIN RSA PRIVATE KEY-----" + private_key[i:]
366 public_key = public_key.decode("utf8")
367 except Exception as e:
368 raise NsException("Cannot create ssh-keys: {}".format(e))
369
370 schema_version = "1.1"
371 private_key_encrypted = self.db.encrypt(
372 private_key, schema_version=schema_version, salt=nsr_id
373 )
374 db_content = {
375 "_id": nsr_id,
376 "_admin": {
377 "created": now,
378 "modified": now,
379 "schema_version": schema_version,
380 },
381 "public_key": public_key,
382 "private_key": private_key_encrypted,
383 "actions": [],
384 }
385 self.db.create("ro_nsrs", db_content)
386
387 return db_content
388
389 @staticmethod
390 def _create_task(
391 deployment_info: Dict[str, Any],
392 target_id: str,
393 item: str,
394 action: str,
395 target_record: str,
396 target_record_id: str,
397 extra_dict: Dict[str, Any] = None,
398 ) -> Dict[str, Any]:
399 """Function to create task dict from deployment information.
400
401 Args:
402 deployment_info (Dict[str, Any]): [description]
403 target_id (str): [description]
404 item (str): [description]
405 action (str): [description]
406 target_record (str): [description]
407 target_record_id (str): [description]
408 extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
409
410 Returns:
411 Dict[str, Any]: [description]
412 """
413 task = {
414 "target_id": target_id, # it will be removed before pushing at database
415 "action_id": deployment_info.get("action_id"),
416 "nsr_id": deployment_info.get("nsr_id"),
417 "task_id": f"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
418 "status": "SCHEDULED",
419 "action": action,
420 "item": item,
421 "target_record": target_record,
422 "target_record_id": target_record_id,
423 }
424
425 if extra_dict:
426 task.update(extra_dict) # params, find_params, depends_on
427
428 deployment_info["task_index"] = deployment_info.get("task_index", 0) + 1
429
430 return task
431
432 def deploy(self, session, indata, version, nsr_id, *args, **kwargs):
433 self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata))
434 validate_input(indata, deploy_schema)
435 action_id = indata.get("action_id", str(uuid4()))
436 task_index = 0
437 # get current deployment
438 db_nsr_update = {} # update operation on nsrs
439 db_vnfrs_update = {}
440 db_vnfrs = {} # vnf's info indexed by _id
441 nb_ro_tasks = 0 # for logging
442 vdu2cloud_init = indata.get("cloud_init_content") or {}
443 step = ""
444 logging_text = "Task deploy nsr_id={} action_id={} ".format(nsr_id, action_id)
445 self.logger.debug(logging_text + "Enter")
446
447 try:
448 step = "Getting ns and vnfr record from db"
449 db_nsr = self.db.get_one("nsrs", {"_id": nsr_id})
450 db_new_tasks = []
451 tasks_by_target_record_id = {}
452 # read from db: vnf's of this ns
453 step = "Getting vnfrs from db"
454 db_vnfrs_list = self.db.get_list("vnfrs", {"nsr-id-ref": nsr_id})
455
456 if not db_vnfrs_list:
457 raise NsException("Cannot obtain associated VNF for ns")
458
459 for vnfr in db_vnfrs_list:
460 db_vnfrs[vnfr["_id"]] = vnfr
461 db_vnfrs_update[vnfr["_id"]] = {}
462
463 now = time()
464 db_ro_nsr = self.db.get_one("ro_nsrs", {"_id": nsr_id}, fail_on_empty=False)
465
466 if not db_ro_nsr:
467 db_ro_nsr = self._create_db_ro_nsrs(nsr_id, now)
468
469 ro_nsr_public_key = db_ro_nsr["public_key"]
470
471 # check that action_id is not in the list of actions. Suffixed with :index
472 if action_id in db_ro_nsr["actions"]:
473 index = 1
474
475 while True:
476 new_action_id = "{}:{}".format(action_id, index)
477
478 if new_action_id not in db_ro_nsr["actions"]:
479 action_id = new_action_id
480 self.logger.debug(
481 logging_text
482 + "Changing action_id in use to {}".format(action_id)
483 )
484 break
485
486 index += 1
487
488 def _create_ro_task(target_id, task):
489 nonlocal action_id
490 nonlocal task_index
491 nonlocal now
492
493 _id = task["task_id"]
494 db_ro_task = {
495 "_id": _id,
496 "locked_by": None,
497 "locked_at": 0.0,
498 "target_id": target_id,
499 "vim_info": {
500 "created": False,
501 "created_items": None,
502 "vim_id": None,
503 "vim_name": None,
504 "vim_status": None,
505 "vim_details": None,
506 "refresh_at": None,
507 },
508 "modified_at": now,
509 "created_at": now,
510 "to_check_at": now,
511 "tasks": [task],
512 }
513
514 return db_ro_task
515
516 def _process_image_params(target_image, vim_info, target_record_id):
517 find_params = {}
518
519 if target_image.get("image"):
520 find_params["filter_dict"] = {"name": target_image.get("image")}
521
522 if target_image.get("vim_image_id"):
523 find_params["filter_dict"] = {
524 "id": target_image.get("vim_image_id")
525 }
526
527 if target_image.get("image_checksum"):
528 find_params["filter_dict"] = {
529 "checksum": target_image.get("image_checksum")
530 }
531
532 return {"find_params": find_params}
533
534 def _process_flavor_params(target_flavor, vim_info, target_record_id):
535 def _get_resource_allocation_params(quota_descriptor):
536 """
537 read the quota_descriptor from vnfd and fetch the resource allocation properties from the
538 descriptor object
539 :param quota_descriptor: cpu/mem/vif/disk-io quota descriptor
540 :return: quota params for limit, reserve, shares from the descriptor object
541 """
542 quota = {}
543
544 if quota_descriptor.get("limit"):
545 quota["limit"] = int(quota_descriptor["limit"])
546
547 if quota_descriptor.get("reserve"):
548 quota["reserve"] = int(quota_descriptor["reserve"])
549
550 if quota_descriptor.get("shares"):
551 quota["shares"] = int(quota_descriptor["shares"])
552
553 return quota
554
555 flavor_data = {
556 "disk": int(target_flavor["storage-gb"]),
557 "ram": int(target_flavor["memory-mb"]),
558 "vcpus": int(target_flavor["vcpu-count"]),
559 }
560 numa = {}
561 extended = {}
562
563 if target_flavor.get("guest-epa"):
564 extended = {}
565 epa_vcpu_set = False
566
567 if target_flavor["guest-epa"].get("numa-node-policy"):
568 numa_node_policy = target_flavor["guest-epa"].get(
569 "numa-node-policy"
570 )
571
572 if numa_node_policy.get("node"):
573 numa_node = numa_node_policy["node"][0]
574
575 if numa_node.get("num-cores"):
576 numa["cores"] = numa_node["num-cores"]
577 epa_vcpu_set = True
578
579 if numa_node.get("paired-threads"):
580 if numa_node["paired-threads"].get(
581 "num-paired-threads"
582 ):
583 numa["paired-threads"] = int(
584 numa_node["paired-threads"][
585 "num-paired-threads"
586 ]
587 )
588 epa_vcpu_set = True
589
590 if len(
591 numa_node["paired-threads"].get("paired-thread-ids")
592 ):
593 numa["paired-threads-id"] = []
594
595 for pair in numa_node["paired-threads"][
596 "paired-thread-ids"
597 ]:
598 numa["paired-threads-id"].append(
599 (
600 str(pair["thread-a"]),
601 str(pair["thread-b"]),
602 )
603 )
604
605 if numa_node.get("num-threads"):
606 numa["threads"] = int(numa_node["num-threads"])
607 epa_vcpu_set = True
608
609 if numa_node.get("memory-mb"):
610 numa["memory"] = max(
611 int(numa_node["memory-mb"] / 1024), 1
612 )
613
614 if target_flavor["guest-epa"].get("mempage-size"):
615 extended["mempage-size"] = target_flavor["guest-epa"].get(
616 "mempage-size"
617 )
618
619 if (
620 target_flavor["guest-epa"].get("cpu-pinning-policy")
621 and not epa_vcpu_set
622 ):
623 if (
624 target_flavor["guest-epa"]["cpu-pinning-policy"]
625 == "DEDICATED"
626 ):
627 if (
628 target_flavor["guest-epa"].get(
629 "cpu-thread-pinning-policy"
630 )
631 and target_flavor["guest-epa"][
632 "cpu-thread-pinning-policy"
633 ]
634 != "PREFER"
635 ):
636 numa["cores"] = max(flavor_data["vcpus"], 1)
637 else:
638 numa["threads"] = max(flavor_data["vcpus"], 1)
639
640 epa_vcpu_set = True
641
642 if target_flavor["guest-epa"].get("cpu-quota") and not epa_vcpu_set:
643 cpuquota = _get_resource_allocation_params(
644 target_flavor["guest-epa"].get("cpu-quota")
645 )
646
647 if cpuquota:
648 extended["cpu-quota"] = cpuquota
649
650 if target_flavor["guest-epa"].get("mem-quota"):
651 vduquota = _get_resource_allocation_params(
652 target_flavor["guest-epa"].get("mem-quota")
653 )
654
655 if vduquota:
656 extended["mem-quota"] = vduquota
657
658 if target_flavor["guest-epa"].get("disk-io-quota"):
659 diskioquota = _get_resource_allocation_params(
660 target_flavor["guest-epa"].get("disk-io-quota")
661 )
662
663 if diskioquota:
664 extended["disk-io-quota"] = diskioquota
665
666 if target_flavor["guest-epa"].get("vif-quota"):
667 vifquota = _get_resource_allocation_params(
668 target_flavor["guest-epa"].get("vif-quota")
669 )
670
671 if vifquota:
672 extended["vif-quota"] = vifquota
673
674 if numa:
675 extended["numas"] = [numa]
676
677 if extended:
678 flavor_data["extended"] = extended
679
680 extra_dict = {"find_params": {"flavor_data": flavor_data}}
681 flavor_data_name = flavor_data.copy()
682 flavor_data_name["name"] = target_flavor["name"]
683 extra_dict["params"] = {"flavor_data": flavor_data_name}
684
685 return extra_dict
686
687 def _ip_profile_2_ro(ip_profile):
688 if not ip_profile:
689 return None
690
691 ro_ip_profile = {
692 "ip_version": "IPv4"
693 if "v4" in ip_profile.get("ip-version", "ipv4")
694 else "IPv6",
695 "subnet_address": ip_profile.get("subnet-address"),
696 "gateway_address": ip_profile.get("gateway-address"),
697 "dhcp_enabled": ip_profile.get("dhcp-params", {}).get(
698 "enabled", False
699 ),
700 "dhcp_start_address": ip_profile.get("dhcp-params", {}).get(
701 "start-address", None
702 ),
703 "dhcp_count": ip_profile.get("dhcp-params", {}).get("count", None),
704 }
705
706 if ip_profile.get("dns-server"):
707 ro_ip_profile["dns_address"] = ";".join(
708 [v["address"] for v in ip_profile["dns-server"]]
709 )
710
711 if ip_profile.get("security-group"):
712 ro_ip_profile["security_group"] = ip_profile["security-group"]
713
714 return ro_ip_profile
715
716 def _process_net_params(target_vld, vim_info, target_record_id):
717 nonlocal indata
718 extra_dict = {}
719
720 if vim_info.get("sdn"):
721 # vnf_preffix = "vnfrs:{}".format(vnfr_id)
722 # ns_preffix = "nsrs:{}".format(nsr_id)
723 # remove the ending ".sdn
724 vld_target_record_id, _, _ = target_record_id.rpartition(".")
725 extra_dict["params"] = {
726 k: vim_info[k]
727 for k in ("sdn-ports", "target_vim", "vlds", "type")
728 if vim_info.get(k)
729 }
730
731 # TODO needed to add target_id in the dependency.
732 if vim_info.get("target_vim"):
733 extra_dict["depends_on"] = [
734 vim_info.get("target_vim") + " " + vld_target_record_id
735 ]
736
737 return extra_dict
738
739 if vim_info.get("vim_network_name"):
740 extra_dict["find_params"] = {
741 "filter_dict": {"name": vim_info.get("vim_network_name")}
742 }
743 elif vim_info.get("vim_network_id"):
744 extra_dict["find_params"] = {
745 "filter_dict": {"id": vim_info.get("vim_network_id")}
746 }
747 elif target_vld.get("mgmt-network"):
748 extra_dict["find_params"] = {"mgmt": True, "name": target_vld["id"]}
749 else:
750 # create
751 extra_dict["params"] = {
752 "net_name": "{}-{}".format(
753 indata["name"][:16],
754 target_vld.get("name", target_vld["id"])[:16],
755 ),
756 "ip_profile": _ip_profile_2_ro(vim_info.get("ip_profile")),
757 "provider_network_profile": vim_info.get("provider_network"),
758 }
759
760 if not target_vld.get("underlay"):
761 extra_dict["params"]["net_type"] = "bridge"
762 else:
763 extra_dict["params"]["net_type"] = (
764 "ptp" if target_vld.get("type") == "ELINE" else "data"
765 )
766
767 return extra_dict
768
769 def _process_vdu_params(target_vdu, vim_info, target_record_id):
770 nonlocal vnfr_id
771 nonlocal nsr_id
772 nonlocal indata
773 nonlocal vnfr
774 nonlocal vdu2cloud_init
775 nonlocal tasks_by_target_record_id
776
777 vnf_preffix = "vnfrs:{}".format(vnfr_id)
778 ns_preffix = "nsrs:{}".format(nsr_id)
779 image_text = ns_preffix + ":image." + target_vdu["ns-image-id"]
780 flavor_text = ns_preffix + ":flavor." + target_vdu["ns-flavor-id"]
781 extra_dict = {"depends_on": [image_text, flavor_text]}
782 net_list = []
783
784 for iface_index, interface in enumerate(target_vdu["interfaces"]):
785 if interface.get("ns-vld-id"):
786 net_text = ns_preffix + ":vld." + interface["ns-vld-id"]
787 elif interface.get("vnf-vld-id"):
788 net_text = vnf_preffix + ":vld." + interface["vnf-vld-id"]
789 else:
790 self.logger.error(
791 "Interface {} from vdu {} not connected to any vld".format(
792 iface_index, target_vdu["vdu-name"]
793 )
794 )
795
796 continue # interface not connected to any vld
797
798 extra_dict["depends_on"].append(net_text)
799
800 if "port-security-enabled" in interface:
801 interface["port_security"] = interface.pop(
802 "port-security-enabled"
803 )
804
805 if "port-security-disable-strategy" in interface:
806 interface["port_security_disable_strategy"] = interface.pop(
807 "port-security-disable-strategy"
808 )
809
810 net_item = {
811 x: v
812 for x, v in interface.items()
813 if x
814 in (
815 "name",
816 "vpci",
817 "port_security",
818 "port_security_disable_strategy",
819 "floating_ip",
820 )
821 }
822 net_item["net_id"] = "TASK-" + net_text
823 net_item["type"] = "virtual"
824
825 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
826 # TODO floating_ip: True/False (or it can be None)
827 if interface.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
828 # mark the net create task as type data
829 if deep_get(
830 tasks_by_target_record_id, net_text, "params", "net_type"
831 ):
832 tasks_by_target_record_id[net_text]["params"][
833 "net_type"
834 ] = "data"
835
836 net_item["use"] = "data"
837 net_item["model"] = interface["type"]
838 net_item["type"] = interface["type"]
839 elif (
840 interface.get("type") == "OM-MGMT"
841 or interface.get("mgmt-interface")
842 or interface.get("mgmt-vnf")
843 ):
844 net_item["use"] = "mgmt"
845 else:
846 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
847 net_item["use"] = "bridge"
848 net_item["model"] = interface.get("type")
849
850 if interface.get("ip-address"):
851 net_item["ip_address"] = interface["ip-address"]
852
853 if interface.get("mac-address"):
854 net_item["mac_address"] = interface["mac-address"]
855
856 net_list.append(net_item)
857
858 if interface.get("mgmt-vnf"):
859 extra_dict["mgmt_vnf_interface"] = iface_index
860 elif interface.get("mgmt-interface"):
861 extra_dict["mgmt_vdu_interface"] = iface_index
862
863 # cloud config
864 cloud_config = {}
865
866 if target_vdu.get("cloud-init"):
867 if target_vdu["cloud-init"] not in vdu2cloud_init:
868 vdu2cloud_init[target_vdu["cloud-init"]] = self._get_cloud_init(
869 target_vdu["cloud-init"]
870 )
871
872 cloud_content_ = vdu2cloud_init[target_vdu["cloud-init"]]
873 cloud_config["user-data"] = self._parse_jinja2(
874 cloud_content_,
875 target_vdu.get("additionalParams"),
876 target_vdu["cloud-init"],
877 )
878
879 if target_vdu.get("boot-data-drive"):
880 cloud_config["boot-data-drive"] = target_vdu.get("boot-data-drive")
881
882 ssh_keys = []
883
884 if target_vdu.get("ssh-keys"):
885 ssh_keys += target_vdu.get("ssh-keys")
886
887 if target_vdu.get("ssh-access-required"):
888 ssh_keys.append(ro_nsr_public_key)
889
890 if ssh_keys:
891 cloud_config["key-pairs"] = ssh_keys
892
893 disk_list = None
894 if target_vdu.get("virtual-storages"):
895 disk_list = [
896 {"size": disk["size-of-storage"]}
897 for disk in target_vdu["virtual-storages"]
898 if disk.get("type-of-storage")
899 == "persistent-storage:persistent-storage"
900 ]
901
902 extra_dict["params"] = {
903 "name": "{}-{}-{}-{}".format(
904 indata["name"][:16],
905 vnfr["member-vnf-index-ref"][:16],
906 target_vdu["vdu-name"][:32],
907 target_vdu.get("count-index") or 0,
908 ),
909 "description": target_vdu["vdu-name"],
910 "start": True,
911 "image_id": "TASK-" + image_text,
912 "flavor_id": "TASK-" + flavor_text,
913 "net_list": net_list,
914 "cloud_config": cloud_config or None,
915 "disk_list": disk_list,
916 "availability_zone_index": None, # TODO
917 "availability_zone_list": None, # TODO
918 }
919
920 return extra_dict
921
922 def _process_items(
923 target_list,
924 existing_list,
925 db_record,
926 db_update,
927 db_path,
928 item,
929 process_params,
930 ):
931 nonlocal db_new_tasks
932 nonlocal tasks_by_target_record_id
933 nonlocal action_id
934 nonlocal nsr_id
935 nonlocal task_index
936
937 # ensure all the target_list elements has an "id". If not assign the index as id
938 for target_index, tl in enumerate(target_list):
939 if tl and not tl.get("id"):
940 tl["id"] = str(target_index)
941
942 # step 1 items (networks,vdus,...) to be deleted/updated
943 for item_index, existing_item in enumerate(existing_list):
944 target_item = next(
945 (t for t in target_list if t["id"] == existing_item["id"]), None
946 )
947
948 for target_vim, existing_viminfo in existing_item.get(
949 "vim_info", {}
950 ).items():
951 if existing_viminfo is None:
952 continue
953
954 if target_item:
955 target_viminfo = target_item.get("vim_info", {}).get(
956 target_vim
957 )
958 else:
959 target_viminfo = None
960
961 if target_viminfo is None:
962 # must be deleted
963 self._assign_vim(target_vim)
964 target_record_id = "{}.{}".format(
965 db_record, existing_item["id"]
966 )
967 item_ = item
968
969 if target_vim.startswith("sdn"):
970 # item must be sdn-net instead of net if target_vim is a sdn
971 item_ = "sdn_net"
972 target_record_id += ".sdn"
973
974 deployment_info = {
975 "action_id": action_id,
976 "nsr_id": nsr_id,
977 "task_index": task_index,
978 }
979
980 task = Ns._create_task(
981 deployment_info=deployment_info,
982 target_id=target_vim,
983 item=item_,
984 action="DELETE",
985 target_record=f"{db_record}.{item_index}.vim_info.{target_vim}",
986 target_record_id=target_record_id,
987 )
988
989 task_index = deployment_info.get("task_index")
990
991 tasks_by_target_record_id[target_record_id] = task
992 db_new_tasks.append(task)
993 # TODO delete
994 # TODO check one by one the vims to be created/deleted
995
996 # step 2 items (networks,vdus,...) to be created
997 for target_item in target_list:
998 item_index = -1
999
1000 for item_index, existing_item in enumerate(existing_list):
1001 if existing_item["id"] == target_item["id"]:
1002 break
1003 else:
1004 item_index += 1
1005 db_update[db_path + ".{}".format(item_index)] = target_item
1006 existing_list.append(target_item)
1007 existing_item = None
1008
1009 for target_vim, target_viminfo in target_item.get(
1010 "vim_info", {}
1011 ).items():
1012 existing_viminfo = None
1013
1014 if existing_item:
1015 existing_viminfo = existing_item.get("vim_info", {}).get(
1016 target_vim
1017 )
1018
1019 # TODO check if different. Delete and create???
1020 # TODO delete if not exist
1021 if existing_viminfo is not None:
1022 continue
1023
1024 target_record_id = "{}.{}".format(db_record, target_item["id"])
1025 item_ = item
1026
1027 if target_vim.startswith("sdn"):
1028 # item must be sdn-net instead of net if target_vim is a sdn
1029 item_ = "sdn_net"
1030 target_record_id += ".sdn"
1031
1032 extra_dict = process_params(
1033 target_item, target_viminfo, target_record_id
1034 )
1035 self._assign_vim(target_vim)
1036
1037 deployment_info = {
1038 "action_id": action_id,
1039 "nsr_id": nsr_id,
1040 "task_index": task_index,
1041 }
1042
1043 task = Ns._create_task(
1044 deployment_info=deployment_info,
1045 target_id=target_vim,
1046 item=item_,
1047 action="CREATE",
1048 target_record=f"{db_record}.{item_index}.vim_info.{target_vim}",
1049 target_record_id=target_record_id,
1050 extra_dict=extra_dict,
1051 )
1052
1053 task_index = deployment_info.get("task_index")
1054
1055 tasks_by_target_record_id[target_record_id] = task
1056 db_new_tasks.append(task)
1057
1058 if target_item.get("common_id"):
1059 task["common_id"] = target_item["common_id"]
1060
1061 db_update[db_path + ".{}".format(item_index)] = target_item
1062
1063 def _process_action(indata):
1064 nonlocal db_new_tasks
1065 nonlocal action_id
1066 nonlocal nsr_id
1067 nonlocal task_index
1068 nonlocal db_vnfrs
1069 nonlocal db_ro_nsr
1070
1071 if indata["action"]["action"] == "inject_ssh_key":
1072 key = indata["action"].get("key")
1073 user = indata["action"].get("user")
1074 password = indata["action"].get("password")
1075
1076 for vnf in indata.get("vnf", ()):
1077 if vnf["_id"] not in db_vnfrs:
1078 raise NsException("Invalid vnf={}".format(vnf["_id"]))
1079
1080 db_vnfr = db_vnfrs[vnf["_id"]]
1081
1082 for target_vdu in vnf.get("vdur", ()):
1083 vdu_index, vdur = next(
1084 (
1085 i_v
1086 for i_v in enumerate(db_vnfr["vdur"])
1087 if i_v[1]["id"] == target_vdu["id"]
1088 ),
1089 (None, None),
1090 )
1091
1092 if not vdur:
1093 raise NsException(
1094 "Invalid vdu vnf={}.{}".format(
1095 vnf["_id"], target_vdu["id"]
1096 )
1097 )
1098
1099 target_vim, vim_info = next(
1100 k_v for k_v in vdur["vim_info"].items()
1101 )
1102 self._assign_vim(target_vim)
1103 target_record = "vnfrs:{}:vdur.{}.ssh_keys".format(
1104 vnf["_id"], vdu_index
1105 )
1106 extra_dict = {
1107 "depends_on": [
1108 "vnfrs:{}:vdur.{}".format(vnf["_id"], vdur["id"])
1109 ],
1110 "params": {
1111 "ip_address": vdur.get("ip-address"),
1112 "user": user,
1113 "key": key,
1114 "password": password,
1115 "private_key": db_ro_nsr["private_key"],
1116 "salt": db_ro_nsr["_id"],
1117 "schema_version": db_ro_nsr["_admin"][
1118 "schema_version"
1119 ],
1120 },
1121 }
1122
1123 deployment_info = {
1124 "action_id": action_id,
1125 "nsr_id": nsr_id,
1126 "task_index": task_index,
1127 }
1128
1129 task = Ns._create_task(
1130 deployment_info=deployment_info,
1131 target_id=target_vim,
1132 item="vdu",
1133 action="EXEC",
1134 target_record=target_record,
1135 target_record_id=None,
1136 extra_dict=extra_dict,
1137 )
1138
1139 task_index = deployment_info.get("task_index")
1140
1141 db_new_tasks.append(task)
1142
1143 with self.write_lock:
1144 if indata.get("action"):
1145 _process_action(indata)
1146 else:
1147 # compute network differences
1148 # NS.vld
1149 step = "process NS VLDs"
1150 _process_items(
1151 target_list=indata["ns"]["vld"] or [],
1152 existing_list=db_nsr.get("vld") or [],
1153 db_record="nsrs:{}:vld".format(nsr_id),
1154 db_update=db_nsr_update,
1155 db_path="vld",
1156 item="net",
1157 process_params=_process_net_params,
1158 )
1159
1160 step = "process NS images"
1161 _process_items(
1162 target_list=indata.get("image") or [],
1163 existing_list=db_nsr.get("image") or [],
1164 db_record="nsrs:{}:image".format(nsr_id),
1165 db_update=db_nsr_update,
1166 db_path="image",
1167 item="image",
1168 process_params=_process_image_params,
1169 )
1170
1171 step = "process NS flavors"
1172 _process_items(
1173 target_list=indata.get("flavor") or [],
1174 existing_list=db_nsr.get("flavor") or [],
1175 db_record="nsrs:{}:flavor".format(nsr_id),
1176 db_update=db_nsr_update,
1177 db_path="flavor",
1178 item="flavor",
1179 process_params=_process_flavor_params,
1180 )
1181
1182 # VNF.vld
1183 for vnfr_id, vnfr in db_vnfrs.items():
1184 # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
1185 step = "process VNF={} VLDs".format(vnfr_id)
1186 target_vnf = next(
1187 (
1188 vnf
1189 for vnf in indata.get("vnf", ())
1190 if vnf["_id"] == vnfr_id
1191 ),
1192 None,
1193 )
1194 target_list = target_vnf.get("vld") if target_vnf else None
1195 _process_items(
1196 target_list=target_list or [],
1197 existing_list=vnfr.get("vld") or [],
1198 db_record="vnfrs:{}:vld".format(vnfr_id),
1199 db_update=db_vnfrs_update[vnfr["_id"]],
1200 db_path="vld",
1201 item="net",
1202 process_params=_process_net_params,
1203 )
1204
1205 target_list = target_vnf.get("vdur") if target_vnf else None
1206 step = "process VNF={} VDUs".format(vnfr_id)
1207 _process_items(
1208 target_list=target_list or [],
1209 existing_list=vnfr.get("vdur") or [],
1210 db_record="vnfrs:{}:vdur".format(vnfr_id),
1211 db_update=db_vnfrs_update[vnfr["_id"]],
1212 db_path="vdur",
1213 item="vdu",
1214 process_params=_process_vdu_params,
1215 )
1216
1217 for db_task in db_new_tasks:
1218 step = "Updating database, Appending tasks to ro_tasks"
1219 target_id = db_task.pop("target_id")
1220 common_id = db_task.get("common_id")
1221
1222 if common_id:
1223 if self.db.set_one(
1224 "ro_tasks",
1225 q_filter={
1226 "target_id": target_id,
1227 "tasks.common_id": common_id,
1228 },
1229 update_dict={"to_check_at": now, "modified_at": now},
1230 push={"tasks": db_task},
1231 fail_on_empty=False,
1232 ):
1233 continue
1234
1235 if not self.db.set_one(
1236 "ro_tasks",
1237 q_filter={
1238 "target_id": target_id,
1239 "tasks.target_record": db_task["target_record"],
1240 },
1241 update_dict={"to_check_at": now, "modified_at": now},
1242 push={"tasks": db_task},
1243 fail_on_empty=False,
1244 ):
1245 # Create a ro_task
1246 step = "Updating database, Creating ro_tasks"
1247 db_ro_task = _create_ro_task(target_id, db_task)
1248 nb_ro_tasks += 1
1249 self.db.create("ro_tasks", db_ro_task)
1250
1251 step = "Updating database, nsrs"
1252 if db_nsr_update:
1253 self.db.set_one("nsrs", {"_id": nsr_id}, db_nsr_update)
1254
1255 for vnfr_id, db_vnfr_update in db_vnfrs_update.items():
1256 if db_vnfr_update:
1257 step = "Updating database, vnfrs={}".format(vnfr_id)
1258 self.db.set_one("vnfrs", {"_id": vnfr_id}, db_vnfr_update)
1259
1260 self.logger.debug(
1261 logging_text
1262 + "Exit. Created {} ro_tasks; {} tasks".format(
1263 nb_ro_tasks, len(db_new_tasks)
1264 )
1265 )
1266
1267 return (
1268 {"status": "ok", "nsr_id": nsr_id, "action_id": action_id},
1269 action_id,
1270 True,
1271 )
1272 except Exception as e:
1273 if isinstance(e, (DbException, NsException)):
1274 self.logger.error(
1275 logging_text + "Exit Exception while '{}': {}".format(step, e)
1276 )
1277 else:
1278 e = traceback_format_exc()
1279 self.logger.critical(
1280 logging_text + "Exit Exception while '{}': {}".format(step, e),
1281 exc_info=True,
1282 )
1283
1284 raise NsException(e)
1285
1286 def delete(self, session, indata, version, nsr_id, *args, **kwargs):
1287 self.logger.debug("ns.delete version={} nsr_id={}".format(version, nsr_id))
1288 # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
1289
1290 with self.write_lock:
1291 try:
1292 NsWorker.delete_db_tasks(self.db, nsr_id, None)
1293 except NsWorkerException as e:
1294 raise NsException(e)
1295
1296 return None, None, True
1297
1298 def status(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
1299 # self.logger.debug("ns.status version={} nsr_id={}, action_id={} indata={}"
1300 # .format(version, nsr_id, action_id, indata))
1301 task_list = []
1302 done = 0
1303 total = 0
1304 ro_tasks = self.db.get_list("ro_tasks", {"tasks.action_id": action_id})
1305 global_status = "DONE"
1306 details = []
1307
1308 for ro_task in ro_tasks:
1309 for task in ro_task["tasks"]:
1310 if task and task["action_id"] == action_id:
1311 task_list.append(task)
1312 total += 1
1313
1314 if task["status"] == "FAILED":
1315 global_status = "FAILED"
1316 error_text = "Error at {} {}: {}".format(
1317 task["action"].lower(),
1318 task["item"],
1319 ro_task["vim_info"].get("vim_details") or "unknown",
1320 )
1321 details.append(error_text)
1322 elif task["status"] in ("SCHEDULED", "BUILD"):
1323 if global_status != "FAILED":
1324 global_status = "BUILD"
1325 else:
1326 done += 1
1327
1328 return_data = {
1329 "status": global_status,
1330 "details": ". ".join(details)
1331 if details
1332 else "progress {}/{}".format(done, total),
1333 "nsr_id": nsr_id,
1334 "action_id": action_id,
1335 "tasks": task_list,
1336 }
1337
1338 return return_data, None, True
1339
1340 def cancel(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
1341 print(
1342 "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
1343 session, indata, version, nsr_id, action_id
1344 )
1345 )
1346
1347 return None, None, True
1348
1349 def get_deploy(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
1350 nsrs = self.db.get_list("nsrs", {})
1351 return_data = []
1352
1353 for ns in nsrs:
1354 return_data.append({"_id": ns["_id"], "name": ns["name"]})
1355
1356 return return_data, None, True
1357
1358 def get_actions(self, session, indata, version, nsr_id, action_id, *args, **kwargs):
1359 ro_tasks = self.db.get_list("ro_tasks", {"tasks.nsr_id": nsr_id})
1360 return_data = []
1361
1362 for ro_task in ro_tasks:
1363 for task in ro_task["tasks"]:
1364 if task["action_id"] not in return_data:
1365 return_data.append(task["action_id"])
1366
1367 return return_data, None, True