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
21 from random
import choice
as random_choice
22 from threading
import Lock
24 from traceback
import format_exc
as traceback_format_exc
25 from typing
import Any
, Dict
, Tuple
, Type
26 from uuid
import uuid4
28 from cryptography
.hazmat
.backends
import default_backend
as crypto_default_backend
29 from cryptography
.hazmat
.primitives
import serialization
as crypto_serialization
30 from cryptography
.hazmat
.primitives
.asymmetric
import rsa
38 from osm_common
import (
45 version
as common_version
,
47 from osm_common
.dbbase
import DbBase
, DbException
48 from osm_common
.fsbase
import FsBase
, FsException
49 from osm_common
.msgbase
import MsgException
50 from osm_ng_ro
.ns_thread
import deep_get
, NsWorker
, NsWorkerException
51 from osm_ng_ro
.validation
import deploy_schema
, validate_input
54 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
55 min_common_version
= "0.1.16"
58 class NsException(Exception):
59 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
60 self
.http_code
= http_code
61 super(Exception, self
).__init
__(message
)
66 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
67 will provide a random one
70 # Try getting docker id. If fails, get pid
72 with
open("/proc/self/cgroup", "r") as f
:
73 text_id_
= f
.readline()
74 _
, _
, text_id
= text_id_
.rpartition("/")
75 text_id
= text_id
.replace("\n", "")[:12]
83 return "".join(random_choice("0123456789abcdef") for _
in range(12))
87 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
90 for point
in v
.split("."):
91 filled
.append(point
.zfill(8))
102 # self.operations = None
104 # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
105 # If done now it will not be linked to parent not getting its handler and level
107 self
.write_lock
= None
108 self
.vims_assigned
= {}
112 self
.process_params_function_map
= {
113 "net": Ns
._process
_net
_params
,
114 "image": Ns
._process
_image
_params
,
115 "flavor": Ns
._process
_flavor
_params
,
116 "vdu": Ns
._process
_vdu
_params
,
117 "affinity-or-anti-affinity-group": Ns
._process
_affinity
_group
_params
,
124 "affinity-or-anti-affinity-group": "affinity-or-anti-affinity-group",
127 def init_db(self
, target_version
):
130 def start(self
, config
):
132 Connect to database, filesystem storage, and messaging
133 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
134 :param config: Configuration of db, storage, etc
138 self
.config
["process_id"] = get_process_id() # used for HA identity
139 self
.logger
= logging
.getLogger("ro.ns")
141 # check right version of common
142 if versiontuple(common_version
) < versiontuple(min_common_version
):
144 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
145 common_version
, min_common_version
151 if config
["database"]["driver"] == "mongo":
152 self
.db
= dbmongo
.DbMongo()
153 self
.db
.db_connect(config
["database"])
154 elif config
["database"]["driver"] == "memory":
155 self
.db
= dbmemory
.DbMemory()
156 self
.db
.db_connect(config
["database"])
159 "Invalid configuration param '{}' at '[database]':'driver'".format(
160 config
["database"]["driver"]
165 if config
["storage"]["driver"] == "local":
166 self
.fs
= fslocal
.FsLocal()
167 self
.fs
.fs_connect(config
["storage"])
168 elif config
["storage"]["driver"] == "mongo":
169 self
.fs
= fsmongo
.FsMongo()
170 self
.fs
.fs_connect(config
["storage"])
171 elif config
["storage"]["driver"] is None:
175 "Invalid configuration param '{}' at '[storage]':'driver'".format(
176 config
["storage"]["driver"]
181 if config
["message"]["driver"] == "local":
182 self
.msg
= msglocal
.MsgLocal()
183 self
.msg
.connect(config
["message"])
184 elif config
["message"]["driver"] == "kafka":
185 self
.msg
= msgkafka
.MsgKafka()
186 self
.msg
.connect(config
["message"])
189 "Invalid configuration param '{}' at '[message]':'driver'".format(
190 config
["message"]["driver"]
194 # TODO load workers to deal with exising database tasks
196 self
.write_lock
= Lock()
197 except (DbException
, FsException
, MsgException
) as e
:
198 raise NsException(str(e
), http_code
=e
.http_code
)
200 def get_assigned_vims(self
):
201 return list(self
.vims_assigned
.keys())
206 self
.db
.db_disconnect()
209 self
.fs
.fs_disconnect()
212 self
.msg
.disconnect()
214 self
.write_lock
= None
215 except (DbException
, FsException
, MsgException
) as e
:
216 raise NsException(str(e
), http_code
=e
.http_code
)
218 for worker
in self
.workers
:
219 worker
.insert_task(("terminate",))
221 def _create_worker(self
):
223 Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
224 limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
225 return the index of the assigned worker thread. Worker threads are storead at self.workers
227 # Look for a thread in idle status
231 for i
in range(len(self
.workers
))
232 if self
.workers
[i
] and self
.workers
[i
].idle
237 if worker_id
is not None:
238 # unset idle status to avoid race conditions
239 self
.workers
[worker_id
].idle
= False
241 worker_id
= len(self
.workers
)
243 if worker_id
< self
.config
["global"]["server.ns_threads"]:
244 # create a new worker
246 NsWorker(worker_id
, self
.config
, self
.plugins
, self
.db
)
248 self
.workers
[worker_id
].start()
250 # reached maximum number of threads, assign VIM to an existing one
251 worker_id
= self
.next_worker
252 self
.next_worker
= (self
.next_worker
+ 1) % self
.config
["global"][
258 def assign_vim(self
, target_id
):
259 with self
.write_lock
:
260 return self
._assign
_vim
(target_id
)
262 def _assign_vim(self
, target_id
):
263 if target_id
not in self
.vims_assigned
:
264 worker_id
= self
.vims_assigned
[target_id
] = self
._create
_worker
()
265 self
.workers
[worker_id
].insert_task(("load_vim", target_id
))
267 def reload_vim(self
, target_id
):
268 # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
269 # this is because database VIM information is cached for threads working with SDN
270 with self
.write_lock
:
271 for worker
in self
.workers
:
272 if worker
and not worker
.idle
:
273 worker
.insert_task(("reload_vim", target_id
))
275 def unload_vim(self
, target_id
):
276 with self
.write_lock
:
277 return self
._unload
_vim
(target_id
)
279 def _unload_vim(self
, target_id
):
280 if target_id
in self
.vims_assigned
:
281 worker_id
= self
.vims_assigned
[target_id
]
282 self
.workers
[worker_id
].insert_task(("unload_vim", target_id
))
283 del self
.vims_assigned
[target_id
]
285 def check_vim(self
, target_id
):
286 with self
.write_lock
:
287 if target_id
in self
.vims_assigned
:
288 worker_id
= self
.vims_assigned
[target_id
]
290 worker_id
= self
._create
_worker
()
292 worker
= self
.workers
[worker_id
]
293 worker
.insert_task(("check_vim", target_id
))
295 def unload_unused_vims(self
):
296 with self
.write_lock
:
299 for target_id
in self
.vims_assigned
:
300 if not self
.db
.get_one(
303 "target_id": target_id
,
304 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
308 vims_to_unload
.append(target_id
)
310 for target_id
in vims_to_unload
:
311 self
._unload
_vim
(target_id
)
319 """This method reads cloud init from a file.
321 Note: Not used as cloud init content is provided in the http body.
324 db (Type[DbBase]): [description]
325 fs (Type[FsBase]): [description]
326 location (str): can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
329 NsException: [description]
330 NsException: [description]
335 vnfd_id
, _
, other
= location
.partition(":")
336 _type
, _
, name
= other
.partition(":")
337 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
340 base_folder
= vnfd
["_admin"]["storage"]
341 cloud_init_file
= "{}/{}/cloud_init/{}".format(
342 base_folder
["folder"], base_folder
["pkg-dir"], name
347 "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
352 with fs
.file_open(cloud_init_file
, "r") as ci_file
:
353 cloud_init_content
= ci_file
.read()
355 cloud_init_content
= vnfd
["vdu"][int(name
)]["cloud-init"]
357 raise NsException("Mismatch descriptor for cloud init: {}".format(location
))
359 return cloud_init_content
363 cloud_init_content
: str,
364 params
: Dict
[str, Any
],
367 """Function that processes the cloud init to replace Jinja2 encoded parameters.
370 cloud_init_content (str): [description]
371 params (Dict[str, Any]): [description]
372 context (str): [description]
375 NsException: [description]
376 NsException: [description]
382 env
= Environment(undefined
=StrictUndefined
)
383 template
= env
.from_string(cloud_init_content
)
385 return template
.render(params
or {})
386 except UndefinedError
as e
:
388 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
389 "inside the 'additionalParamsForVnf' block".format(e
, context
)
391 except (TemplateError
, TemplateNotFound
) as e
:
393 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
398 def _create_db_ro_nsrs(self
, nsr_id
, now
):
400 key
= rsa
.generate_private_key(
401 backend
=crypto_default_backend(), public_exponent
=65537, key_size
=2048
403 private_key
= key
.private_bytes(
404 crypto_serialization
.Encoding
.PEM
,
405 crypto_serialization
.PrivateFormat
.PKCS8
,
406 crypto_serialization
.NoEncryption(),
408 public_key
= key
.public_key().public_bytes(
409 crypto_serialization
.Encoding
.OpenSSH
,
410 crypto_serialization
.PublicFormat
.OpenSSH
,
412 private_key
= private_key
.decode("utf8")
413 # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
414 i
= private_key
.find("\n")
415 private_key
= "-----BEGIN RSA PRIVATE KEY-----" + private_key
[i
:]
416 public_key
= public_key
.decode("utf8")
417 except Exception as e
:
418 raise NsException("Cannot create ssh-keys: {}".format(e
))
420 schema_version
= "1.1"
421 private_key_encrypted
= self
.db
.encrypt(
422 private_key
, schema_version
=schema_version
, salt
=nsr_id
429 "schema_version": schema_version
,
431 "public_key": public_key
,
432 "private_key": private_key_encrypted
,
435 self
.db
.create("ro_nsrs", db_content
)
441 deployment_info
: Dict
[str, Any
],
446 target_record_id
: str,
447 extra_dict
: Dict
[str, Any
] = None,
449 """Function to create task dict from deployment information.
452 deployment_info (Dict[str, Any]): [description]
453 target_id (str): [description]
454 item (str): [description]
455 action (str): [description]
456 target_record (str): [description]
457 target_record_id (str): [description]
458 extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
461 Dict[str, Any]: [description]
464 "target_id": target_id
, # it will be removed before pushing at database
465 "action_id": deployment_info
.get("action_id"),
466 "nsr_id": deployment_info
.get("nsr_id"),
467 "task_id": f
"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
468 "status": "SCHEDULED",
471 "target_record": target_record
,
472 "target_record_id": target_record_id
,
476 task
.update(extra_dict
) # params, find_params, depends_on
478 deployment_info
["task_index"] = deployment_info
.get("task_index", 0) + 1
485 task
: Dict
[str, Any
],
487 """Function to create an RO task from task information.
490 target_id (str): [description]
491 task (Dict[str, Any]): [description]
494 Dict[str, Any]: [description]
498 _id
= task
.get("task_id")
503 "target_id": target_id
,
506 "created_items": None,
523 def _process_image_params(
524 target_image
: Dict
[str, Any
],
525 indata
: Dict
[str, Any
],
526 vim_info
: Dict
[str, Any
],
527 target_record_id
: str,
528 **kwargs
: Dict
[str, Any
],
530 """Function to process VDU image parameters.
533 target_image (Dict[str, Any]): [description]
534 indata (Dict[str, Any]): [description]
535 vim_info (Dict[str, Any]): [description]
536 target_record_id (str): [description]
539 Dict[str, Any]: [description]
543 if target_image
.get("image"):
544 find_params
["filter_dict"] = {"name": target_image
.get("image")}
546 if target_image
.get("vim_image_id"):
547 find_params
["filter_dict"] = {"id": target_image
.get("vim_image_id")}
549 if target_image
.get("image_checksum"):
550 find_params
["filter_dict"] = {
551 "checksum": target_image
.get("image_checksum")
554 return {"find_params": find_params
}
557 def _get_resource_allocation_params(
558 quota_descriptor
: Dict
[str, Any
],
560 """Read the quota_descriptor from vnfd and fetch the resource allocation properties from the
564 quota_descriptor (Dict[str, Any]): cpu/mem/vif/disk-io quota descriptor
567 Dict[str, Any]: quota params for limit, reserve, shares from the descriptor object
571 if quota_descriptor
.get("limit"):
572 quota
["limit"] = int(quota_descriptor
["limit"])
574 if quota_descriptor
.get("reserve"):
575 quota
["reserve"] = int(quota_descriptor
["reserve"])
577 if quota_descriptor
.get("shares"):
578 quota
["shares"] = int(quota_descriptor
["shares"])
583 def _process_guest_epa_quota_params(
584 guest_epa_quota
: Dict
[str, Any
],
587 """Function to extract the guest epa quota parameters.
590 guest_epa_quota (Dict[str, Any]): [description]
591 epa_vcpu_set (bool): [description]
594 Dict[str, Any]: [description]
598 if guest_epa_quota
.get("cpu-quota") and not epa_vcpu_set
:
599 cpuquota
= Ns
._get
_resource
_allocation
_params
(
600 guest_epa_quota
.get("cpu-quota")
604 result
["cpu-quota"] = cpuquota
606 if guest_epa_quota
.get("mem-quota"):
607 vduquota
= Ns
._get
_resource
_allocation
_params
(
608 guest_epa_quota
.get("mem-quota")
612 result
["mem-quota"] = vduquota
614 if guest_epa_quota
.get("disk-io-quota"):
615 diskioquota
= Ns
._get
_resource
_allocation
_params
(
616 guest_epa_quota
.get("disk-io-quota")
620 result
["disk-io-quota"] = diskioquota
622 if guest_epa_quota
.get("vif-quota"):
623 vifquota
= Ns
._get
_resource
_allocation
_params
(
624 guest_epa_quota
.get("vif-quota")
628 result
["vif-quota"] = vifquota
633 def _process_guest_epa_numa_params(
634 guest_epa_quota
: Dict
[str, Any
],
635 ) -> Tuple
[Dict
[str, Any
], bool]:
639 guest_epa_quota (Dict[str, Any]): [description]
642 Tuple[Dict[str, Any], bool]: [description]
647 if guest_epa_quota
.get("numa-node-policy"):
648 numa_node_policy
= guest_epa_quota
.get("numa-node-policy")
650 if numa_node_policy
.get("node"):
651 numa_node
= numa_node_policy
["node"][0]
653 if numa_node
.get("num-cores"):
654 numa
["cores"] = numa_node
["num-cores"]
657 paired_threads
= numa_node
.get("paired-threads", {})
658 if paired_threads
.get("num-paired-threads"):
659 numa
["paired-threads"] = int(
660 numa_node
["paired-threads"]["num-paired-threads"]
664 if paired_threads
.get("paired-thread-ids"):
665 numa
["paired-threads-id"] = []
667 for pair
in paired_threads
["paired-thread-ids"]:
668 numa
["paired-threads-id"].append(
670 str(pair
["thread-a"]),
671 str(pair
["thread-b"]),
675 if numa_node
.get("num-threads"):
676 numa
["threads"] = int(numa_node
["num-threads"])
679 if numa_node
.get("memory-mb"):
680 numa
["memory"] = max(int(int(numa_node
["memory-mb"]) / 1024), 1)
682 return numa
, epa_vcpu_set
685 def _process_guest_epa_cpu_pinning_params(
686 guest_epa_quota
: Dict
[str, Any
],
689 ) -> Tuple
[Dict
[str, Any
], bool]:
693 guest_epa_quota (Dict[str, Any]): [description]
694 vcpu_count (int): [description]
695 epa_vcpu_set (bool): [description]
698 Tuple[Dict[str, Any], bool]: [description]
701 local_epa_vcpu_set
= epa_vcpu_set
704 guest_epa_quota
.get("cpu-pinning-policy") == "DEDICATED"
709 if guest_epa_quota
.get("cpu-thread-pinning-policy") != "PREFER"
711 ] = max(vcpu_count
, 1)
712 local_epa_vcpu_set
= True
714 return numa
, local_epa_vcpu_set
717 def _process_epa_params(
718 target_flavor
: Dict
[str, Any
],
723 target_flavor (Dict[str, Any]): [description]
726 Dict[str, Any]: [description]
731 if target_flavor
.get("guest-epa"):
732 guest_epa
= target_flavor
["guest-epa"]
734 numa
, epa_vcpu_set
= Ns
._process
_guest
_epa
_numa
_params
(
735 guest_epa_quota
=guest_epa
738 if guest_epa
.get("mempage-size"):
739 extended
["mempage-size"] = guest_epa
.get("mempage-size")
741 tmp_numa
, epa_vcpu_set
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
742 guest_epa_quota
=guest_epa
,
743 vcpu_count
=int(target_flavor
.get("vcpu-count", 1)),
744 epa_vcpu_set
=epa_vcpu_set
,
746 numa
.update(tmp_numa
)
749 Ns
._process
_guest
_epa
_quota
_params
(
750 guest_epa_quota
=guest_epa
,
751 epa_vcpu_set
=epa_vcpu_set
,
756 extended
["numas"] = [numa
]
761 def _process_flavor_params(
762 target_flavor
: Dict
[str, Any
],
763 indata
: Dict
[str, Any
],
764 vim_info
: Dict
[str, Any
],
765 target_record_id
: str,
766 **kwargs
: Dict
[str, Any
],
771 target_flavor (Dict[str, Any]): [description]
772 indata (Dict[str, Any]): [description]
773 vim_info (Dict[str, Any]): [description]
774 target_record_id (str): [description]
777 Dict[str, Any]: [description]
780 "disk": int(target_flavor
["storage-gb"]),
781 "ram": int(target_flavor
["memory-mb"]),
782 "vcpus": int(target_flavor
["vcpu-count"]),
786 for vnf
in indata
.get("vnf", []):
787 for vdur
in vnf
.get("vdur", []):
788 if vdur
.get("ns-flavor-id") == target_flavor
["id"]:
791 for storage
in target_vdur
.get("virtual-storages", []):
793 storage
.get("type-of-storage")
794 == "etsi-nfv-descriptors:ephemeral-storage"
796 flavor_data
["ephemeral"] = int(storage
.get("size-of-storage", 0))
797 elif storage
.get("type-of-storage") == "etsi-nfv-descriptors:swap-storage":
798 flavor_data
["swap"] = int(storage
.get("size-of-storage", 0))
800 extended
= Ns
._process
_epa
_params
(target_flavor
)
802 flavor_data
["extended"] = extended
804 extra_dict
= {"find_params": {"flavor_data": flavor_data
}}
805 flavor_data_name
= flavor_data
.copy()
806 flavor_data_name
["name"] = target_flavor
["name"]
807 extra_dict
["params"] = {"flavor_data": flavor_data_name
}
812 def _ip_profile_to_ro(
813 ip_profile
: Dict
[str, Any
],
818 ip_profile (Dict[str, Any]): [description]
821 Dict[str, Any]: [description]
828 if "v4" in ip_profile
.get("ip-version", "ipv4")
830 "subnet_address": ip_profile
.get("subnet-address"),
831 "gateway_address": ip_profile
.get("gateway-address"),
832 "dhcp_enabled": ip_profile
.get("dhcp-params", {}).get("enabled", False),
833 "dhcp_start_address": ip_profile
.get("dhcp-params", {}).get(
834 "start-address", None
836 "dhcp_count": ip_profile
.get("dhcp-params", {}).get("count", None),
839 if ip_profile
.get("dns-server"):
840 ro_ip_profile
["dns_address"] = ";".join(
841 [v
["address"] for v
in ip_profile
["dns-server"] if v
.get("address")]
844 if ip_profile
.get("security-group"):
845 ro_ip_profile
["security_group"] = ip_profile
["security-group"]
850 def _process_net_params(
851 target_vld
: Dict
[str, Any
],
852 indata
: Dict
[str, Any
],
853 vim_info
: Dict
[str, Any
],
854 target_record_id
: str,
855 **kwargs
: Dict
[str, Any
],
857 """Function to process network parameters.
860 target_vld (Dict[str, Any]): [description]
861 indata (Dict[str, Any]): [description]
862 vim_info (Dict[str, Any]): [description]
863 target_record_id (str): [description]
866 Dict[str, Any]: [description]
870 if vim_info
.get("sdn"):
871 # vnf_preffix = "vnfrs:{}".format(vnfr_id)
872 # ns_preffix = "nsrs:{}".format(nsr_id)
873 # remove the ending ".sdn
874 vld_target_record_id
, _
, _
= target_record_id
.rpartition(".")
875 extra_dict
["params"] = {
877 for k
in ("sdn-ports", "target_vim", "vlds", "type")
881 # TODO needed to add target_id in the dependency.
882 if vim_info
.get("target_vim"):
883 extra_dict
["depends_on"] = [
884 f
"{vim_info.get('target_vim')} {vld_target_record_id}"
889 if vim_info
.get("vim_network_name"):
890 extra_dict
["find_params"] = {
892 "name": vim_info
.get("vim_network_name"),
895 elif vim_info
.get("vim_network_id"):
896 extra_dict
["find_params"] = {
898 "id": vim_info
.get("vim_network_id"),
901 elif target_vld
.get("mgmt-network"):
902 extra_dict
["find_params"] = {
904 "name": target_vld
["id"],
908 extra_dict
["params"] = {
910 f
"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
912 "ip_profile": Ns
._ip
_profile
_to
_ro
(vim_info
.get("ip_profile")),
913 "provider_network_profile": vim_info
.get("provider_network"),
916 if not target_vld
.get("underlay"):
917 extra_dict
["params"]["net_type"] = "bridge"
919 extra_dict
["params"]["net_type"] = (
920 "ptp" if target_vld
.get("type") == "ELINE" else "data"
926 def _process_vdu_params(
927 target_vdu
: Dict
[str, Any
],
928 indata
: Dict
[str, Any
],
929 vim_info
: Dict
[str, Any
],
930 target_record_id
: str,
931 **kwargs
: Dict
[str, Any
],
933 """Function to process VDU parameters.
936 target_vdu (Dict[str, Any]): [description]
937 indata (Dict[str, Any]): [description]
938 vim_info (Dict[str, Any]): [description]
939 target_record_id (str): [description]
942 Dict[str, Any]: [description]
944 vnfr_id
= kwargs
.get("vnfr_id")
945 nsr_id
= kwargs
.get("nsr_id")
946 vnfr
= kwargs
.get("vnfr")
947 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
948 tasks_by_target_record_id
= kwargs
.get("tasks_by_target_record_id")
949 logger
= kwargs
.get("logger")
950 db
= kwargs
.get("db")
951 fs
= kwargs
.get("fs")
952 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
954 vnf_preffix
= "vnfrs:{}".format(vnfr_id
)
955 ns_preffix
= "nsrs:{}".format(nsr_id
)
956 image_text
= ns_preffix
+ ":image." + target_vdu
["ns-image-id"]
957 flavor_text
= ns_preffix
+ ":flavor." + target_vdu
["ns-flavor-id"]
958 extra_dict
= {"depends_on": [image_text
, flavor_text
]}
961 # If the position info is provided for all the interfaces, it will be sorted
962 # according to position number ascendingly.
963 if all(i
.get("position") for i
in target_vdu
["interfaces"]):
964 sorted_interfaces
= sorted(
965 target_vdu
["interfaces"],
966 key
=lambda x
: (x
.get("position") is None, x
.get("position")),
968 target_vdu
["interfaces"] = sorted_interfaces
970 # If the position info is provided for some interfaces but not all of them, the interfaces
971 # which has specific position numbers will be placed and others' positions will not be taken care.
973 if any(i
.get("position") for i
in target_vdu
["interfaces"]):
974 n
= len(target_vdu
["interfaces"])
975 sorted_interfaces
= [-1] * n
978 if target_vdu
["interfaces"][k
].get("position"):
979 idx
= target_vdu
["interfaces"][k
]["position"]
980 sorted_interfaces
[idx
- 1] = target_vdu
["interfaces"][k
]
983 if not target_vdu
["interfaces"][m
].get("position"):
984 idy
= sorted_interfaces
.index(-1)
985 sorted_interfaces
[idy
] = target_vdu
["interfaces"][m
]
988 target_vdu
["interfaces"] = sorted_interfaces
990 # If the position info is not provided for the interfaces, interfaces will be attached
991 # according to the order in the VNFD.
992 for iface_index
, interface
in enumerate(target_vdu
["interfaces"]):
993 if interface
.get("ns-vld-id"):
994 net_text
= ns_preffix
+ ":vld." + interface
["ns-vld-id"]
995 elif interface
.get("vnf-vld-id"):
996 net_text
= vnf_preffix
+ ":vld." + interface
["vnf-vld-id"]
999 "Interface {} from vdu {} not connected to any vld".format(
1000 iface_index
, target_vdu
["vdu-name"]
1004 continue # interface not connected to any vld
1006 extra_dict
["depends_on"].append(net_text
)
1008 if "port-security-enabled" in interface
:
1009 interface
["port_security"] = interface
.pop("port-security-enabled")
1011 if "port-security-disable-strategy" in interface
:
1012 interface
["port_security_disable_strategy"] = interface
.pop(
1013 "port-security-disable-strategy"
1018 for x
, v
in interface
.items()
1024 "port_security_disable_strategy",
1028 net_item
["net_id"] = "TASK-" + net_text
1029 net_item
["type"] = "virtual"
1031 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1032 # TODO floating_ip: True/False (or it can be None)
1033 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1034 # mark the net create task as type data
1036 tasks_by_target_record_id
,
1042 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"][
1046 net_item
["use"] = "data"
1047 net_item
["model"] = interface
["type"]
1048 net_item
["type"] = interface
["type"]
1050 interface
.get("type") == "OM-MGMT"
1051 or interface
.get("mgmt-interface")
1052 or interface
.get("mgmt-vnf")
1054 net_item
["use"] = "mgmt"
1056 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1057 net_item
["use"] = "bridge"
1058 net_item
["model"] = interface
.get("type")
1060 if interface
.get("ip-address"):
1061 net_item
["ip_address"] = interface
["ip-address"]
1063 if interface
.get("mac-address"):
1064 net_item
["mac_address"] = interface
["mac-address"]
1066 net_list
.append(net_item
)
1068 if interface
.get("mgmt-vnf"):
1069 extra_dict
["mgmt_vnf_interface"] = iface_index
1070 elif interface
.get("mgmt-interface"):
1071 extra_dict
["mgmt_vdu_interface"] = iface_index
1076 if target_vdu
.get("cloud-init"):
1077 if target_vdu
["cloud-init"] not in vdu2cloud_init
:
1078 vdu2cloud_init
[target_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1081 location
=target_vdu
["cloud-init"],
1084 cloud_content_
= vdu2cloud_init
[target_vdu
["cloud-init"]]
1085 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1086 cloud_init_content
=cloud_content_
,
1087 params
=target_vdu
.get("additionalParams"),
1088 context
=target_vdu
["cloud-init"],
1091 if target_vdu
.get("boot-data-drive"):
1092 cloud_config
["boot-data-drive"] = target_vdu
.get("boot-data-drive")
1096 if target_vdu
.get("ssh-keys"):
1097 ssh_keys
+= target_vdu
.get("ssh-keys")
1099 if target_vdu
.get("ssh-access-required"):
1100 ssh_keys
.append(ro_nsr_public_key
)
1103 cloud_config
["key-pairs"] = ssh_keys
1105 persistent_root_disk
= {}
1107 vnfd_id
= vnfr
["vnfd-id"]
1108 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
1109 for vdu
in vnfd
.get("vdu", ()):
1110 if vdu
["name"] == target_vdu
["vdu-name"]:
1111 for vsd
in vnfd
.get("virtual-storage-desc", ()):
1112 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
1116 ) == "persistent-storage:persistent-storage" and root_disk
.get(
1119 persistent_root_disk
[vsd
["id"]] = {
1120 "image_id": vdu
.get("sw-image-desc"),
1121 "size": root_disk
["size-of-storage"],
1123 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1125 if target_vdu
.get("virtual-storages"):
1126 for disk
in target_vdu
["virtual-storages"]:
1128 disk
.get("type-of-storage")
1129 == "persistent-storage:persistent-storage"
1130 and disk
["id"] not in persistent_root_disk
.keys()
1132 disk_list
.append({"size": disk
["size-of-storage"]})
1134 affinity_group_list
= []
1136 if target_vdu
.get("affinity-or-anti-affinity-group-id"):
1138 for affinity_group_id
in target_vdu
["affinity-or-anti-affinity-group-id"]:
1139 affinity_group_text
= (
1140 ns_preffix
+ ":affinity-or-anti-affinity-group." + affinity_group_id
1143 extra_dict
["depends_on"].append(affinity_group_text
)
1144 affinity_group
["affinity_group_id"] = "TASK-" + affinity_group_text
1145 affinity_group_list
.append(affinity_group
)
1147 extra_dict
["params"] = {
1148 "name": "{}-{}-{}-{}".format(
1149 indata
["name"][:16],
1150 vnfr
["member-vnf-index-ref"][:16],
1151 target_vdu
["vdu-name"][:32],
1152 target_vdu
.get("count-index") or 0,
1154 "description": target_vdu
["vdu-name"],
1156 "image_id": "TASK-" + image_text
,
1157 "flavor_id": "TASK-" + flavor_text
,
1158 "affinity_group_list": affinity_group_list
,
1159 "net_list": net_list
,
1160 "cloud_config": cloud_config
or None,
1161 "disk_list": disk_list
,
1162 "availability_zone_index": None, # TODO
1163 "availability_zone_list": None, # TODO
1169 def _process_affinity_group_params(
1170 target_affinity_group
: Dict
[str, Any
],
1171 indata
: Dict
[str, Any
],
1172 vim_info
: Dict
[str, Any
],
1173 target_record_id
: str,
1174 **kwargs
: Dict
[str, Any
],
1175 ) -> Dict
[str, Any
]:
1176 """Get affinity or anti-affinity group parameters.
1179 target_affinity_group (Dict[str, Any]): [description]
1180 indata (Dict[str, Any]): [description]
1181 vim_info (Dict[str, Any]): [description]
1182 target_record_id (str): [description]
1185 Dict[str, Any]: [description]
1189 affinity_group_data
= {
1190 "name": target_affinity_group
["name"],
1191 "type": target_affinity_group
["type"],
1192 "scope": target_affinity_group
["scope"],
1195 if target_affinity_group
.get("vim-affinity-group-id"):
1196 affinity_group_data
["vim-affinity-group-id"] = target_affinity_group
[
1197 "vim-affinity-group-id"
1200 extra_dict
["params"] = {
1201 "affinity_group_data": affinity_group_data
,
1207 def _process_recreate_vdu_params(
1208 existing_vdu
: Dict
[str, Any
],
1209 db_nsr
: Dict
[str, Any
],
1210 vim_info
: Dict
[str, Any
],
1211 target_record_id
: str,
1213 **kwargs
: Dict
[str, Any
],
1214 ) -> Dict
[str, Any
]:
1215 """Function to process VDU parameters to recreate.
1218 existing_vdu (Dict[str, Any]): [description]
1219 db_nsr (Dict[str, Any]): [description]
1220 vim_info (Dict[str, Any]): [description]
1221 target_record_id (str): [description]
1222 target_id (str): [description]
1225 Dict[str, Any]: [description]
1227 vnfr
= kwargs
.get("vnfr")
1228 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1229 # logger = kwargs.get("logger")
1230 db
= kwargs
.get("db")
1231 fs
= kwargs
.get("fs")
1232 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1238 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
1239 if vim_details_text
:
1240 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
1242 for iface_index
, interface
in enumerate(existing_vdu
["interfaces"]):
1244 if "port-security-enabled" in interface
:
1245 interface
["port_security"] = interface
.pop("port-security-enabled")
1247 if "port-security-disable-strategy" in interface
:
1248 interface
["port_security_disable_strategy"] = interface
.pop(
1249 "port-security-disable-strategy"
1254 for x
, v
in interface
.items()
1260 "port_security_disable_strategy",
1264 existing_ifaces
= existing_vdu
["vim_info"][target_id
].get(
1265 "interfaces_backup", []
1270 for i
in existing_ifaces
1271 if i
["ip_address"] == interface
["ip-address"]
1276 net_item
["net_id"] = net_id
1277 net_item
["type"] = "virtual"
1279 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1280 # TODO floating_ip: True/False (or it can be None)
1281 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1282 net_item
["use"] = "data"
1283 net_item
["model"] = interface
["type"]
1284 net_item
["type"] = interface
["type"]
1286 interface
.get("type") == "OM-MGMT"
1287 or interface
.get("mgmt-interface")
1288 or interface
.get("mgmt-vnf")
1290 net_item
["use"] = "mgmt"
1292 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1293 net_item
["use"] = "bridge"
1294 net_item
["model"] = interface
.get("type")
1296 if interface
.get("ip-address"):
1297 net_item
["ip_address"] = interface
["ip-address"]
1299 if interface
.get("mac-address"):
1300 net_item
["mac_address"] = interface
["mac-address"]
1302 net_list
.append(net_item
)
1304 if interface
.get("mgmt-vnf"):
1305 extra_dict
["mgmt_vnf_interface"] = iface_index
1306 elif interface
.get("mgmt-interface"):
1307 extra_dict
["mgmt_vdu_interface"] = iface_index
1312 if existing_vdu
.get("cloud-init"):
1313 if existing_vdu
["cloud-init"] not in vdu2cloud_init
:
1314 vdu2cloud_init
[existing_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1317 location
=existing_vdu
["cloud-init"],
1320 cloud_content_
= vdu2cloud_init
[existing_vdu
["cloud-init"]]
1321 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1322 cloud_init_content
=cloud_content_
,
1323 params
=existing_vdu
.get("additionalParams"),
1324 context
=existing_vdu
["cloud-init"],
1327 if existing_vdu
.get("boot-data-drive"):
1328 cloud_config
["boot-data-drive"] = existing_vdu
.get("boot-data-drive")
1332 if existing_vdu
.get("ssh-keys"):
1333 ssh_keys
+= existing_vdu
.get("ssh-keys")
1335 if existing_vdu
.get("ssh-access-required"):
1336 ssh_keys
.append(ro_nsr_public_key
)
1339 cloud_config
["key-pairs"] = ssh_keys
1342 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
1343 disk_list
.append({"vim_id": vol_id
["id"]})
1345 affinity_group_list
= []
1347 if existing_vdu
.get("affinity-or-anti-affinity-group-id"):
1349 for affinity_group_id
in existing_vdu
["affinity-or-anti-affinity-group-id"]:
1350 for group
in db_nsr
.get("affinity-or-anti-affinity-group"):
1352 group
["id"] == affinity_group_id
1353 and group
["vim_info"][target_id
].get("vim_id", None) is not None
1355 affinity_group
["affinity_group_id"] = group
["vim_info"][
1357 ].get("vim_id", None)
1358 affinity_group_list
.append(affinity_group
)
1360 extra_dict
["params"] = {
1361 "name": "{}-{}-{}-{}".format(
1362 db_nsr
["name"][:16],
1363 vnfr
["member-vnf-index-ref"][:16],
1364 existing_vdu
["vdu-name"][:32],
1365 existing_vdu
.get("count-index") or 0,
1367 "description": existing_vdu
["vdu-name"],
1369 "image_id": vim_details
["image"]["id"],
1370 "flavor_id": vim_details
["flavor"]["id"],
1371 "affinity_group_list": affinity_group_list
,
1372 "net_list": net_list
,
1373 "cloud_config": cloud_config
or None,
1374 "disk_list": disk_list
,
1375 "availability_zone_index": None, # TODO
1376 "availability_zone_list": None, # TODO
1381 def calculate_diff_items(
1388 tasks_by_target_record_id
,
1395 """Function that returns the incremental changes (creation, deletion)
1396 related to a specific item `item` to be done. This function should be
1397 called for NS instantiation, NS termination, NS update to add a new VNF
1398 or a new VLD, remove a VNF or VLD, etc.
1399 Item can be `net`, `flavor`, `image` or `vdu`.
1400 It takes a list of target items from indata (which came from the REST API)
1401 and compares with the existing items from db_ro_nsr, identifying the
1402 incremental changes to be done. During the comparison, it calls the method
1403 `process_params` (which was passed as parameter, and is particular for each
1407 indata (Dict[str, Any]): deployment info
1408 db_nsr: NSR record from DB
1409 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1410 db_nsr_update (Dict[str, Any]): NSR info to update in DB
1411 item (str): element to process (net, vdu...)
1412 tasks_by_target_record_id (Dict[str, Any]):
1413 [<target_record_id>, <task>]
1414 action_id (str): action id
1415 nsr_id (str): NSR id
1416 task_index (number): task index to add to task name
1417 vnfr_id (str): VNFR id
1418 vnfr (Dict[str, Any]): VNFR info
1421 List: list with the incremental changes (deletes, creates) for each item
1422 number: current task index
1430 process_params
= None
1431 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
1432 ro_nsr_public_key
= db_ro_nsr
["public_key"]
1434 # According to the type of item, the path, the target_list,
1435 # the existing_list and the method to process params are set
1436 db_path
= self
.db_path_map
[item
]
1437 process_params
= self
.process_params_function_map
[item
]
1438 if item
in ("net", "vdu"):
1439 # This case is specific for the NS VLD (not applied to VDU)
1441 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1442 target_list
= indata
.get("ns", []).get(db_path
, [])
1443 existing_list
= db_nsr
.get(db_path
, [])
1444 # This case is common for VNF VLDs and VNF VDUs
1446 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
1448 (vnf
for vnf
in indata
.get("vnf", ()) if vnf
["_id"] == vnfr_id
),
1451 target_list
= target_vnf
.get(db_path
, []) if target_vnf
else []
1452 existing_list
= vnfr
.get(db_path
, [])
1453 elif item
in ("image", "flavor", "affinity-or-anti-affinity-group"):
1454 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1455 target_list
= indata
.get(item
, [])
1456 existing_list
= db_nsr
.get(item
, [])
1458 raise NsException("Item not supported: {}", item
)
1460 # ensure all the target_list elements has an "id". If not assign the index as id
1461 if target_list
is None:
1463 for target_index
, tl
in enumerate(target_list
):
1464 if tl
and not tl
.get("id"):
1465 tl
["id"] = str(target_index
)
1467 # step 1 items (networks,vdus,...) to be deleted/updated
1468 for item_index
, existing_item
in enumerate(existing_list
):
1470 (t
for t
in target_list
if t
["id"] == existing_item
["id"]),
1474 for target_vim
, existing_viminfo
in existing_item
.get(
1477 if existing_viminfo
is None:
1481 target_viminfo
= target_item
.get("vim_info", {}).get(target_vim
)
1483 target_viminfo
= None
1485 if target_viminfo
is None:
1487 self
._assign
_vim
(target_vim
)
1488 target_record_id
= "{}.{}".format(db_record
, existing_item
["id"])
1491 if target_vim
.startswith("sdn"):
1492 # item must be sdn-net instead of net if target_vim is a sdn
1494 target_record_id
+= ".sdn"
1497 "action_id": action_id
,
1499 "task_index": task_index
,
1504 "deployment_info": deployment_info
,
1505 "target_id": target_vim
,
1508 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
1509 "target_record_id": target_record_id
,
1514 # step 2 items (networks,vdus,...) to be created
1515 for target_item
in target_list
:
1518 for item_index
, existing_item
in enumerate(existing_list
):
1519 if existing_item
["id"] == target_item
["id"]:
1523 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
1524 existing_list
.append(target_item
)
1525 existing_item
= None
1527 for target_vim
, target_viminfo
in target_item
.get("vim_info", {}).items():
1528 existing_viminfo
= None
1531 existing_viminfo
= existing_item
.get("vim_info", {}).get(target_vim
)
1533 if existing_viminfo
is not None:
1536 target_record_id
= "{}.{}".format(db_record
, target_item
["id"])
1539 if target_vim
.startswith("sdn"):
1540 # item must be sdn-net instead of net if target_vim is a sdn
1542 target_record_id
+= ".sdn"
1545 self
.logger
.warning(
1546 "ns.calculate_diff_items target_item={}".format(target_item
)
1548 if process_params
== Ns
._process
_vdu
_params
:
1549 self
.logger
.warning(
1550 "calculate_diff_items self.fs={}".format(self
.fs
)
1557 "vdu2cloud_init": vdu2cloud_init
,
1558 "tasks_by_target_record_id": tasks_by_target_record_id
,
1559 "logger": self
.logger
,
1562 "ro_nsr_public_key": ro_nsr_public_key
,
1565 self
.logger
.warning("calculate_diff_items kwargs={}".format(kwargs
))
1567 extra_dict
= process_params(
1574 self
._assign
_vim
(target_vim
)
1577 "action_id": action_id
,
1579 "task_index": task_index
,
1583 "deployment_info": deployment_info
,
1584 "target_id": target_vim
,
1587 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
1588 "target_record_id": target_record_id
,
1589 "extra_dict": extra_dict
,
1590 "common_id": target_item
.get("common_id", None),
1592 diff_items
.append(new_item
)
1593 tasks_by_target_record_id
[target_record_id
] = new_item
1596 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
1598 return diff_items
, task_index
1600 def calculate_all_differences_to_deploy(
1610 tasks_by_target_record_id
,
1612 """This method calculates the ordered list of items (`changes_list`)
1613 to be created and deleted.
1616 indata (Dict[str, Any]): deployment info
1617 nsr_id (str): NSR id
1618 db_nsr: NSR record from DB
1619 db_vnfrs: VNFRS record from DB
1620 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1621 db_nsr_update (Dict[str, Any]): NSR info to update in DB
1622 db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
1623 action_id (str): action id
1624 tasks_by_target_record_id (Dict[str, Any]):
1625 [<target_record_id>, <task>]
1628 List: ordered list of items to be created and deleted.
1632 # set list with diffs:
1635 # NS vld, image and flavor
1636 for item
in ["net", "image", "flavor", "affinity-or-anti-affinity-group"]:
1637 self
.logger
.debug("process NS={} {}".format(nsr_id
, item
))
1638 diff_items
, task_index
= self
.calculate_diff_items(
1641 db_ro_nsr
=db_ro_nsr
,
1642 db_nsr_update
=db_nsr_update
,
1644 tasks_by_target_record_id
=tasks_by_target_record_id
,
1645 action_id
=action_id
,
1647 task_index
=task_index
,
1650 changes_list
+= diff_items
1653 for vnfr_id
, vnfr
in db_vnfrs
.items():
1654 # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
1655 for item
in ["net", "vdu"]:
1656 self
.logger
.debug("process VNF={} {}".format(vnfr_id
, item
))
1657 diff_items
, task_index
= self
.calculate_diff_items(
1660 db_ro_nsr
=db_ro_nsr
,
1661 db_nsr_update
=db_vnfrs_update
[vnfr
["_id"]],
1663 tasks_by_target_record_id
=tasks_by_target_record_id
,
1664 action_id
=action_id
,
1666 task_index
=task_index
,
1670 changes_list
+= diff_items
1674 def define_all_tasks(
1678 tasks_by_target_record_id
,
1680 """Function to create all the task structures obtanied from
1681 the method calculate_all_differences_to_deploy
1684 changes_list (List): ordered list of items to be created or deleted
1685 db_new_tasks (List): tasks list to be created
1686 action_id (str): action id
1687 tasks_by_target_record_id (Dict[str, Any]):
1688 [<target_record_id>, <task>]
1692 for change
in changes_list
:
1693 task
= Ns
._create
_task
(
1694 deployment_info
=change
["deployment_info"],
1695 target_id
=change
["target_id"],
1696 item
=change
["item"],
1697 action
=change
["action"],
1698 target_record
=change
["target_record"],
1699 target_record_id
=change
["target_record_id"],
1700 extra_dict
=change
.get("extra_dict", None),
1703 self
.logger
.warning("ns.define_all_tasks task={}".format(task
))
1704 tasks_by_target_record_id
[change
["target_record_id"]] = task
1705 db_new_tasks
.append(task
)
1707 if change
.get("common_id"):
1708 task
["common_id"] = change
["common_id"]
1710 def upload_all_tasks(
1715 """Function to save all tasks in the common DB
1718 db_new_tasks (List): tasks list to be created
1719 now (time): current time
1723 nb_ro_tasks
= 0 # for logging
1725 for db_task
in db_new_tasks
:
1726 target_id
= db_task
.pop("target_id")
1727 common_id
= db_task
.get("common_id")
1729 # Do not chek tasks with vim_status DELETED
1730 # because in manual heealing there are two tasks for the same vdur:
1731 # one with vim_status deleted and the other one with the actual VM status.
1737 "target_id": target_id
,
1738 "tasks.common_id": common_id
,
1739 "vim_info.vim_status.ne": "DELETED",
1741 update_dict
={"to_check_at": now
, "modified_at": now
},
1742 push
={"tasks": db_task
},
1743 fail_on_empty
=False,
1747 if not self
.db
.set_one(
1750 "target_id": target_id
,
1751 "tasks.target_record": db_task
["target_record"],
1752 "vim_info.vim_status.ne": "DELETED",
1754 update_dict
={"to_check_at": now
, "modified_at": now
},
1755 push
={"tasks": db_task
},
1756 fail_on_empty
=False,
1759 self
.logger
.debug("Updating database, Creating ro_tasks")
1760 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
1762 self
.db
.create("ro_tasks", db_ro_task
)
1765 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
1766 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
1770 def upload_recreate_tasks(
1775 """Function to save recreate tasks in the common DB
1778 db_new_tasks (List): tasks list to be created
1779 now (time): current time
1783 nb_ro_tasks
= 0 # for logging
1785 for db_task
in db_new_tasks
:
1786 target_id
= db_task
.pop("target_id")
1787 self
.logger
.warning("target_id={} db_task={}".format(target_id
, db_task
))
1789 action
= db_task
.get("action", None)
1792 self
.logger
.debug("Updating database, Creating ro_tasks")
1793 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
1795 # If DELETE task: the associated created items should be removed
1796 # (except persistent volumes):
1797 if action
== "DELETE":
1798 db_ro_task
["vim_info"]["created"] = True
1799 db_ro_task
["vim_info"]["created_items"] = db_task
.get(
1802 db_ro_task
["vim_info"]["volumes_to_hold"] = db_task
.get(
1803 "volumes_to_hold", []
1805 db_ro_task
["vim_info"]["vim_id"] = db_task
.get("vim_id", None)
1808 self
.logger
.warning("upload_all_tasks db_ro_task={}".format(db_ro_task
))
1809 self
.db
.create("ro_tasks", db_ro_task
)
1812 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
1813 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
1817 def _prepare_created_items_for_healing(
1823 # Get created_items from ro_task
1824 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
1825 for ro_task
in ro_tasks
:
1826 for task
in ro_task
["tasks"]:
1828 task
["target_record"] == target_record
1829 and task
["action"] == "CREATE"
1830 and ro_task
["vim_info"]["created_items"]
1832 created_items
= ro_task
["vim_info"]["created_items"]
1835 return created_items
1837 def _prepare_persistent_volumes_for_healing(
1842 # The associated volumes of the VM shouldn't be removed
1845 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
1846 if vim_details_text
:
1847 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
1849 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
1850 volumes_list
.append(vol_id
["id"])
1854 def prepare_changes_to_recreate(
1862 tasks_by_target_record_id
,
1864 """This method will obtain an ordered list of items (`changes_list`)
1865 to be created and deleted to meet the recreate request.
1869 "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id
, indata
)
1873 # set list with diffs:
1875 db_path
= self
.db_path_map
["vdu"]
1876 target_list
= indata
.get("healVnfData", {})
1877 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
1878 ro_nsr_public_key
= db_ro_nsr
["public_key"]
1880 # Check each VNF of the target
1881 for target_vnf
in target_list
:
1882 # Find this VNF in the list from DB
1883 vnfr_id
= target_vnf
.get("vnfInstanceId", None)
1885 existing_vnf
= db_vnfrs
.get(vnfr_id
)
1886 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
1887 # vim_account_id = existing_vnf.get("vim-account-id", "")
1889 # Check each VDU of this VNF
1890 for target_vdu
in target_vnf
["additionalParams"].get("vdu", None):
1891 vdu_name
= target_vdu
.get("vdu-id", None)
1892 # For multi instance VDU count-index is mandatory
1893 # For single session VDU count-indes is 0
1894 count_index
= target_vdu
.get("count-index", 0)
1896 existing_instance
= None
1897 for instance
in existing_vnf
.get("vdur", None):
1899 instance
["vdu-name"] == vdu_name
1900 and instance
["count-index"] == count_index
1902 existing_instance
= instance
1907 target_record_id
= "{}.{}".format(db_record
, existing_instance
["id"])
1909 # The target VIM is the one already existing in DB to recreate
1910 for target_vim
, target_viminfo
in existing_instance
.get(
1913 # step 1 vdu to be deleted
1914 self
._assign
_vim
(target_vim
)
1916 "action_id": action_id
,
1918 "task_index": task_index
,
1921 target_record
= f
"{db_record}.{item_index}.vim_info.{target_vim}"
1922 created_items
= self
._prepare
_created
_items
_for
_healing
(
1923 nsr_id
, target_record
1926 volumes_to_hold
= self
._prepare
_persistent
_volumes
_for
_healing
(
1927 target_vim
, existing_instance
1930 # Specific extra params for recreate tasks:
1932 "created_items": created_items
,
1933 "vim_id": existing_instance
["vim-id"],
1934 "volumes_to_hold": volumes_to_hold
,
1937 changes_list
.append(
1939 "deployment_info": deployment_info
,
1940 "target_id": target_vim
,
1943 "target_record": target_record
,
1944 "target_record_id": target_record_id
,
1945 "extra_dict": extra_dict
,
1948 delete_task_id
= f
"{action_id}:{task_index}"
1951 # step 2 vdu to be created
1957 "vnfr": existing_vnf
,
1958 "vdu2cloud_init": vdu2cloud_init
,
1959 "tasks_by_target_record_id": tasks_by_target_record_id
,
1960 "logger": self
.logger
,
1963 "ro_nsr_public_key": ro_nsr_public_key
,
1967 extra_dict
= self
._process
_recreate
_vdu
_params
(
1976 # The CREATE task depens on the DELETE task
1977 extra_dict
["depends_on"] = [delete_task_id
]
1979 # Add volumes created from created_items if any
1980 # Ports should be deleted with delete task and automatically created with create task
1982 for k
, v
in created_items
.items():
1984 k_item
, _
, k_id
= k
.partition(":")
1985 if k_item
== "volume":
1987 except Exception as e
:
1989 "Error evaluating created item {}: {}".format(k
, e
)
1991 extra_dict
["previous_created_volumes"] = volumes
1994 "action_id": action_id
,
1996 "task_index": task_index
,
1998 self
._assign
_vim
(target_vim
)
2001 "deployment_info": deployment_info
,
2002 "target_id": target_vim
,
2005 "target_record": target_record
,
2006 "target_record_id": target_record_id
,
2007 "extra_dict": extra_dict
,
2009 changes_list
.append(new_item
)
2010 tasks_by_target_record_id
[target_record_id
] = new_item
2015 def recreate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2016 self
.logger
.debug("ns.recreate nsr_id={} indata={}".format(nsr_id
, indata
))
2017 # TODO: validate_input(indata, recreate_schema)
2018 action_id
= indata
.get("action_id", str(uuid4()))
2019 # get current deployment
2020 db_vnfrs
= {} # vnf's info indexed by _id
2022 logging_text
= "Recreate nsr_id={} action_id={} indata={}".format(
2023 nsr_id
, action_id
, indata
2025 self
.logger
.debug(logging_text
+ "Enter")
2028 step
= "Getting ns and vnfr record from db"
2029 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2031 tasks_by_target_record_id
= {}
2032 # read from db: vnf's of this ns
2033 step
= "Getting vnfrs from db"
2034 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2035 self
.logger
.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list
))
2037 if not db_vnfrs_list
:
2038 raise NsException("Cannot obtain associated VNF for ns")
2040 for vnfr
in db_vnfrs_list
:
2041 db_vnfrs
[vnfr
["_id"]] = vnfr
2044 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2045 self
.logger
.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr
))
2048 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2050 with self
.write_lock
:
2052 step
= "process NS elements"
2053 changes_list
= self
.prepare_changes_to_recreate(
2058 db_ro_nsr
=db_ro_nsr
,
2059 action_id
=action_id
,
2060 tasks_by_target_record_id
=tasks_by_target_record_id
,
2063 self
.define_all_tasks(
2064 changes_list
=changes_list
,
2065 db_new_tasks
=db_new_tasks
,
2066 tasks_by_target_record_id
=tasks_by_target_record_id
,
2069 # Delete all ro_tasks registered for the targets vdurs (target_record)
2070 # If task of type CREATE exist then vim will try to get info form deleted VMs.
2071 # So remove all task related to target record.
2072 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2073 for change
in changes_list
:
2074 for ro_task
in ro_tasks
:
2075 for task
in ro_task
["tasks"]:
2076 if task
["target_record"] == change
["target_record"]:
2080 "_id": ro_task
["_id"],
2081 "modified_at": ro_task
["modified_at"],
2083 fail_on_empty
=False,
2086 step
= "Updating database, Appending tasks to ro_tasks"
2087 self
.upload_recreate_tasks(
2088 db_new_tasks
=db_new_tasks
,
2093 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2097 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2101 except Exception as e
:
2102 if isinstance(e
, (DbException
, NsException
)):
2104 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2107 e
= traceback_format_exc()
2108 self
.logger
.critical(
2109 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2113 raise NsException(e
)
2115 def deploy(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2116 self
.logger
.debug("ns.deploy nsr_id={} indata={}".format(nsr_id
, indata
))
2117 validate_input(indata
, deploy_schema
)
2118 action_id
= indata
.get("action_id", str(uuid4()))
2120 # get current deployment
2121 db_nsr_update
= {} # update operation on nsrs
2122 db_vnfrs_update
= {}
2123 db_vnfrs
= {} # vnf's info indexed by _id
2125 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2126 self
.logger
.debug(logging_text
+ "Enter")
2129 step
= "Getting ns and vnfr record from db"
2130 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2131 self
.logger
.debug("ns.deploy: db_nsr={}".format(db_nsr
))
2133 tasks_by_target_record_id
= {}
2134 # read from db: vnf's of this ns
2135 step
= "Getting vnfrs from db"
2136 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2138 if not db_vnfrs_list
:
2139 raise NsException("Cannot obtain associated VNF for ns")
2141 for vnfr
in db_vnfrs_list
:
2142 db_vnfrs
[vnfr
["_id"]] = vnfr
2143 db_vnfrs_update
[vnfr
["_id"]] = {}
2144 self
.logger
.debug("ns.deploy db_vnfrs={}".format(db_vnfrs
))
2147 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2150 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2152 # check that action_id is not in the list of actions. Suffixed with :index
2153 if action_id
in db_ro_nsr
["actions"]:
2157 new_action_id
= "{}:{}".format(action_id
, index
)
2159 if new_action_id
not in db_ro_nsr
["actions"]:
2160 action_id
= new_action_id
2163 + "Changing action_id in use to {}".format(action_id
)
2169 def _process_action(indata
):
2170 nonlocal db_new_tasks
2177 if indata
["action"]["action"] == "inject_ssh_key":
2178 key
= indata
["action"].get("key")
2179 user
= indata
["action"].get("user")
2180 password
= indata
["action"].get("password")
2182 for vnf
in indata
.get("vnf", ()):
2183 if vnf
["_id"] not in db_vnfrs
:
2184 raise NsException("Invalid vnf={}".format(vnf
["_id"]))
2186 db_vnfr
= db_vnfrs
[vnf
["_id"]]
2188 for target_vdu
in vnf
.get("vdur", ()):
2189 vdu_index
, vdur
= next(
2192 for i_v
in enumerate(db_vnfr
["vdur"])
2193 if i_v
[1]["id"] == target_vdu
["id"]
2200 "Invalid vdu vnf={}.{}".format(
2201 vnf
["_id"], target_vdu
["id"]
2205 target_vim
, vim_info
= next(
2206 k_v
for k_v
in vdur
["vim_info"].items()
2208 self
._assign
_vim
(target_vim
)
2209 target_record
= "vnfrs:{}:vdur.{}.ssh_keys".format(
2210 vnf
["_id"], vdu_index
2214 "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdur
["id"])
2217 "ip_address": vdur
.get("ip-address"),
2220 "password": password
,
2221 "private_key": db_ro_nsr
["private_key"],
2222 "salt": db_ro_nsr
["_id"],
2223 "schema_version": db_ro_nsr
["_admin"][
2230 "action_id": action_id
,
2232 "task_index": task_index
,
2235 task
= Ns
._create
_task
(
2236 deployment_info
=deployment_info
,
2237 target_id
=target_vim
,
2240 target_record
=target_record
,
2241 target_record_id
=None,
2242 extra_dict
=extra_dict
,
2245 task_index
= deployment_info
.get("task_index")
2247 db_new_tasks
.append(task
)
2249 with self
.write_lock
:
2250 if indata
.get("action"):
2251 _process_action(indata
)
2253 # compute network differences
2255 step
= "process NS elements"
2256 changes_list
= self
.calculate_all_differences_to_deploy(
2261 db_ro_nsr
=db_ro_nsr
,
2262 db_nsr_update
=db_nsr_update
,
2263 db_vnfrs_update
=db_vnfrs_update
,
2264 action_id
=action_id
,
2265 tasks_by_target_record_id
=tasks_by_target_record_id
,
2267 self
.define_all_tasks(
2268 changes_list
=changes_list
,
2269 db_new_tasks
=db_new_tasks
,
2270 tasks_by_target_record_id
=tasks_by_target_record_id
,
2273 step
= "Updating database, Appending tasks to ro_tasks"
2274 self
.upload_all_tasks(
2275 db_new_tasks
=db_new_tasks
,
2279 step
= "Updating database, nsrs"
2281 self
.db
.set_one("nsrs", {"_id": nsr_id
}, db_nsr_update
)
2283 for vnfr_id
, db_vnfr_update
in db_vnfrs_update
.items():
2285 step
= "Updating database, vnfrs={}".format(vnfr_id
)
2286 self
.db
.set_one("vnfrs", {"_id": vnfr_id
}, db_vnfr_update
)
2289 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2293 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2297 except Exception as e
:
2298 if isinstance(e
, (DbException
, NsException
)):
2300 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2303 e
= traceback_format_exc()
2304 self
.logger
.critical(
2305 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2309 raise NsException(e
)
2311 def delete(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2312 self
.logger
.debug("ns.delete version={} nsr_id={}".format(version
, nsr_id
))
2313 # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
2315 with self
.write_lock
:
2317 NsWorker
.delete_db_tasks(self
.db
, nsr_id
, None)
2318 except NsWorkerException
as e
:
2319 raise NsException(e
)
2321 return None, None, True
2323 def status(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2325 "ns.status version={} nsr_id={}, action_id={} indata={}".format(
2326 version
, nsr_id
, action_id
, indata
2332 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.action_id": action_id
})
2333 global_status
= "DONE"
2336 for ro_task
in ro_tasks
:
2337 for task
in ro_task
["tasks"]:
2338 if task
and task
["action_id"] == action_id
:
2339 task_list
.append(task
)
2342 if task
["status"] == "FAILED":
2343 global_status
= "FAILED"
2344 error_text
= "Error at {} {}: {}".format(
2345 task
["action"].lower(),
2347 ro_task
["vim_info"].get("vim_message") or "unknown",
2349 details
.append(error_text
)
2350 elif task
["status"] in ("SCHEDULED", "BUILD"):
2351 if global_status
!= "FAILED":
2352 global_status
= "BUILD"
2357 "status": global_status
,
2358 "details": ". ".join(details
)
2360 else "progress {}/{}".format(done
, total
),
2362 "action_id": action_id
,
2366 return return_data
, None, True
2368 def recreate_status(
2369 self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
2371 return self
.status(session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
)
2373 def cancel(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2375 "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
2376 session
, indata
, version
, nsr_id
, action_id
2380 return None, None, True
2382 def rebuild_start_stop_task(
2393 self
._assign
_vim
(target_vim
)
2394 target_record
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_index
)
2395 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_id
)
2397 "action_id": action_id
,
2399 "task_index": task_index
,
2402 task
= Ns
._create
_task
(
2403 deployment_info
=deployment_info
,
2404 target_id
=target_vim
,
2407 target_record
=target_record
,
2408 target_record_id
=target_record_id
,
2409 extra_dict
=extra_dict
,
2413 def rebuild_start_stop(
2414 self
, session
, action_dict
, version
, nsr_id
, *args
, **kwargs
2419 action_id
= action_dict
.get("action_id", str(uuid4()))
2421 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2422 self
.logger
.debug(logging_text
+ "Enter")
2424 action
= list(action_dict
.keys())[0]
2425 task_dict
= action_dict
.get(action
)
2426 vim_vm_id
= action_dict
.get(action
).get("vim_vm_id")
2428 if action_dict
.get("stop"):
2432 step
= "lock the operation & do task creation"
2433 with self
.write_lock
:
2434 extra_dict
["params"] = {
2435 "vim_vm_id": vim_vm_id
,
2438 task
= self
.rebuild_start_stop_task(
2439 task_dict
["vdu_id"],
2440 task_dict
["vnf_id"],
2441 task_dict
["vdu_index"],
2445 task_dict
["target_vim"],
2448 db_new_tasks
.append(task
)
2449 step
= "upload Task to db"
2450 self
.upload_all_tasks(
2451 db_new_tasks
=db_new_tasks
,
2455 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2458 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2462 except Exception as e
:
2463 if isinstance(e
, (DbException
, NsException
)):
2465 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2468 e
= traceback_format_exc()
2469 self
.logger
.critical(
2470 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2473 raise NsException(e
)
2475 def get_deploy(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2476 nsrs
= self
.db
.get_list("nsrs", {})
2480 return_data
.append({"_id": ns
["_id"], "name": ns
["name"]})
2482 return return_data
, None, True
2484 def get_actions(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2485 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2488 for ro_task
in ro_tasks
:
2489 for task
in ro_task
["tasks"]:
2490 if task
["action_id"] not in return_data
:
2491 return_data
.append(task
["action_id"])
2493 return return_data
, None, True
2496 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
2498 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
2499 self
._assign
_vim
(target_vim
)
2500 target_record
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu_index
)
2501 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
2503 "action_id": action_id
,
2505 "task_index": task_index
,
2508 task
= Ns
._create
_task
(
2509 deployment_info
=deployment_info
,
2510 target_id
=target_vim
,
2513 target_record
=target_record
,
2514 target_record_id
=target_record_id
,
2515 extra_dict
=extra_dict
,
2520 def migrate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2524 action_id
= indata
.get("action_id", str(uuid4()))
2526 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2527 self
.logger
.debug(logging_text
+ "Enter")
2529 vnf_instance_id
= indata
["vnfInstanceId"]
2530 step
= "Getting vnfrs from db"
2531 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
2532 vdu
= indata
.get("vdu")
2533 migrateToHost
= indata
.get("migrateToHost")
2536 with self
.write_lock
:
2538 vdu_id
= indata
["vdu"]["vduId"]
2539 vdu_count_index
= indata
["vdu"].get("vduCountIndex", 0)
2540 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
2542 vdu
["vdu-id-ref"] == vdu_id
2543 and vdu
["count-index"] == vdu_count_index
2545 extra_dict
["params"] = {
2546 "vim_vm_id": vdu
["vim-id"],
2547 "migrate_host": migrateToHost
,
2548 "vdu_vim_info": vdu
["vim_info"],
2550 step
= "Creating migration task for vdu:{}".format(vdu
)
2551 task
= self
.migrate_task(
2560 db_new_tasks
.append(task
)
2565 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
2566 extra_dict
["params"] = {
2567 "vim_vm_id": vdu
["vim-id"],
2568 "migrate_host": migrateToHost
,
2569 "vdu_vim_info": vdu
["vim_info"],
2571 step
= "Creating migration task for vdu:{}".format(vdu
)
2572 task
= self
.migrate_task(
2581 db_new_tasks
.append(task
)
2584 self
.upload_all_tasks(
2585 db_new_tasks
=db_new_tasks
,
2590 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2593 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2597 except Exception as e
:
2598 if isinstance(e
, (DbException
, NsException
)):
2600 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2603 e
= traceback_format_exc()
2604 self
.logger
.critical(
2605 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2608 raise NsException(e
)
2610 def verticalscale_task(
2611 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
2613 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
2614 self
._assign
_vim
(target_vim
)
2615 target_record
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu_index
)
2616 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
2618 "action_id": action_id
,
2620 "task_index": task_index
,
2623 task
= Ns
._create
_task
(
2624 deployment_info
=deployment_info
,
2625 target_id
=target_vim
,
2626 item
="verticalscale",
2628 target_record
=target_record
,
2629 target_record_id
=target_record_id
,
2630 extra_dict
=extra_dict
,
2634 def verticalscale(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2638 action_id
= indata
.get("action_id", str(uuid4()))
2640 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2641 self
.logger
.debug(logging_text
+ "Enter")
2643 VnfFlavorData
= indata
.get("changeVnfFlavorData")
2644 vnf_instance_id
= VnfFlavorData
["vnfInstanceId"]
2645 step
= "Getting vnfrs from db"
2646 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
2647 vduid
= VnfFlavorData
["additionalParams"]["vduid"]
2648 vduCountIndex
= VnfFlavorData
["additionalParams"]["vduCountIndex"]
2649 virtualMemory
= VnfFlavorData
["additionalParams"]["virtualMemory"]
2650 numVirtualCpu
= VnfFlavorData
["additionalParams"]["numVirtualCpu"]
2651 sizeOfStorage
= VnfFlavorData
["additionalParams"]["sizeOfStorage"]
2653 "name": vduid
+ "-flv",
2654 "ram": virtualMemory
,
2655 "vcpus": numVirtualCpu
,
2656 "disk": sizeOfStorage
,
2659 step
= "Creating Tasks for vertical scaling"
2660 with self
.write_lock
:
2661 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
2663 vdu
["vdu-id-ref"] == vduid
2664 and vdu
["count-index"] == vduCountIndex
2666 extra_dict
["params"] = {
2667 "vim_vm_id": vdu
["vim-id"],
2668 "flavor_dict": flavor_dict
,
2670 task
= self
.verticalscale_task(
2679 db_new_tasks
.append(task
)
2682 self
.upload_all_tasks(
2683 db_new_tasks
=db_new_tasks
,
2687 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2690 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2694 except Exception as e
:
2695 if isinstance(e
, (DbException
, NsException
)):
2697 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2700 e
= traceback_format_exc()
2701 self
.logger
.critical(
2702 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2705 raise NsException(e
)