1 # -*- coding: utf-8 -*-
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
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
19 from http
import HTTPStatus
20 from itertools
import product
22 from random
import choice
as random_choice
23 from threading
import Lock
25 from traceback
import format_exc
as traceback_format_exc
26 from typing
import Any
, Dict
, List
, Optional
, Tuple
, Type
27 from uuid
import uuid4
29 from cryptography
.hazmat
.backends
import default_backend
as crypto_default_backend
30 from cryptography
.hazmat
.primitives
import serialization
as crypto_serialization
31 from cryptography
.hazmat
.primitives
.asymmetric
import rsa
40 from osm_common
import (
47 version
as common_version
,
49 from osm_common
.dbbase
import DbBase
, DbException
50 from osm_common
.fsbase
import FsBase
, FsException
51 from osm_common
.msgbase
import MsgException
52 from osm_ng_ro
.ns_thread
import deep_get
, NsWorker
, NsWorkerException
53 from osm_ng_ro
.validation
import deploy_schema
, validate_input
56 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
57 min_common_version
= "0.1.16"
60 class NsException(Exception):
61 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
62 self
.http_code
= http_code
63 super(Exception, self
).__init
__(message
)
68 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
69 will provide a random one
72 # Try getting docker id. If fails, get pid
74 with
open("/proc/self/cgroup", "r") as f
:
75 text_id_
= f
.readline()
76 _
, _
, text_id
= text_id_
.rpartition("/")
77 text_id
= text_id
.replace("\n", "")[:12]
81 except Exception as error
:
82 logging
.exception(f
"{error} occured while getting process id")
85 return "".join(random_choice("0123456789abcdef") for _
in range(12))
89 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
92 for point
in v
.split("."):
93 filled
.append(point
.zfill(8))
104 # self.operations = None
106 # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
107 # If done now it will not be linked to parent not getting its handler and level
109 self
.write_lock
= None
110 self
.vims_assigned
= {}
114 self
.process_params_function_map
= {
115 "net": Ns
._process
_net
_params
,
116 "image": Ns
._process
_image
_params
,
117 "flavor": Ns
._process
_flavor
_params
,
118 "vdu": Ns
._process
_vdu
_params
,
119 "affinity-or-anti-affinity-group": Ns
._process
_affinity
_group
_params
,
120 "shared-volumes": Ns
._process
_shared
_volumes
_params
,
127 "affinity-or-anti-affinity-group": "affinity-or-anti-affinity-group",
128 "shared-volumes": "shared-volumes",
131 def init_db(self
, target_version
):
134 def start(self
, config
):
136 Connect to database, filesystem storage, and messaging
137 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
138 :param config: Configuration of db, storage, etc
142 self
.config
["process_id"] = get_process_id() # used for HA identity
143 self
.logger
= logging
.getLogger("ro.ns")
145 # check right version of common
146 if versiontuple(common_version
) < versiontuple(min_common_version
):
148 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
149 common_version
, min_common_version
155 if config
["database"]["driver"] == "mongo":
156 self
.db
= dbmongo
.DbMongo()
157 self
.db
.db_connect(config
["database"])
158 elif config
["database"]["driver"] == "memory":
159 self
.db
= dbmemory
.DbMemory()
160 self
.db
.db_connect(config
["database"])
163 "Invalid configuration param '{}' at '[database]':'driver'".format(
164 config
["database"]["driver"]
169 if config
["storage"]["driver"] == "local":
170 self
.fs
= fslocal
.FsLocal()
171 self
.fs
.fs_connect(config
["storage"])
172 elif config
["storage"]["driver"] == "mongo":
173 self
.fs
= fsmongo
.FsMongo()
174 self
.fs
.fs_connect(config
["storage"])
175 elif config
["storage"]["driver"] is None:
179 "Invalid configuration param '{}' at '[storage]':'driver'".format(
180 config
["storage"]["driver"]
185 if config
["message"]["driver"] == "local":
186 self
.msg
= msglocal
.MsgLocal()
187 self
.msg
.connect(config
["message"])
188 elif config
["message"]["driver"] == "kafka":
189 self
.msg
= msgkafka
.MsgKafka()
190 self
.msg
.connect(config
["message"])
193 "Invalid configuration param '{}' at '[message]':'driver'".format(
194 config
["message"]["driver"]
198 # TODO load workers to deal with exising database tasks
200 self
.write_lock
= Lock()
201 except (DbException
, FsException
, MsgException
) as e
:
202 raise NsException(str(e
), http_code
=e
.http_code
)
204 def get_assigned_vims(self
):
205 return list(self
.vims_assigned
.keys())
210 self
.db
.db_disconnect()
213 self
.fs
.fs_disconnect()
216 self
.msg
.disconnect()
218 self
.write_lock
= None
219 except (DbException
, FsException
, MsgException
) as e
:
220 raise NsException(str(e
), http_code
=e
.http_code
)
222 for worker
in self
.workers
:
223 worker
.insert_task(("terminate",))
225 def _create_worker(self
):
227 Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
228 limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
229 return the index of the assigned worker thread. Worker threads are storead at self.workers
231 # Look for a thread in idle status
235 for i
in range(len(self
.workers
))
236 if self
.workers
[i
] and self
.workers
[i
].idle
241 if worker_id
is not None:
242 # unset idle status to avoid race conditions
243 self
.workers
[worker_id
].idle
= False
245 worker_id
= len(self
.workers
)
247 if worker_id
< self
.config
["global"]["server.ns_threads"]:
248 # create a new worker
250 NsWorker(worker_id
, self
.config
, self
.plugins
, self
.db
)
252 self
.workers
[worker_id
].start()
254 # reached maximum number of threads, assign VIM to an existing one
255 worker_id
= self
.next_worker
256 self
.next_worker
= (self
.next_worker
+ 1) % self
.config
["global"][
262 def assign_vim(self
, target_id
):
263 with self
.write_lock
:
264 return self
._assign
_vim
(target_id
)
266 def _assign_vim(self
, target_id
):
267 if target_id
not in self
.vims_assigned
:
268 worker_id
= self
.vims_assigned
[target_id
] = self
._create
_worker
()
269 self
.workers
[worker_id
].insert_task(("load_vim", target_id
))
271 def reload_vim(self
, target_id
):
272 # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
273 # this is because database VIM information is cached for threads working with SDN
274 with self
.write_lock
:
275 for worker
in self
.workers
:
276 if worker
and not worker
.idle
:
277 worker
.insert_task(("reload_vim", target_id
))
279 def unload_vim(self
, target_id
):
280 with self
.write_lock
:
281 return self
._unload
_vim
(target_id
)
283 def _unload_vim(self
, target_id
):
284 if target_id
in self
.vims_assigned
:
285 worker_id
= self
.vims_assigned
[target_id
]
286 self
.workers
[worker_id
].insert_task(("unload_vim", target_id
))
287 del self
.vims_assigned
[target_id
]
289 def check_vim(self
, target_id
):
290 with self
.write_lock
:
291 if target_id
in self
.vims_assigned
:
292 worker_id
= self
.vims_assigned
[target_id
]
294 worker_id
= self
._create
_worker
()
296 worker
= self
.workers
[worker_id
]
297 worker
.insert_task(("check_vim", target_id
))
299 def unload_unused_vims(self
):
300 with self
.write_lock
:
303 for target_id
in self
.vims_assigned
:
304 if not self
.db
.get_one(
307 "target_id": target_id
,
308 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
312 vims_to_unload
.append(target_id
)
314 for target_id
in vims_to_unload
:
315 self
._unload
_vim
(target_id
)
323 """This method reads cloud init from a file.
325 Note: Not used as cloud init content is provided in the http body.
328 db (Type[DbBase]): [description]
329 fs (Type[FsBase]): [description]
330 location (str): can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
333 NsException: [description]
334 NsException: [description]
339 vnfd_id
, _
, other
= location
.partition(":")
340 _type
, _
, name
= other
.partition(":")
341 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
344 base_folder
= vnfd
["_admin"]["storage"]
345 cloud_init_file
= "{}/{}/cloud_init/{}".format(
346 base_folder
["folder"], base_folder
["pkg-dir"], name
351 "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
356 with fs
.file_open(cloud_init_file
, "r") as ci_file
:
357 cloud_init_content
= ci_file
.read()
359 cloud_init_content
= vnfd
["vdu"][int(name
)]["cloud-init"]
361 raise NsException("Mismatch descriptor for cloud init: {}".format(location
))
363 return cloud_init_content
367 cloud_init_content
: str,
368 params
: Dict
[str, Any
],
371 """Function that processes the cloud init to replace Jinja2 encoded parameters.
374 cloud_init_content (str): [description]
375 params (Dict[str, Any]): [description]
376 context (str): [description]
379 NsException: [description]
380 NsException: [description]
387 undefined
=StrictUndefined
,
388 autoescape
=select_autoescape(default_for_string
=True, default
=True),
390 template
= env
.from_string(cloud_init_content
)
392 return template
.render(params
or {})
393 except UndefinedError
as e
:
395 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
396 "inside the 'additionalParamsForVnf' block".format(e
, context
)
398 except (TemplateError
, TemplateNotFound
) as e
:
400 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
405 def _create_db_ro_nsrs(self
, nsr_id
, now
):
407 key
= rsa
.generate_private_key(
408 backend
=crypto_default_backend(), public_exponent
=65537, key_size
=2048
410 private_key
= key
.private_bytes(
411 crypto_serialization
.Encoding
.PEM
,
412 crypto_serialization
.PrivateFormat
.PKCS8
,
413 crypto_serialization
.NoEncryption(),
415 public_key
= key
.public_key().public_bytes(
416 crypto_serialization
.Encoding
.OpenSSH
,
417 crypto_serialization
.PublicFormat
.OpenSSH
,
419 private_key
= private_key
.decode("utf8")
420 # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
421 i
= private_key
.find("\n")
422 private_key
= "-----BEGIN RSA PRIVATE KEY-----" + private_key
[i
:]
423 public_key
= public_key
.decode("utf8")
424 except Exception as e
:
425 raise NsException("Cannot create ssh-keys: {}".format(e
))
427 schema_version
= "1.1"
428 private_key_encrypted
= self
.db
.encrypt(
429 private_key
, schema_version
=schema_version
, salt
=nsr_id
436 "schema_version": schema_version
,
438 "public_key": public_key
,
439 "private_key": private_key_encrypted
,
442 self
.db
.create("ro_nsrs", db_content
)
448 deployment_info
: Dict
[str, Any
],
453 target_record_id
: str,
454 extra_dict
: Dict
[str, Any
] = None,
456 """Function to create task dict from deployment information.
459 deployment_info (Dict[str, Any]): [description]
460 target_id (str): [description]
461 item (str): [description]
462 action (str): [description]
463 target_record (str): [description]
464 target_record_id (str): [description]
465 extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
468 Dict[str, Any]: [description]
471 "target_id": target_id
, # it will be removed before pushing at database
472 "action_id": deployment_info
.get("action_id"),
473 "nsr_id": deployment_info
.get("nsr_id"),
474 "task_id": f
"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
475 "status": "SCHEDULED",
478 "target_record": target_record
,
479 "target_record_id": target_record_id
,
483 task
.update(extra_dict
) # params, find_params, depends_on
485 deployment_info
["task_index"] = deployment_info
.get("task_index", 0) + 1
492 task
: Dict
[str, Any
],
494 """Function to create an RO task from task information.
497 target_id (str): [description]
498 task (Dict[str, Any]): [description]
501 Dict[str, Any]: [description]
505 _id
= task
.get("task_id")
510 "target_id": target_id
,
513 "created_items": None,
530 def _process_image_params(
531 target_image
: Dict
[str, Any
],
532 indata
: Dict
[str, Any
],
533 vim_info
: Dict
[str, Any
],
534 target_record_id
: str,
535 **kwargs
: Dict
[str, Any
],
537 """Function to process VDU image parameters.
540 target_image (Dict[str, Any]): [description]
541 indata (Dict[str, Any]): [description]
542 vim_info (Dict[str, Any]): [description]
543 target_record_id (str): [description]
546 Dict[str, Any]: [description]
550 if target_image
.get("image"):
551 find_params
["filter_dict"] = {"name": target_image
.get("image")}
553 if target_image
.get("vim_image_id"):
554 find_params
["filter_dict"] = {"id": target_image
.get("vim_image_id")}
556 if target_image
.get("image_checksum"):
557 find_params
["filter_dict"] = {
558 "checksum": target_image
.get("image_checksum")
561 return {"find_params": find_params
}
564 def _get_resource_allocation_params(
565 quota_descriptor
: Dict
[str, Any
],
567 """Read the quota_descriptor from vnfd and fetch the resource allocation properties from the
571 quota_descriptor (Dict[str, Any]): cpu/mem/vif/disk-io quota descriptor
574 Dict[str, Any]: quota params for limit, reserve, shares from the descriptor object
578 if quota_descriptor
.get("limit"):
579 quota
["limit"] = int(quota_descriptor
["limit"])
581 if quota_descriptor
.get("reserve"):
582 quota
["reserve"] = int(quota_descriptor
["reserve"])
584 if quota_descriptor
.get("shares"):
585 quota
["shares"] = int(quota_descriptor
["shares"])
590 def _process_guest_epa_quota_params(
591 guest_epa_quota
: Dict
[str, Any
],
594 """Function to extract the guest epa quota parameters.
597 guest_epa_quota (Dict[str, Any]): [description]
598 epa_vcpu_set (bool): [description]
601 Dict[str, Any]: [description]
605 if guest_epa_quota
.get("cpu-quota") and not epa_vcpu_set
:
606 cpuquota
= Ns
._get
_resource
_allocation
_params
(
607 guest_epa_quota
.get("cpu-quota")
611 result
["cpu-quota"] = cpuquota
613 if guest_epa_quota
.get("mem-quota"):
614 vduquota
= Ns
._get
_resource
_allocation
_params
(
615 guest_epa_quota
.get("mem-quota")
619 result
["mem-quota"] = vduquota
621 if guest_epa_quota
.get("disk-io-quota"):
622 diskioquota
= Ns
._get
_resource
_allocation
_params
(
623 guest_epa_quota
.get("disk-io-quota")
627 result
["disk-io-quota"] = diskioquota
629 if guest_epa_quota
.get("vif-quota"):
630 vifquota
= Ns
._get
_resource
_allocation
_params
(
631 guest_epa_quota
.get("vif-quota")
635 result
["vif-quota"] = vifquota
640 def _process_guest_epa_numa_params(
641 guest_epa_quota
: Dict
[str, Any
],
642 ) -> Tuple
[Dict
[str, Any
], bool]:
646 guest_epa_quota (Dict[str, Any]): [description]
649 Tuple[Dict[str, Any], bool]: [description]
655 if guest_epa_quota
.get("numa-node-policy"):
656 numa_node_policy
= guest_epa_quota
.get("numa-node-policy")
658 if numa_node_policy
.get("node"):
659 for numa_node
in numa_node_policy
["node"]:
661 if numa_node
.get("id"):
662 numa
["id"] = int(numa_node
["id"])
664 if numa_node
.get("vcpu"):
665 for vcpu
in numa_node
.get("vcpu"):
666 vcpu_id
= int(vcpu
.get("id"))
667 vcpu_list
.append(vcpu_id
)
668 numa
["vcpu"] = vcpu_list
670 if numa_node
.get("num-cores"):
671 numa
["cores"] = numa_node
["num-cores"]
674 paired_threads
= numa_node
.get("paired-threads", {})
675 if paired_threads
.get("num-paired-threads"):
676 numa
["paired_threads"] = int(
677 numa_node
["paired-threads"]["num-paired-threads"]
681 if paired_threads
.get("paired-thread-ids"):
682 numa
["paired-threads-id"] = []
684 for pair
in paired_threads
["paired-thread-ids"]:
685 numa
["paired-threads-id"].append(
687 str(pair
["thread-a"]),
688 str(pair
["thread-b"]),
692 if numa_node
.get("num-threads"):
693 numa
["threads"] = int(numa_node
["num-threads"])
696 if numa_node
.get("memory-mb"):
697 numa
["memory"] = max(int(int(numa_node
["memory-mb"]) / 1024), 1)
699 numa_list
.append(numa
)
702 return numa_list
, epa_vcpu_set
705 def _process_guest_epa_cpu_pinning_params(
706 guest_epa_quota
: Dict
[str, Any
],
709 ) -> Tuple
[Dict
[str, Any
], bool]:
713 guest_epa_quota (Dict[str, Any]): [description]
714 vcpu_count (int): [description]
715 epa_vcpu_set (bool): [description]
718 Tuple[Dict[str, Any], bool]: [description]
721 local_epa_vcpu_set
= epa_vcpu_set
724 guest_epa_quota
.get("cpu-pinning-policy") == "DEDICATED"
727 # Pinning policy "REQUIRE" uses threads as host should support SMT architecture
728 # Pinning policy "ISOLATE" uses cores as host should not support SMT architecture
729 # Pinning policy "PREFER" uses threads in case host supports SMT architecture
732 if guest_epa_quota
.get("cpu-thread-pinning-policy") == "ISOLATE"
734 ] = max(vcpu_count
, 1)
735 local_epa_vcpu_set
= True
737 return numa
, local_epa_vcpu_set
740 def _process_epa_params(
741 target_flavor
: Dict
[str, Any
],
746 target_flavor (Dict[str, Any]): [description]
749 Dict[str, Any]: [description]
755 if target_flavor
.get("guest-epa"):
756 guest_epa
= target_flavor
["guest-epa"]
758 numa_list
, epa_vcpu_set
= Ns
._process
_guest
_epa
_numa
_params
(
759 guest_epa_quota
=guest_epa
762 if guest_epa
.get("mempage-size"):
763 extended
["mempage-size"] = guest_epa
.get("mempage-size")
765 if guest_epa
.get("cpu-pinning-policy"):
766 extended
["cpu-pinning-policy"] = guest_epa
.get("cpu-pinning-policy")
768 if guest_epa
.get("cpu-thread-pinning-policy"):
769 extended
["cpu-thread-pinning-policy"] = guest_epa
.get(
770 "cpu-thread-pinning-policy"
773 if guest_epa
.get("numa-node-policy"):
774 if guest_epa
.get("numa-node-policy").get("mem-policy"):
775 extended
["mem-policy"] = guest_epa
.get("numa-node-policy").get(
779 tmp_numa
, epa_vcpu_set
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
780 guest_epa_quota
=guest_epa
,
781 vcpu_count
=int(target_flavor
.get("vcpu-count", 1)),
782 epa_vcpu_set
=epa_vcpu_set
,
784 for numa
in numa_list
:
785 numa
.update(tmp_numa
)
788 Ns
._process
_guest
_epa
_quota
_params
(
789 guest_epa_quota
=guest_epa
,
790 epa_vcpu_set
=epa_vcpu_set
,
795 extended
["numas"] = numa_list
800 def _process_flavor_params(
801 target_flavor
: Dict
[str, Any
],
802 indata
: Dict
[str, Any
],
803 vim_info
: Dict
[str, Any
],
804 target_record_id
: str,
805 **kwargs
: Dict
[str, Any
],
810 target_flavor (Dict[str, Any]): [description]
811 indata (Dict[str, Any]): [description]
812 vim_info (Dict[str, Any]): [description]
813 target_record_id (str): [description]
816 Dict[str, Any]: [description]
818 db
= kwargs
.get("db")
822 "disk": int(target_flavor
["storage-gb"]),
823 "ram": int(target_flavor
["memory-mb"]),
824 "vcpus": int(target_flavor
["vcpu-count"]),
827 for vnf
in indata
.get("vnf", []):
828 for vdur
in vnf
.get("vdur", []):
829 if vdur
.get("ns-flavor-id") == target_flavor
.get("id"):
832 if db
and isinstance(indata
.get("vnf"), list):
833 vnfd_id
= indata
.get("vnf")[0].get("vnfd-id")
834 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
835 # check if there is persistent root disk
836 for vdu
in vnfd
.get("vdu", ()):
837 if vdu
["name"] == target_vdur
.get("vdu-name"):
838 for vsd
in vnfd
.get("virtual-storage-desc", ()):
839 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
842 root_disk
.get("type-of-storage")
843 == "persistent-storage:persistent-storage"
845 flavor_data
["disk"] = 0
847 for storage
in target_vdur
.get("virtual-storages", []):
849 storage
.get("type-of-storage")
850 == "etsi-nfv-descriptors:ephemeral-storage"
852 flavor_data
["ephemeral"] = int(storage
.get("size-of-storage", 0))
853 elif storage
.get("type-of-storage") == "etsi-nfv-descriptors:swap-storage":
854 flavor_data
["swap"] = int(storage
.get("size-of-storage", 0))
856 extended
= Ns
._process
_epa
_params
(target_flavor
)
858 flavor_data
["extended"] = extended
860 extra_dict
= {"find_params": {"flavor_data": flavor_data
}}
861 flavor_data_name
= flavor_data
.copy()
862 flavor_data_name
["name"] = target_flavor
["name"]
863 extra_dict
["params"] = {"flavor_data": flavor_data_name
}
867 def _process_net_params(
868 target_vld
: Dict
[str, Any
],
869 indata
: Dict
[str, Any
],
870 vim_info
: Dict
[str, Any
],
871 target_record_id
: str,
872 **kwargs
: Dict
[str, Any
],
874 """Function to process network parameters.
877 target_vld (Dict[str, Any]): [description]
878 indata (Dict[str, Any]): [description]
879 vim_info (Dict[str, Any]): [description]
880 target_record_id (str): [description]
883 Dict[str, Any]: [description]
887 if vim_info
.get("sdn"):
888 # vnf_preffix = "vnfrs:{}".format(vnfr_id)
889 # ns_preffix = "nsrs:{}".format(nsr_id)
890 # remove the ending ".sdn
891 vld_target_record_id
, _
, _
= target_record_id
.rpartition(".")
892 extra_dict
["params"] = {
894 for k
in ("sdn-ports", "target_vim", "vlds", "type")
898 # TODO needed to add target_id in the dependency.
899 if vim_info
.get("target_vim"):
900 extra_dict
["depends_on"] = [
901 f
"{vim_info.get('target_vim')} {vld_target_record_id}"
906 if vim_info
.get("vim_network_name"):
907 extra_dict
["find_params"] = {
909 "name": vim_info
.get("vim_network_name"),
912 elif vim_info
.get("vim_network_id"):
913 extra_dict
["find_params"] = {
915 "id": vim_info
.get("vim_network_id"),
918 elif target_vld
.get("mgmt-network") and not vim_info
.get("provider_network"):
919 extra_dict
["find_params"] = {
921 "name": target_vld
["id"],
925 extra_dict
["params"] = {
927 f
"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
929 "ip_profile": vim_info
.get("ip_profile"),
930 "provider_network_profile": vim_info
.get("provider_network"),
933 if not target_vld
.get("underlay"):
934 extra_dict
["params"]["net_type"] = "bridge"
936 extra_dict
["params"]["net_type"] = (
937 "ptp" if target_vld
.get("type") == "ELINE" else "data"
943 def find_persistent_root_volumes(
946 vdu_instantiation_volumes_list
: list,
949 """Find the persistent root volumes and add them to the disk_list
950 by parsing the instantiation parameters.
953 vnfd (dict): VNF descriptor
954 target_vdu (dict): processed VDU
955 vdu_instantiation_volumes_list (list): instantiation parameters for the each VDU as a list
956 disk_list (list): to be filled up
959 persistent_root_disk (dict): Details of persistent root disk
962 persistent_root_disk
= {}
963 # There can be only one root disk, when we find it, it will return the result
965 for vdu
, vsd
in product(
966 vnfd
.get("vdu", ()), vnfd
.get("virtual-storage-desc", ())
969 vdu
["name"] == target_vdu
["vdu-name"]
970 and vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]
974 root_disk
.get("type-of-storage")
975 == "persistent-storage:persistent-storage"
977 for vdu_volume
in vdu_instantiation_volumes_list
:
979 vdu_volume
["vim-volume-id"]
980 and root_disk
["id"] == vdu_volume
["name"]
982 persistent_root_disk
[vsd
["id"]] = {
983 "vim_volume_id": vdu_volume
["vim-volume-id"],
984 "image_id": vdu
.get("sw-image-desc"),
987 disk_list
.append(persistent_root_disk
[vsd
["id"]])
989 return persistent_root_disk
992 if root_disk
.get("size-of-storage"):
993 persistent_root_disk
[vsd
["id"]] = {
994 "image_id": vdu
.get("sw-image-desc"),
995 "size": root_disk
.get("size-of-storage"),
996 "keep": Ns
.is_volume_keeping_required(root_disk
),
999 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1001 return persistent_root_disk
1002 return persistent_root_disk
1005 def find_persistent_volumes(
1006 persistent_root_disk
: dict,
1008 vdu_instantiation_volumes_list
: list,
1011 """Find the ordinary persistent volumes and add them to the disk_list
1012 by parsing the instantiation parameters.
1015 persistent_root_disk: persistent root disk dictionary
1016 target_vdu: processed VDU
1017 vdu_instantiation_volumes_list: instantiation parameters for the each VDU as a list
1018 disk_list: to be filled up
1021 # Find the ordinary volumes which are not added to the persistent_root_disk
1022 persistent_disk
= {}
1023 for disk
in target_vdu
.get("virtual-storages", {}):
1025 disk
.get("type-of-storage") == "persistent-storage:persistent-storage"
1026 and disk
["id"] not in persistent_root_disk
.keys()
1028 for vdu_volume
in vdu_instantiation_volumes_list
:
1029 if vdu_volume
["vim-volume-id"] and disk
["id"] == vdu_volume
["name"]:
1030 persistent_disk
[disk
["id"]] = {
1031 "vim_volume_id": vdu_volume
["vim-volume-id"],
1033 disk_list
.append(persistent_disk
[disk
["id"]])
1036 if disk
["id"] not in persistent_disk
.keys():
1037 persistent_disk
[disk
["id"]] = {
1038 "size": disk
.get("size-of-storage"),
1039 "keep": Ns
.is_volume_keeping_required(disk
),
1041 disk_list
.append(persistent_disk
[disk
["id"]])
1044 def is_volume_keeping_required(virtual_storage_desc
: Dict
[str, Any
]) -> bool:
1045 """Function to decide keeping persistent volume
1049 virtual_storage_desc (Dict[str, Any]): virtual storage description dictionary
1055 if not virtual_storage_desc
.get("vdu-storage-requirements"):
1057 for item
in virtual_storage_desc
.get("vdu-storage-requirements", {}):
1058 if item
.get("key") == "keep-volume" and item
.get("value").lower() == "true":
1063 def is_shared_volume(
1064 virtual_storage_desc
: Dict
[str, Any
], vnfd_id
: str
1066 """Function to decide if the volume type is multi attached or not .
1069 virtual_storage_desc (Dict[str, Any]): virtual storage description dictionary
1070 vnfd_id (str): vnfd id
1074 name (str) New name if it is a multiattach disk
1077 if vdu_storage_requirements
:= virtual_storage_desc
.get(
1078 "vdu-storage-requirements", {}
1080 for item
in vdu_storage_requirements
:
1082 item
.get("key") == "multiattach"
1083 and item
.get("value").lower() == "true"
1085 name
= f
"shared-{virtual_storage_desc['id']}-{vnfd_id}"
1087 return virtual_storage_desc
["id"], False
1090 def _sort_vdu_interfaces(target_vdu
: dict) -> None:
1091 """Sort the interfaces according to position number.
1094 target_vdu (dict): Details of VDU to be created
1097 # If the position info is provided for all the interfaces, it will be sorted
1098 # according to position number ascendingly.
1099 sorted_interfaces
= sorted(
1100 target_vdu
["interfaces"],
1101 key
=lambda x
: (x
.get("position") is None, x
.get("position")),
1103 target_vdu
["interfaces"] = sorted_interfaces
1106 def _partially_locate_vdu_interfaces(target_vdu
: dict) -> None:
1107 """Only place the interfaces which has specific position.
1110 target_vdu (dict): Details of VDU to be created
1113 # If the position info is provided for some interfaces but not all of them, the interfaces
1114 # which has specific position numbers will be placed and others' positions will not be taken care.
1116 i
.get("position") + 1
1117 for i
in target_vdu
["interfaces"]
1118 if i
.get("position") is not None
1120 n
= len(target_vdu
["interfaces"])
1121 sorted_interfaces
= [-1] * n
1125 if target_vdu
["interfaces"][k
].get("position") is not None:
1126 if any(i
.get("position") == 0 for i
in target_vdu
["interfaces"]):
1127 idx
= target_vdu
["interfaces"][k
]["position"] + 1
1129 idx
= target_vdu
["interfaces"][k
]["position"]
1130 sorted_interfaces
[idx
- 1] = target_vdu
["interfaces"][k
]
1134 if target_vdu
["interfaces"][m
].get("position") is None:
1135 idy
= sorted_interfaces
.index(-1)
1136 sorted_interfaces
[idy
] = target_vdu
["interfaces"][m
]
1139 target_vdu
["interfaces"] = sorted_interfaces
1142 def _prepare_vdu_cloud_init(
1143 target_vdu
: dict, vdu2cloud_init
: dict, db
: object, fs
: object
1145 """Fill cloud_config dict with cloud init details.
1148 target_vdu (dict): Details of VDU to be created
1149 vdu2cloud_init (dict): Cloud init dict
1150 db (object): DB object
1151 fs (object): FS object
1154 cloud_config (dict): Cloud config details of VDU
1160 if target_vdu
.get("cloud-init"):
1161 if target_vdu
["cloud-init"] not in vdu2cloud_init
:
1162 vdu2cloud_init
[target_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1165 location
=target_vdu
["cloud-init"],
1168 cloud_content_
= vdu2cloud_init
[target_vdu
["cloud-init"]]
1169 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1170 cloud_init_content
=cloud_content_
,
1171 params
=target_vdu
.get("additionalParams"),
1172 context
=target_vdu
["cloud-init"],
1175 if target_vdu
.get("boot-data-drive"):
1176 cloud_config
["boot-data-drive"] = target_vdu
.get("boot-data-drive")
1181 def _check_vld_information_of_interfaces(
1182 interface
: dict, ns_preffix
: str, vnf_preffix
: str
1184 """Prepare the net_text by the virtual link information for vnf and ns level.
1186 interface (dict): Interface details
1187 ns_preffix (str): Prefix of NS
1188 vnf_preffix (str): Prefix of VNF
1191 net_text (str): information of net
1195 if interface
.get("ns-vld-id"):
1196 net_text
= ns_preffix
+ ":vld." + interface
["ns-vld-id"]
1197 elif interface
.get("vnf-vld-id"):
1198 net_text
= vnf_preffix
+ ":vld." + interface
["vnf-vld-id"]
1203 def _prepare_interface_port_security(interface
: dict) -> None:
1207 interface (dict): Interface details
1210 if "port-security-enabled" in interface
:
1211 interface
["port_security"] = interface
.pop("port-security-enabled")
1213 if "port-security-disable-strategy" in interface
:
1214 interface
["port_security_disable_strategy"] = interface
.pop(
1215 "port-security-disable-strategy"
1219 def _create_net_item_of_interface(interface
: dict, net_text
: str) -> dict:
1220 """Prepare net item including name, port security, floating ip etc.
1223 interface (dict): Interface details
1224 net_text (str): information of net
1227 net_item (dict): Dict including net details
1233 for x
, v
in interface
.items()
1239 "port_security_disable_strategy",
1243 net_item
["net_id"] = "TASK-" + net_text
1244 net_item
["type"] = "virtual"
1249 def _prepare_type_of_interface(
1250 interface
: dict, tasks_by_target_record_id
: dict, net_text
: str, net_item
: dict
1252 """Fill the net item type by interface type such as SR-IOV, OM-MGMT, bridge etc.
1255 interface (dict): Interface details
1256 tasks_by_target_record_id (dict): Task details
1257 net_text (str): information of net
1258 net_item (dict): Dict including net details
1261 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1262 # TODO floating_ip: True/False (or it can be None)
1264 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1265 # Mark the net create task as type data
1267 tasks_by_target_record_id
,
1273 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"][
1277 net_item
["use"] = "data"
1278 net_item
["model"] = interface
["type"]
1279 net_item
["type"] = interface
["type"]
1282 interface
.get("type") == "OM-MGMT"
1283 or interface
.get("mgmt-interface")
1284 or interface
.get("mgmt-vnf")
1286 net_item
["use"] = "mgmt"
1289 # If interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1290 net_item
["use"] = "bridge"
1291 net_item
["model"] = interface
.get("type")
1294 def _prepare_vdu_interfaces(
1300 tasks_by_target_record_id
: dict,
1303 """Prepare the net_item and add net_list, add mgmt interface to extra_dict.
1306 target_vdu (dict): VDU to be created
1307 extra_dict (dict): Dictionary to be filled
1308 ns_preffix (str): NS prefix as string
1309 vnf_preffix (str): VNF prefix as string
1310 logger (object): Logger Object
1311 tasks_by_target_record_id (dict): Task details
1312 net_list (list): Net list of VDU
1314 for iface_index
, interface
in enumerate(target_vdu
["interfaces"]):
1315 net_text
= Ns
._check
_vld
_information
_of
_interfaces
(
1316 interface
, ns_preffix
, vnf_preffix
1319 # Interface not connected to any vld
1321 "Interface {} from vdu {} not connected to any vld".format(
1322 iface_index
, target_vdu
["vdu-name"]
1327 extra_dict
["depends_on"].append(net_text
)
1329 Ns
._prepare
_interface
_port
_security
(interface
)
1331 net_item
= Ns
._create
_net
_item
_of
_interface
(interface
, net_text
)
1333 Ns
._prepare
_type
_of
_interface
(
1334 interface
, tasks_by_target_record_id
, net_text
, net_item
1337 if interface
.get("ip-address"):
1338 net_item
["ip_address"] = interface
["ip-address"]
1340 if interface
.get("mac-address"):
1341 net_item
["mac_address"] = interface
["mac-address"]
1343 net_list
.append(net_item
)
1345 if interface
.get("mgmt-vnf"):
1346 extra_dict
["mgmt_vnf_interface"] = iface_index
1347 elif interface
.get("mgmt-interface"):
1348 extra_dict
["mgmt_vdu_interface"] = iface_index
1351 def _prepare_vdu_ssh_keys(
1352 target_vdu
: dict, ro_nsr_public_key
: dict, cloud_config
: dict
1354 """Add ssh keys to cloud config.
1357 target_vdu (dict): Details of VDU to be created
1358 ro_nsr_public_key (dict): RO NSR public Key
1359 cloud_config (dict): Cloud config details
1364 if target_vdu
.get("ssh-keys"):
1365 ssh_keys
+= target_vdu
.get("ssh-keys")
1367 if target_vdu
.get("ssh-access-required"):
1368 ssh_keys
.append(ro_nsr_public_key
)
1371 cloud_config
["key-pairs"] = ssh_keys
1374 def _select_persistent_root_disk(vsd
: dict, vdu
: dict) -> dict:
1375 """Selects the persistent root disk if exists.
1377 vsd (dict): Virtual storage descriptors in VNFD
1378 vdu (dict): VNF descriptor
1381 root_disk (dict): Selected persistent root disk
1383 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
1387 ) == "persistent-storage:persistent-storage" and root_disk
.get(
1393 def _add_persistent_root_disk_to_disk_list(
1394 vnfd
: dict, target_vdu
: dict, persistent_root_disk
: dict, disk_list
: list
1396 """Find the persistent root disk and add to disk list.
1399 vnfd (dict): VNF descriptor
1400 target_vdu (dict): Details of VDU to be created
1401 persistent_root_disk (dict): Details of persistent root disk
1402 disk_list (list): Disks of VDU
1405 for vdu
in vnfd
.get("vdu", ()):
1406 if vdu
["name"] == target_vdu
["vdu-name"]:
1407 for vsd
in vnfd
.get("virtual-storage-desc", ()):
1408 root_disk
= Ns
._select
_persistent
_root
_disk
(vsd
, vdu
)
1412 persistent_root_disk
[vsd
["id"]] = {
1413 "image_id": vdu
.get("sw-image-desc"),
1414 "size": root_disk
["size-of-storage"],
1415 "keep": Ns
.is_volume_keeping_required(root_disk
),
1417 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1421 def _add_persistent_ordinary_disks_to_disk_list(
1423 persistent_root_disk
: dict,
1424 persistent_ordinary_disk
: dict,
1430 """Fill the disk list by adding persistent ordinary disks.
1433 target_vdu (dict): Details of VDU to be created
1434 persistent_root_disk (dict): Details of persistent root disk
1435 persistent_ordinary_disk (dict): Details of persistent ordinary disk
1436 disk_list (list): Disks of VDU
1439 if target_vdu
.get("virtual-storages"):
1440 for disk
in target_vdu
["virtual-storages"]:
1442 disk
.get("type-of-storage")
1443 == "persistent-storage:persistent-storage"
1444 and disk
["id"] not in persistent_root_disk
.keys()
1446 name
, multiattach
= Ns
.is_shared_volume(disk
, vnf_id
)
1447 persistent_ordinary_disk
[disk
["id"]] = {
1449 "size": disk
["size-of-storage"],
1450 "keep": Ns
.is_volume_keeping_required(disk
),
1451 "multiattach": multiattach
,
1453 disk_list
.append(persistent_ordinary_disk
[disk
["id"]])
1454 if multiattach
: # VDU creation has to wait for shared volumes
1455 extra_dict
["depends_on"].append(
1456 f
"nsrs:{nsr_id}:shared-volumes.{name}"
1460 def _prepare_vdu_affinity_group_list(
1461 target_vdu
: dict, extra_dict
: dict, ns_preffix
: str
1462 ) -> List
[Dict
[str, any
]]:
1463 """Process affinity group details to prepare affinity group list.
1466 target_vdu (dict): Details of VDU to be created
1467 extra_dict (dict): Dictionary to be filled
1468 ns_preffix (str): Prefix as string
1472 affinity_group_list (list): Affinity group details
1475 affinity_group_list
= []
1477 if target_vdu
.get("affinity-or-anti-affinity-group-id"):
1478 for affinity_group_id
in target_vdu
["affinity-or-anti-affinity-group-id"]:
1480 affinity_group_text
= (
1481 ns_preffix
+ ":affinity-or-anti-affinity-group." + affinity_group_id
1484 if not isinstance(extra_dict
.get("depends_on"), list):
1485 raise NsException("Invalid extra_dict format.")
1487 extra_dict
["depends_on"].append(affinity_group_text
)
1488 affinity_group
["affinity_group_id"] = "TASK-" + affinity_group_text
1489 affinity_group_list
.append(affinity_group
)
1491 return affinity_group_list
1494 def _process_vdu_params(
1495 target_vdu
: Dict
[str, Any
],
1496 indata
: Dict
[str, Any
],
1497 vim_info
: Dict
[str, Any
],
1498 target_record_id
: str,
1499 **kwargs
: Dict
[str, Any
],
1500 ) -> Dict
[str, Any
]:
1501 """Function to process VDU parameters.
1504 target_vdu (Dict[str, Any]): [description]
1505 indata (Dict[str, Any]): [description]
1506 vim_info (Dict[str, Any]): [description]
1507 target_record_id (str): [description]
1510 Dict[str, Any]: [description]
1512 vnfr_id
= kwargs
.get("vnfr_id")
1513 nsr_id
= kwargs
.get("nsr_id")
1514 vnfr
= kwargs
.get("vnfr")
1515 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1516 tasks_by_target_record_id
= kwargs
.get("tasks_by_target_record_id")
1517 logger
= kwargs
.get("logger")
1518 db
= kwargs
.get("db")
1519 fs
= kwargs
.get("fs")
1520 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1522 vnf_preffix
= "vnfrs:{}".format(vnfr_id
)
1523 ns_preffix
= "nsrs:{}".format(nsr_id
)
1524 image_text
= ns_preffix
+ ":image." + target_vdu
["ns-image-id"]
1525 extra_dict
= {"depends_on": [image_text
]}
1527 persistent_root_disk
= {}
1528 persistent_ordinary_disk
= {}
1529 vdu_instantiation_volumes_list
= []
1530 vdu_instantiation_flavor_id
= None
1532 vnfd_id
= vnfr
["vnfd-id"]
1533 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
1534 # If the position info is provided for all the interfaces, it will be sorted
1535 # according to position number ascendingly.
1537 True if i
.get("position") is not None else False
1538 for i
in target_vdu
["interfaces"]
1540 Ns
._sort
_vdu
_interfaces
(target_vdu
)
1542 # If the position info is provided for some interfaces but not all of them, the interfaces
1543 # which has specific position numbers will be placed and others' positions will not be taken care.
1545 Ns
._partially
_locate
_vdu
_interfaces
(target_vdu
)
1547 # If the position info is not provided for the interfaces, interfaces will be attached
1548 # according to the order in the VNFD.
1549 Ns
._prepare
_vdu
_interfaces
(
1555 tasks_by_target_record_id
,
1560 cloud_config
= Ns
._prepare
_vdu
_cloud
_init
(target_vdu
, vdu2cloud_init
, db
, fs
)
1562 # Prepare VDU ssh keys
1563 Ns
._prepare
_vdu
_ssh
_keys
(target_vdu
, ro_nsr_public_key
, cloud_config
)
1565 if target_vdu
.get("additionalParams"):
1566 vdu_instantiation_volumes_list
= (
1567 target_vdu
.get("additionalParams").get("OSM", {}).get("vdu_volumes")
1569 vdu_instantiation_flavor_id
= (
1570 target_vdu
.get("additionalParams").get("OSM", {}).get("vim_flavor_id")
1574 if vdu_instantiation_flavor_id
:
1575 flavor_id
= vdu_instantiation_flavor_id
1577 flavor_text
= ns_preffix
+ ":flavor." + target_vdu
["ns-flavor-id"]
1578 flavor_id
= "TASK-" + flavor_text
1579 extra_dict
["depends_on"].append(flavor_text
)
1581 if vdu_instantiation_volumes_list
:
1582 # Find the root volumes and add to the disk_list
1583 persistent_root_disk
= Ns
.find_persistent_root_volumes(
1584 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
1587 # Find the ordinary volumes which are not added to the persistent_root_disk
1588 # and put them to the disk list
1589 Ns
.find_persistent_volumes(
1590 persistent_root_disk
,
1592 vdu_instantiation_volumes_list
,
1597 # Vdu_instantiation_volumes_list is empty
1598 # First get add the persistent root disks to disk_list
1599 Ns
._add
_persistent
_root
_disk
_to
_disk
_list
(
1600 vnfd
, target_vdu
, persistent_root_disk
, disk_list
1602 # Add the persistent non-root disks to disk_list
1603 Ns
._add
_persistent
_ordinary
_disks
_to
_disk
_list
(
1605 persistent_root_disk
,
1606 persistent_ordinary_disk
,
1613 affinity_group_list
= Ns
._prepare
_vdu
_affinity
_group
_list
(
1614 target_vdu
, extra_dict
, ns_preffix
1617 extra_dict
["params"] = {
1618 "name": "{}-{}-{}-{}".format(
1619 indata
["name"][:16],
1620 vnfr
["member-vnf-index-ref"][:16],
1621 target_vdu
["vdu-name"][:32],
1622 target_vdu
.get("count-index") or 0,
1624 "description": target_vdu
["vdu-name"],
1626 "image_id": "TASK-" + image_text
,
1627 "flavor_id": flavor_id
,
1628 "affinity_group_list": affinity_group_list
,
1629 "net_list": net_list
,
1630 "cloud_config": cloud_config
or None,
1631 "disk_list": disk_list
,
1632 "availability_zone_index": None, # TODO
1633 "availability_zone_list": None, # TODO
1638 def _process_shared_volumes_params(
1639 target_shared_volume
: Dict
[str, Any
],
1640 indata
: Dict
[str, Any
],
1641 vim_info
: Dict
[str, Any
],
1642 target_record_id
: str,
1643 **kwargs
: Dict
[str, Any
],
1644 ) -> Dict
[str, Any
]:
1646 shared_volume_data
= {
1647 "size": target_shared_volume
["size-of-storage"],
1648 "name": target_shared_volume
["id"],
1649 "type": target_shared_volume
["type-of-storage"],
1650 "keep": Ns
.is_volume_keeping_required(target_shared_volume
),
1652 extra_dict
["params"] = shared_volume_data
1656 def _process_affinity_group_params(
1657 target_affinity_group
: Dict
[str, Any
],
1658 indata
: Dict
[str, Any
],
1659 vim_info
: Dict
[str, Any
],
1660 target_record_id
: str,
1661 **kwargs
: Dict
[str, Any
],
1662 ) -> Dict
[str, Any
]:
1663 """Get affinity or anti-affinity group parameters.
1666 target_affinity_group (Dict[str, Any]): [description]
1667 indata (Dict[str, Any]): [description]
1668 vim_info (Dict[str, Any]): [description]
1669 target_record_id (str): [description]
1672 Dict[str, Any]: [description]
1676 affinity_group_data
= {
1677 "name": target_affinity_group
["name"],
1678 "type": target_affinity_group
["type"],
1679 "scope": target_affinity_group
["scope"],
1682 if target_affinity_group
.get("vim-affinity-group-id"):
1683 affinity_group_data
["vim-affinity-group-id"] = target_affinity_group
[
1684 "vim-affinity-group-id"
1687 extra_dict
["params"] = {
1688 "affinity_group_data": affinity_group_data
,
1693 def _process_recreate_vdu_params(
1694 existing_vdu
: Dict
[str, Any
],
1695 db_nsr
: Dict
[str, Any
],
1696 vim_info
: Dict
[str, Any
],
1697 target_record_id
: str,
1699 **kwargs
: Dict
[str, Any
],
1700 ) -> Dict
[str, Any
]:
1701 """Function to process VDU parameters to recreate.
1704 existing_vdu (Dict[str, Any]): [description]
1705 db_nsr (Dict[str, Any]): [description]
1706 vim_info (Dict[str, Any]): [description]
1707 target_record_id (str): [description]
1708 target_id (str): [description]
1711 Dict[str, Any]: [description]
1713 vnfr
= kwargs
.get("vnfr")
1714 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1715 # logger = kwargs.get("logger")
1716 db
= kwargs
.get("db")
1717 fs
= kwargs
.get("fs")
1718 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1724 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
1726 if vim_details_text
:
1727 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
1729 for iface_index
, interface
in enumerate(existing_vdu
["interfaces"]):
1730 if "port-security-enabled" in interface
:
1731 interface
["port_security"] = interface
.pop("port-security-enabled")
1733 if "port-security-disable-strategy" in interface
:
1734 interface
["port_security_disable_strategy"] = interface
.pop(
1735 "port-security-disable-strategy"
1740 for x
, v
in interface
.items()
1746 "port_security_disable_strategy",
1750 existing_ifaces
= existing_vdu
["vim_info"][target_id
].get(
1751 "interfaces_backup", []
1756 for i
in existing_ifaces
1757 if i
["ip_address"] == interface
["ip-address"]
1762 net_item
["net_id"] = net_id
1763 net_item
["type"] = "virtual"
1765 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1766 # TODO floating_ip: True/False (or it can be None)
1767 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1768 net_item
["use"] = "data"
1769 net_item
["model"] = interface
["type"]
1770 net_item
["type"] = interface
["type"]
1772 interface
.get("type") == "OM-MGMT"
1773 or interface
.get("mgmt-interface")
1774 or interface
.get("mgmt-vnf")
1776 net_item
["use"] = "mgmt"
1778 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1779 net_item
["use"] = "bridge"
1780 net_item
["model"] = interface
.get("type")
1782 if interface
.get("ip-address"):
1783 net_item
["ip_address"] = interface
["ip-address"]
1785 if interface
.get("mac-address"):
1786 net_item
["mac_address"] = interface
["mac-address"]
1788 net_list
.append(net_item
)
1790 if interface
.get("mgmt-vnf"):
1791 extra_dict
["mgmt_vnf_interface"] = iface_index
1792 elif interface
.get("mgmt-interface"):
1793 extra_dict
["mgmt_vdu_interface"] = iface_index
1798 if existing_vdu
.get("cloud-init"):
1799 if existing_vdu
["cloud-init"] not in vdu2cloud_init
:
1800 vdu2cloud_init
[existing_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1803 location
=existing_vdu
["cloud-init"],
1806 cloud_content_
= vdu2cloud_init
[existing_vdu
["cloud-init"]]
1807 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1808 cloud_init_content
=cloud_content_
,
1809 params
=existing_vdu
.get("additionalParams"),
1810 context
=existing_vdu
["cloud-init"],
1813 if existing_vdu
.get("boot-data-drive"):
1814 cloud_config
["boot-data-drive"] = existing_vdu
.get("boot-data-drive")
1818 if existing_vdu
.get("ssh-keys"):
1819 ssh_keys
+= existing_vdu
.get("ssh-keys")
1821 if existing_vdu
.get("ssh-access-required"):
1822 ssh_keys
.append(ro_nsr_public_key
)
1825 cloud_config
["key-pairs"] = ssh_keys
1828 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
1829 disk_list
.append({"vim_id": vol_id
["id"]})
1831 affinity_group_list
= []
1833 if existing_vdu
.get("affinity-or-anti-affinity-group-id"):
1835 for affinity_group_id
in existing_vdu
["affinity-or-anti-affinity-group-id"]:
1836 for group
in db_nsr
.get("affinity-or-anti-affinity-group"):
1838 group
["id"] == affinity_group_id
1839 and group
["vim_info"][target_id
].get("vim_id", None) is not None
1841 affinity_group
["affinity_group_id"] = group
["vim_info"][
1843 ].get("vim_id", None)
1844 affinity_group_list
.append(affinity_group
)
1846 extra_dict
["params"] = {
1847 "name": "{}-{}-{}-{}".format(
1848 db_nsr
["name"][:16],
1849 vnfr
["member-vnf-index-ref"][:16],
1850 existing_vdu
["vdu-name"][:32],
1851 existing_vdu
.get("count-index") or 0,
1853 "description": existing_vdu
["vdu-name"],
1855 "image_id": vim_details
["image"]["id"],
1856 "flavor_id": vim_details
["flavor"]["id"],
1857 "affinity_group_list": affinity_group_list
,
1858 "net_list": net_list
,
1859 "cloud_config": cloud_config
or None,
1860 "disk_list": disk_list
,
1861 "availability_zone_index": None, # TODO
1862 "availability_zone_list": None, # TODO
1867 def calculate_diff_items(
1874 tasks_by_target_record_id
,
1881 """Function that returns the incremental changes (creation, deletion)
1882 related to a specific item `item` to be done. This function should be
1883 called for NS instantiation, NS termination, NS update to add a new VNF
1884 or a new VLD, remove a VNF or VLD, etc.
1885 Item can be `net`, `flavor`, `image` or `vdu`.
1886 It takes a list of target items from indata (which came from the REST API)
1887 and compares with the existing items from db_ro_nsr, identifying the
1888 incremental changes to be done. During the comparison, it calls the method
1889 `process_params` (which was passed as parameter, and is particular for each
1893 indata (Dict[str, Any]): deployment info
1894 db_nsr: NSR record from DB
1895 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1896 db_nsr_update (Dict[str, Any]): NSR info to update in DB
1897 item (str): element to process (net, vdu...)
1898 tasks_by_target_record_id (Dict[str, Any]):
1899 [<target_record_id>, <task>]
1900 action_id (str): action id
1901 nsr_id (str): NSR id
1902 task_index (number): task index to add to task name
1903 vnfr_id (str): VNFR id
1904 vnfr (Dict[str, Any]): VNFR info
1907 List: list with the incremental changes (deletes, creates) for each item
1908 number: current task index
1916 process_params
= None
1917 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
1918 ro_nsr_public_key
= db_ro_nsr
["public_key"]
1919 # According to the type of item, the path, the target_list,
1920 # the existing_list and the method to process params are set
1921 db_path
= self
.db_path_map
[item
]
1922 process_params
= self
.process_params_function_map
[item
]
1923 if item
in ("net", "vdu"):
1924 # This case is specific for the NS VLD (not applied to VDU)
1926 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1927 target_list
= indata
.get("ns", []).get(db_path
, [])
1928 existing_list
= db_nsr
.get(db_path
, [])
1929 # This case is common for VNF VLDs and VNF VDUs
1931 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
1933 (vnf
for vnf
in indata
.get("vnf", ()) if vnf
["_id"] == vnfr_id
),
1936 target_list
= target_vnf
.get(db_path
, []) if target_vnf
else []
1937 existing_list
= vnfr
.get(db_path
, [])
1941 "affinity-or-anti-affinity-group",
1944 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1945 target_list
= indata
.get(item
, [])
1946 existing_list
= db_nsr
.get(item
, [])
1948 raise NsException("Item not supported: {}", item
)
1949 # ensure all the target_list elements has an "id". If not assign the index as id
1950 if target_list
is None:
1952 for target_index
, tl
in enumerate(target_list
):
1953 if tl
and not tl
.get("id"):
1954 tl
["id"] = str(target_index
)
1955 # step 1 items (networks,vdus,...) to be deleted/updated
1956 for item_index
, existing_item
in enumerate(existing_list
):
1958 (t
for t
in target_list
if t
["id"] == existing_item
["id"]),
1961 for target_vim
, existing_viminfo
in existing_item
.get(
1964 if existing_viminfo
is None:
1968 target_viminfo
= target_item
.get("vim_info", {}).get(target_vim
)
1970 target_viminfo
= None
1972 if target_viminfo
is None:
1974 self
._assign
_vim
(target_vim
)
1975 target_record_id
= "{}.{}".format(db_record
, existing_item
["id"])
1978 if target_vim
.startswith("sdn") or target_vim
.startswith("wim"):
1979 # item must be sdn-net instead of net if target_vim is a sdn
1981 target_record_id
+= ".sdn"
1984 "action_id": action_id
,
1986 "task_index": task_index
,
1991 "deployment_info": deployment_info
,
1992 "target_id": target_vim
,
1995 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
1996 "target_record_id": target_record_id
,
2001 # step 2 items (networks,vdus,...) to be created
2002 for target_item
in target_list
:
2004 for item_index
, existing_item
in enumerate(existing_list
):
2005 if existing_item
["id"] == target_item
["id"]:
2009 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
2010 existing_list
.append(target_item
)
2011 existing_item
= None
2013 for target_vim
, target_viminfo
in target_item
.get("vim_info", {}).items():
2014 existing_viminfo
= None
2017 existing_viminfo
= existing_item
.get("vim_info", {}).get(target_vim
)
2019 if existing_viminfo
is not None:
2022 target_record_id
= "{}.{}".format(db_record
, target_item
["id"])
2025 if target_vim
.startswith("sdn") or target_vim
.startswith("wim"):
2026 # item must be sdn-net instead of net if target_vim is a sdn
2028 target_record_id
+= ".sdn"
2032 "ns.calculate_diff_items target_item={}".format(target_item
)
2034 if process_params
== Ns
._process
_flavor
_params
:
2041 "calculate_diff_items for flavor kwargs={}".format(kwargs
)
2044 if process_params
== Ns
._process
_vdu
_params
:
2045 self
.logger
.debug("calculate_diff_items self.fs={}".format(self
.fs
))
2051 "vdu2cloud_init": vdu2cloud_init
,
2052 "tasks_by_target_record_id": tasks_by_target_record_id
,
2053 "logger": self
.logger
,
2056 "ro_nsr_public_key": ro_nsr_public_key
,
2059 self
.logger
.debug("calculate_diff_items kwargs={}".format(kwargs
))
2060 extra_dict
= process_params(
2067 self
._assign
_vim
(target_vim
)
2070 "action_id": action_id
,
2072 "task_index": task_index
,
2076 "deployment_info": deployment_info
,
2077 "target_id": target_vim
,
2080 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
2081 "target_record_id": target_record_id
,
2082 "extra_dict": extra_dict
,
2083 "common_id": target_item
.get("common_id", None),
2085 diff_items
.append(new_item
)
2086 tasks_by_target_record_id
[target_record_id
] = new_item
2089 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
2091 return diff_items
, task_index
2093 def calculate_all_differences_to_deploy(
2103 tasks_by_target_record_id
,
2105 """This method calculates the ordered list of items (`changes_list`)
2106 to be created and deleted.
2109 indata (Dict[str, Any]): deployment info
2110 nsr_id (str): NSR id
2111 db_nsr: NSR record from DB
2112 db_vnfrs: VNFRS record from DB
2113 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
2114 db_nsr_update (Dict[str, Any]): NSR info to update in DB
2115 db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
2116 action_id (str): action id
2117 tasks_by_target_record_id (Dict[str, Any]):
2118 [<target_record_id>, <task>]
2121 List: ordered list of items to be created and deleted.
2125 # set list with diffs:
2128 # NS vld, image and flavor
2133 "affinity-or-anti-affinity-group",
2135 self
.logger
.debug("process NS={} {}".format(nsr_id
, item
))
2136 diff_items
, task_index
= self
.calculate_diff_items(
2139 db_ro_nsr
=db_ro_nsr
,
2140 db_nsr_update
=db_nsr_update
,
2142 tasks_by_target_record_id
=tasks_by_target_record_id
,
2143 action_id
=action_id
,
2145 task_index
=task_index
,
2148 changes_list
+= diff_items
2151 for vnfr_id
, vnfr
in db_vnfrs
.items():
2152 # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
2153 for item
in ["net", "vdu", "shared-volumes"]:
2154 self
.logger
.debug("process VNF={} {}".format(vnfr_id
, item
))
2155 diff_items
, task_index
= self
.calculate_diff_items(
2158 db_ro_nsr
=db_ro_nsr
,
2159 db_nsr_update
=db_vnfrs_update
[vnfr
["_id"]],
2161 tasks_by_target_record_id
=tasks_by_target_record_id
,
2162 action_id
=action_id
,
2164 task_index
=task_index
,
2168 changes_list
+= diff_items
2172 def define_all_tasks(
2176 tasks_by_target_record_id
,
2178 """Function to create all the task structures obtanied from
2179 the method calculate_all_differences_to_deploy
2182 changes_list (List): ordered list of items to be created or deleted
2183 db_new_tasks (List): tasks list to be created
2184 action_id (str): action id
2185 tasks_by_target_record_id (Dict[str, Any]):
2186 [<target_record_id>, <task>]
2190 for change
in changes_list
:
2191 task
= Ns
._create
_task
(
2192 deployment_info
=change
["deployment_info"],
2193 target_id
=change
["target_id"],
2194 item
=change
["item"],
2195 action
=change
["action"],
2196 target_record
=change
["target_record"],
2197 target_record_id
=change
["target_record_id"],
2198 extra_dict
=change
.get("extra_dict", None),
2201 self
.logger
.debug("ns.define_all_tasks task={}".format(task
))
2202 tasks_by_target_record_id
[change
["target_record_id"]] = task
2203 db_new_tasks
.append(task
)
2205 if change
.get("common_id"):
2206 task
["common_id"] = change
["common_id"]
2208 def upload_all_tasks(
2213 """Function to save all tasks in the common DB
2216 db_new_tasks (List): tasks list to be created
2217 now (time): current time
2221 nb_ro_tasks
= 0 # for logging
2223 for db_task
in db_new_tasks
:
2224 target_id
= db_task
.pop("target_id")
2225 common_id
= db_task
.get("common_id")
2227 # Do not chek tasks with vim_status DELETED
2228 # because in manual heealing there are two tasks for the same vdur:
2229 # one with vim_status deleted and the other one with the actual VM status.
2235 "target_id": target_id
,
2236 "tasks.common_id": common_id
,
2237 "vim_info.vim_status.ne": "DELETED",
2239 update_dict
={"to_check_at": now
, "modified_at": now
},
2240 push
={"tasks": db_task
},
2241 fail_on_empty
=False,
2245 if not self
.db
.set_one(
2248 "target_id": target_id
,
2249 "tasks.target_record": db_task
["target_record"],
2250 "vim_info.vim_status.ne": "DELETED",
2252 update_dict
={"to_check_at": now
, "modified_at": now
},
2253 push
={"tasks": db_task
},
2254 fail_on_empty
=False,
2257 self
.logger
.debug("Updating database, Creating ro_tasks")
2258 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
2260 self
.db
.create("ro_tasks", db_ro_task
)
2263 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2264 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
2268 def upload_recreate_tasks(
2273 """Function to save recreate tasks in the common DB
2276 db_new_tasks (List): tasks list to be created
2277 now (time): current time
2281 nb_ro_tasks
= 0 # for logging
2283 for db_task
in db_new_tasks
:
2284 target_id
= db_task
.pop("target_id")
2285 self
.logger
.debug("target_id={} db_task={}".format(target_id
, db_task
))
2287 action
= db_task
.get("action", None)
2290 self
.logger
.debug("Updating database, Creating ro_tasks")
2291 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
2293 # If DELETE task: the associated created items should be removed
2294 # (except persistent volumes):
2295 if action
== "DELETE":
2296 db_ro_task
["vim_info"]["created"] = True
2297 db_ro_task
["vim_info"]["created_items"] = db_task
.get(
2300 db_ro_task
["vim_info"]["volumes_to_hold"] = db_task
.get(
2301 "volumes_to_hold", []
2303 db_ro_task
["vim_info"]["vim_id"] = db_task
.get("vim_id", None)
2306 self
.logger
.debug("upload_all_tasks db_ro_task={}".format(db_ro_task
))
2307 self
.db
.create("ro_tasks", db_ro_task
)
2310 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
2311 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
2315 def _prepare_created_items_for_healing(
2321 # Get created_items from ro_task
2322 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2323 for ro_task
in ro_tasks
:
2324 for task
in ro_task
["tasks"]:
2326 task
["target_record"] == target_record
2327 and task
["action"] == "CREATE"
2328 and ro_task
["vim_info"]["created_items"]
2330 created_items
= ro_task
["vim_info"]["created_items"]
2333 return created_items
2335 def _prepare_persistent_volumes_for_healing(
2340 # The associated volumes of the VM shouldn't be removed
2343 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
2344 if vim_details_text
:
2345 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
2347 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
2348 volumes_list
.append(vol_id
["id"])
2352 def prepare_changes_to_recreate(
2360 tasks_by_target_record_id
,
2362 """This method will obtain an ordered list of items (`changes_list`)
2363 to be created and deleted to meet the recreate request.
2367 "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id
, indata
)
2371 # set list with diffs:
2373 db_path
= self
.db_path_map
["vdu"]
2374 target_list
= indata
.get("healVnfData", {})
2375 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
2376 ro_nsr_public_key
= db_ro_nsr
["public_key"]
2378 # Check each VNF of the target
2379 for target_vnf
in target_list
:
2380 # Find this VNF in the list from DB, raise exception if vnfInstanceId is not found
2381 vnfr_id
= target_vnf
["vnfInstanceId"]
2382 existing_vnf
= db_vnfrs
.get(vnfr_id
, {})
2383 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
2384 # vim_account_id = existing_vnf.get("vim-account-id", "")
2386 target_vdus
= target_vnf
.get("additionalParams", {}).get("vdu", [])
2387 # Check each VDU of this VNF
2389 # Create target_vdu_list from DB, if VDUs are not specified
2391 for existing_vdu
in existing_vnf
.get("vdur"):
2392 vdu_name
= existing_vdu
.get("vdu-name", None)
2393 vdu_index
= existing_vdu
.get("count-index", 0)
2394 vdu_to_be_healed
= {"vdu-id": vdu_name
, "count-index": vdu_index
}
2395 target_vdus
.append(vdu_to_be_healed
)
2396 for target_vdu
in target_vdus
:
2397 vdu_name
= target_vdu
.get("vdu-id", None)
2398 # For multi instance VDU count-index is mandatory
2399 # For single session VDU count-indes is 0
2400 count_index
= target_vdu
.get("count-index", 0)
2402 existing_instance
= {}
2404 for instance
in existing_vnf
.get("vdur", {}):
2406 instance
["vdu-name"] == vdu_name
2407 and instance
["count-index"] == count_index
2409 existing_instance
= instance
2414 target_record_id
= "{}.{}".format(db_record
, existing_instance
["id"])
2416 # The target VIM is the one already existing in DB to recreate
2417 for target_vim
, target_viminfo
in existing_instance
.get(
2420 # step 1 vdu to be deleted
2421 self
._assign
_vim
(target_vim
)
2423 "action_id": action_id
,
2425 "task_index": task_index
,
2428 target_record
= f
"{db_record}.{item_index}.vim_info.{target_vim}"
2429 created_items
= self
._prepare
_created
_items
_for
_healing
(
2430 nsr_id
, target_record
2433 volumes_to_hold
= self
._prepare
_persistent
_volumes
_for
_healing
(
2434 target_vim
, existing_instance
2437 # Specific extra params for recreate tasks:
2439 "created_items": created_items
,
2440 "vim_id": existing_instance
["vim-id"],
2441 "volumes_to_hold": volumes_to_hold
,
2444 changes_list
.append(
2446 "deployment_info": deployment_info
,
2447 "target_id": target_vim
,
2450 "target_record": target_record
,
2451 "target_record_id": target_record_id
,
2452 "extra_dict": extra_dict
,
2455 delete_task_id
= f
"{action_id}:{task_index}"
2458 # step 2 vdu to be created
2464 "vnfr": existing_vnf
,
2465 "vdu2cloud_init": vdu2cloud_init
,
2466 "tasks_by_target_record_id": tasks_by_target_record_id
,
2467 "logger": self
.logger
,
2470 "ro_nsr_public_key": ro_nsr_public_key
,
2474 extra_dict
= self
._process
_recreate
_vdu
_params
(
2483 # The CREATE task depens on the DELETE task
2484 extra_dict
["depends_on"] = [delete_task_id
]
2486 # Add volumes created from created_items if any
2487 # Ports should be deleted with delete task and automatically created with create task
2489 for k
, v
in created_items
.items():
2491 k_item
, _
, k_id
= k
.partition(":")
2492 if k_item
== "volume":
2494 except Exception as e
:
2496 "Error evaluating created item {}: {}".format(k
, e
)
2498 extra_dict
["previous_created_volumes"] = volumes
2501 "action_id": action_id
,
2503 "task_index": task_index
,
2505 self
._assign
_vim
(target_vim
)
2508 "deployment_info": deployment_info
,
2509 "target_id": target_vim
,
2512 "target_record": target_record
,
2513 "target_record_id": target_record_id
,
2514 "extra_dict": extra_dict
,
2516 changes_list
.append(new_item
)
2517 tasks_by_target_record_id
[target_record_id
] = new_item
2522 def recreate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2523 self
.logger
.debug("ns.recreate nsr_id={} indata={}".format(nsr_id
, indata
))
2524 # TODO: validate_input(indata, recreate_schema)
2525 action_id
= indata
.get("action_id", str(uuid4()))
2526 # get current deployment
2527 db_vnfrs
= {} # vnf's info indexed by _id
2529 logging_text
= "Recreate nsr_id={} action_id={} indata={}".format(
2530 nsr_id
, action_id
, indata
2532 self
.logger
.debug(logging_text
+ "Enter")
2535 step
= "Getting ns and vnfr record from db"
2536 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2538 tasks_by_target_record_id
= {}
2539 # read from db: vnf's of this ns
2540 step
= "Getting vnfrs from db"
2541 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2542 self
.logger
.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list
))
2544 if not db_vnfrs_list
:
2545 raise NsException("Cannot obtain associated VNF for ns")
2547 for vnfr
in db_vnfrs_list
:
2548 db_vnfrs
[vnfr
["_id"]] = vnfr
2551 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2552 self
.logger
.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr
))
2555 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2557 with self
.write_lock
:
2559 step
= "process NS elements"
2560 changes_list
= self
.prepare_changes_to_recreate(
2565 db_ro_nsr
=db_ro_nsr
,
2566 action_id
=action_id
,
2567 tasks_by_target_record_id
=tasks_by_target_record_id
,
2570 self
.define_all_tasks(
2571 changes_list
=changes_list
,
2572 db_new_tasks
=db_new_tasks
,
2573 tasks_by_target_record_id
=tasks_by_target_record_id
,
2576 # Delete all ro_tasks registered for the targets vdurs (target_record)
2577 # If task of type CREATE exist then vim will try to get info form deleted VMs.
2578 # So remove all task related to target record.
2579 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2580 for change
in changes_list
:
2581 for ro_task
in ro_tasks
:
2582 for task
in ro_task
["tasks"]:
2583 if task
["target_record"] == change
["target_record"]:
2587 "_id": ro_task
["_id"],
2588 "modified_at": ro_task
["modified_at"],
2590 fail_on_empty
=False,
2593 step
= "Updating database, Appending tasks to ro_tasks"
2594 self
.upload_recreate_tasks(
2595 db_new_tasks
=db_new_tasks
,
2600 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2604 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2608 except Exception as e
:
2609 if isinstance(e
, (DbException
, NsException
)):
2611 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2614 e
= traceback_format_exc()
2615 self
.logger
.critical(
2616 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2620 raise NsException(e
)
2622 def deploy(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2623 self
.logger
.debug("ns.deploy nsr_id={} indata={}".format(nsr_id
, indata
))
2624 validate_input(indata
, deploy_schema
)
2625 action_id
= indata
.get("action_id", str(uuid4()))
2627 # get current deployment
2628 db_nsr_update
= {} # update operation on nsrs
2629 db_vnfrs_update
= {}
2630 db_vnfrs
= {} # vnf's info indexed by _id
2632 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2633 self
.logger
.debug(logging_text
+ "Enter")
2636 step
= "Getting ns and vnfr record from db"
2637 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2638 self
.logger
.debug("ns.deploy: db_nsr={}".format(db_nsr
))
2640 tasks_by_target_record_id
= {}
2641 # read from db: vnf's of this ns
2642 step
= "Getting vnfrs from db"
2643 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2645 if not db_vnfrs_list
:
2646 raise NsException("Cannot obtain associated VNF for ns")
2648 for vnfr
in db_vnfrs_list
:
2649 db_vnfrs
[vnfr
["_id"]] = vnfr
2650 db_vnfrs_update
[vnfr
["_id"]] = {}
2651 self
.logger
.debug("ns.deploy db_vnfrs={}".format(db_vnfrs
))
2654 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2657 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2659 # check that action_id is not in the list of actions. Suffixed with :index
2660 if action_id
in db_ro_nsr
["actions"]:
2664 new_action_id
= "{}:{}".format(action_id
, index
)
2666 if new_action_id
not in db_ro_nsr
["actions"]:
2667 action_id
= new_action_id
2670 + "Changing action_id in use to {}".format(action_id
)
2676 def _process_action(indata
):
2677 nonlocal db_new_tasks
2684 if indata
["action"]["action"] == "inject_ssh_key":
2685 key
= indata
["action"].get("key")
2686 user
= indata
["action"].get("user")
2687 password
= indata
["action"].get("password")
2689 for vnf
in indata
.get("vnf", ()):
2690 if vnf
["_id"] not in db_vnfrs
:
2691 raise NsException("Invalid vnf={}".format(vnf
["_id"]))
2693 db_vnfr
= db_vnfrs
[vnf
["_id"]]
2695 for target_vdu
in vnf
.get("vdur", ()):
2696 vdu_index
, vdur
= next(
2699 for i_v
in enumerate(db_vnfr
["vdur"])
2700 if i_v
[1]["id"] == target_vdu
["id"]
2707 "Invalid vdu vnf={}.{}".format(
2708 vnf
["_id"], target_vdu
["id"]
2712 target_vim
, vim_info
= next(
2713 k_v
for k_v
in vdur
["vim_info"].items()
2715 self
._assign
_vim
(target_vim
)
2716 target_record
= "vnfrs:{}:vdur.{}.ssh_keys".format(
2717 vnf
["_id"], vdu_index
2721 "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdur
["id"])
2724 "ip_address": vdur
.get("ip-address"),
2727 "password": password
,
2728 "private_key": db_ro_nsr
["private_key"],
2729 "salt": db_ro_nsr
["_id"],
2730 "schema_version": db_ro_nsr
["_admin"][
2737 "action_id": action_id
,
2739 "task_index": task_index
,
2742 task
= Ns
._create
_task
(
2743 deployment_info
=deployment_info
,
2744 target_id
=target_vim
,
2747 target_record
=target_record
,
2748 target_record_id
=None,
2749 extra_dict
=extra_dict
,
2752 task_index
= deployment_info
.get("task_index")
2754 db_new_tasks
.append(task
)
2756 with self
.write_lock
:
2757 if indata
.get("action"):
2758 _process_action(indata
)
2760 # compute network differences
2762 step
= "process NS elements"
2763 changes_list
= self
.calculate_all_differences_to_deploy(
2768 db_ro_nsr
=db_ro_nsr
,
2769 db_nsr_update
=db_nsr_update
,
2770 db_vnfrs_update
=db_vnfrs_update
,
2771 action_id
=action_id
,
2772 tasks_by_target_record_id
=tasks_by_target_record_id
,
2774 self
.define_all_tasks(
2775 changes_list
=changes_list
,
2776 db_new_tasks
=db_new_tasks
,
2777 tasks_by_target_record_id
=tasks_by_target_record_id
,
2780 step
= "Updating database, Appending tasks to ro_tasks"
2781 self
.upload_all_tasks(
2782 db_new_tasks
=db_new_tasks
,
2786 step
= "Updating database, nsrs"
2788 self
.db
.set_one("nsrs", {"_id": nsr_id
}, db_nsr_update
)
2790 for vnfr_id
, db_vnfr_update
in db_vnfrs_update
.items():
2792 step
= "Updating database, vnfrs={}".format(vnfr_id
)
2793 self
.db
.set_one("vnfrs", {"_id": vnfr_id
}, db_vnfr_update
)
2796 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2800 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2804 except Exception as e
:
2805 if isinstance(e
, (DbException
, NsException
)):
2807 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2810 e
= traceback_format_exc()
2811 self
.logger
.critical(
2812 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2816 raise NsException(e
)
2818 def delete(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2819 self
.logger
.debug("ns.delete version={} nsr_id={}".format(version
, nsr_id
))
2820 # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
2822 with self
.write_lock
:
2824 NsWorker
.delete_db_tasks(self
.db
, nsr_id
, None)
2825 except NsWorkerException
as e
:
2826 raise NsException(e
)
2828 return None, None, True
2830 def status(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2832 "ns.status version={} nsr_id={}, action_id={} indata={}".format(
2833 version
, nsr_id
, action_id
, indata
2839 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.action_id": action_id
})
2840 global_status
= "DONE"
2843 for ro_task
in ro_tasks
:
2844 for task
in ro_task
["tasks"]:
2845 if task
and task
["action_id"] == action_id
:
2846 task_list
.append(task
)
2849 if task
["status"] == "FAILED":
2850 global_status
= "FAILED"
2851 error_text
= "Error at {} {}: {}".format(
2852 task
["action"].lower(),
2854 ro_task
["vim_info"].get("vim_message") or "unknown",
2856 details
.append(error_text
)
2857 elif task
["status"] in ("SCHEDULED", "BUILD"):
2858 if global_status
!= "FAILED":
2859 global_status
= "BUILD"
2864 "status": global_status
,
2865 "details": ". ".join(details
)
2867 else "progress {}/{}".format(done
, total
),
2869 "action_id": action_id
,
2873 return return_data
, None, True
2875 def recreate_status(
2876 self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
2878 return self
.status(session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
)
2880 def cancel(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2882 "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
2883 session
, indata
, version
, nsr_id
, action_id
2887 return None, None, True
2889 def rebuild_start_stop_task(
2900 self
._assign
_vim
(target_vim
)
2901 target_record
= "vnfrs:{}:vdur.{}.vim_info.{}".format(
2902 vnf_id
, vdu_index
, target_vim
2904 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_id
)
2906 "action_id": action_id
,
2908 "task_index": task_index
,
2911 task
= Ns
._create
_task
(
2912 deployment_info
=deployment_info
,
2913 target_id
=target_vim
,
2916 target_record
=target_record
,
2917 target_record_id
=target_record_id
,
2918 extra_dict
=extra_dict
,
2922 def rebuild_start_stop(
2923 self
, session
, action_dict
, version
, nsr_id
, *args
, **kwargs
2928 action_id
= action_dict
.get("action_id", str(uuid4()))
2930 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2931 self
.logger
.debug(logging_text
+ "Enter")
2933 action
= list(action_dict
.keys())[0]
2934 task_dict
= action_dict
.get(action
)
2935 vim_vm_id
= action_dict
.get(action
).get("vim_vm_id")
2937 if action_dict
.get("stop"):
2941 step
= "lock the operation & do task creation"
2942 with self
.write_lock
:
2943 extra_dict
["params"] = {
2944 "vim_vm_id": vim_vm_id
,
2947 task
= self
.rebuild_start_stop_task(
2948 task_dict
["vdu_id"],
2949 task_dict
["vnf_id"],
2950 task_dict
["vdu_index"],
2954 task_dict
["target_vim"],
2957 db_new_tasks
.append(task
)
2958 step
= "upload Task to db"
2959 self
.upload_all_tasks(
2960 db_new_tasks
=db_new_tasks
,
2964 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2967 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2971 except Exception as e
:
2972 if isinstance(e
, (DbException
, NsException
)):
2974 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2977 e
= traceback_format_exc()
2978 self
.logger
.critical(
2979 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2982 raise NsException(e
)
2984 def get_deploy(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2985 nsrs
= self
.db
.get_list("nsrs", {})
2989 return_data
.append({"_id": ns
["_id"], "name": ns
["name"]})
2991 return return_data
, None, True
2993 def get_actions(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2994 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2997 for ro_task
in ro_tasks
:
2998 for task
in ro_task
["tasks"]:
2999 if task
["action_id"] not in return_data
:
3000 return_data
.append(task
["action_id"])
3002 return return_data
, None, True
3005 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3007 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
3008 self
._assign
_vim
(target_vim
)
3009 target_record
= "vnfrs:{}:vdur.{}.vim_info.{}".format(
3010 vnf
["_id"], vdu_index
, target_vim
3012 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
3014 "action_id": action_id
,
3016 "task_index": task_index
,
3019 task
= Ns
._create
_task
(
3020 deployment_info
=deployment_info
,
3021 target_id
=target_vim
,
3024 target_record
=target_record
,
3025 target_record_id
=target_record_id
,
3026 extra_dict
=extra_dict
,
3031 def migrate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3035 action_id
= indata
.get("action_id", str(uuid4()))
3037 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
3038 self
.logger
.debug(logging_text
+ "Enter")
3040 vnf_instance_id
= indata
["vnfInstanceId"]
3041 step
= "Getting vnfrs from db"
3042 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
3043 vdu
= indata
.get("vdu")
3044 migrateToHost
= indata
.get("migrateToHost")
3047 with self
.write_lock
:
3049 vdu_id
= indata
["vdu"]["vduId"]
3050 vdu_count_index
= indata
["vdu"].get("vduCountIndex", 0)
3051 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3053 vdu
["vdu-id-ref"] == vdu_id
3054 and vdu
["count-index"] == vdu_count_index
3056 extra_dict
["params"] = {
3057 "vim_vm_id": vdu
["vim-id"],
3058 "migrate_host": migrateToHost
,
3059 "vdu_vim_info": vdu
["vim_info"],
3061 step
= "Creating migration task for vdu:{}".format(vdu
)
3062 task
= self
.migrate_task(
3071 db_new_tasks
.append(task
)
3075 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3076 extra_dict
["params"] = {
3077 "vim_vm_id": vdu
["vim-id"],
3078 "migrate_host": migrateToHost
,
3079 "vdu_vim_info": vdu
["vim_info"],
3081 step
= "Creating migration task for vdu:{}".format(vdu
)
3082 task
= self
.migrate_task(
3091 db_new_tasks
.append(task
)
3094 self
.upload_all_tasks(
3095 db_new_tasks
=db_new_tasks
,
3100 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3103 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3107 except Exception as e
:
3108 if isinstance(e
, (DbException
, NsException
)):
3110 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3113 e
= traceback_format_exc()
3114 self
.logger
.critical(
3115 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3118 raise NsException(e
)
3120 def verticalscale_task(
3121 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
3123 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
3124 self
._assign
_vim
(target_vim
)
3125 target_record
= "vnfrs:{}:vdur.{}.vim_info.{}".format(
3126 vnf
["_id"], vdu_index
, target_vim
3128 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
3130 "action_id": action_id
,
3132 "task_index": task_index
,
3135 task
= Ns
._create
_task
(
3136 deployment_info
=deployment_info
,
3137 target_id
=target_vim
,
3138 item
="verticalscale",
3140 target_record
=target_record
,
3141 target_record_id
=target_record_id
,
3142 extra_dict
=extra_dict
,
3146 def verticalscale(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
3150 action_id
= indata
.get("action_id", str(uuid4()))
3152 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
3153 self
.logger
.debug(logging_text
+ "Enter")
3155 VnfFlavorData
= indata
.get("changeVnfFlavorData")
3156 vnf_instance_id
= VnfFlavorData
["vnfInstanceId"]
3157 step
= "Getting vnfrs from db"
3158 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
3159 vduid
= VnfFlavorData
["additionalParams"]["vduid"]
3160 vduCountIndex
= VnfFlavorData
["additionalParams"]["vduCountIndex"]
3161 virtualMemory
= VnfFlavorData
["additionalParams"]["virtualMemory"]
3162 numVirtualCpu
= VnfFlavorData
["additionalParams"]["numVirtualCpu"]
3163 sizeOfStorage
= VnfFlavorData
["additionalParams"]["sizeOfStorage"]
3165 "name": vduid
+ "-flv",
3166 "ram": virtualMemory
,
3167 "vcpus": numVirtualCpu
,
3168 "disk": sizeOfStorage
,
3171 step
= "Creating Tasks for vertical scaling"
3172 with self
.write_lock
:
3173 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
3175 vdu
["vdu-id-ref"] == vduid
3176 and vdu
["count-index"] == vduCountIndex
3178 extra_dict
["params"] = {
3179 "vim_vm_id": vdu
["vim-id"],
3180 "flavor_dict": flavor_dict
,
3182 task
= self
.verticalscale_task(
3191 db_new_tasks
.append(task
)
3194 self
.upload_all_tasks(
3195 db_new_tasks
=db_new_tasks
,
3199 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
3202 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
3206 except Exception as e
:
3207 if isinstance(e
, (DbException
, NsException
)):
3209 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
3212 e
= traceback_format_exc()
3213 self
.logger
.critical(
3214 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
3217 raise NsException(e
)