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
, 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
39 from osm_common
import (
46 version
as common_version
,
48 from osm_common
.dbbase
import DbBase
, DbException
49 from osm_common
.fsbase
import FsBase
, FsException
50 from osm_common
.msgbase
import MsgException
51 from osm_ng_ro
.ns_thread
import deep_get
, NsWorker
, NsWorkerException
52 from osm_ng_ro
.validation
import deploy_schema
, validate_input
55 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
56 min_common_version
= "0.1.16"
59 class NsException(Exception):
60 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
61 self
.http_code
= http_code
62 super(Exception, self
).__init
__(message
)
67 Obtain a unique ID for this process. If running from inside docker, it will get docker ID. If not it
68 will provide a random one
71 # Try getting docker id. If fails, get pid
73 with
open("/proc/self/cgroup", "r") as f
:
74 text_id_
= f
.readline()
75 _
, _
, text_id
= text_id_
.rpartition("/")
76 text_id
= text_id
.replace("\n", "")[:12]
84 return "".join(random_choice("0123456789abcdef") for _
in range(12))
88 """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
91 for point
in v
.split("."):
92 filled
.append(point
.zfill(8))
103 # self.operations = None
105 # ^ Getting logger inside method self.start because parent logger (ro) is not available yet.
106 # If done now it will not be linked to parent not getting its handler and level
108 self
.write_lock
= None
109 self
.vims_assigned
= {}
113 self
.process_params_function_map
= {
114 "net": Ns
._process
_net
_params
,
115 "image": Ns
._process
_image
_params
,
116 "flavor": Ns
._process
_flavor
_params
,
117 "vdu": Ns
._process
_vdu
_params
,
118 "affinity-or-anti-affinity-group": Ns
._process
_affinity
_group
_params
,
125 "affinity-or-anti-affinity-group": "affinity-or-anti-affinity-group",
128 def init_db(self
, target_version
):
131 def start(self
, config
):
133 Connect to database, filesystem storage, and messaging
134 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
135 :param config: Configuration of db, storage, etc
139 self
.config
["process_id"] = get_process_id() # used for HA identity
140 self
.logger
= logging
.getLogger("ro.ns")
142 # check right version of common
143 if versiontuple(common_version
) < versiontuple(min_common_version
):
145 "Not compatible osm/common version '{}'. Needed '{}' or higher".format(
146 common_version
, min_common_version
152 if config
["database"]["driver"] == "mongo":
153 self
.db
= dbmongo
.DbMongo()
154 self
.db
.db_connect(config
["database"])
155 elif config
["database"]["driver"] == "memory":
156 self
.db
= dbmemory
.DbMemory()
157 self
.db
.db_connect(config
["database"])
160 "Invalid configuration param '{}' at '[database]':'driver'".format(
161 config
["database"]["driver"]
166 if config
["storage"]["driver"] == "local":
167 self
.fs
= fslocal
.FsLocal()
168 self
.fs
.fs_connect(config
["storage"])
169 elif config
["storage"]["driver"] == "mongo":
170 self
.fs
= fsmongo
.FsMongo()
171 self
.fs
.fs_connect(config
["storage"])
172 elif config
["storage"]["driver"] is None:
176 "Invalid configuration param '{}' at '[storage]':'driver'".format(
177 config
["storage"]["driver"]
182 if config
["message"]["driver"] == "local":
183 self
.msg
= msglocal
.MsgLocal()
184 self
.msg
.connect(config
["message"])
185 elif config
["message"]["driver"] == "kafka":
186 self
.msg
= msgkafka
.MsgKafka()
187 self
.msg
.connect(config
["message"])
190 "Invalid configuration param '{}' at '[message]':'driver'".format(
191 config
["message"]["driver"]
195 # TODO load workers to deal with exising database tasks
197 self
.write_lock
= Lock()
198 except (DbException
, FsException
, MsgException
) as e
:
199 raise NsException(str(e
), http_code
=e
.http_code
)
201 def get_assigned_vims(self
):
202 return list(self
.vims_assigned
.keys())
207 self
.db
.db_disconnect()
210 self
.fs
.fs_disconnect()
213 self
.msg
.disconnect()
215 self
.write_lock
= None
216 except (DbException
, FsException
, MsgException
) as e
:
217 raise NsException(str(e
), http_code
=e
.http_code
)
219 for worker
in self
.workers
:
220 worker
.insert_task(("terminate",))
222 def _create_worker(self
):
224 Look for a worker thread in idle status. If not found it creates one unless the number of threads reach the
225 limit of 'server.ns_threads' configuration. If reached, it just assigns one existing thread
226 return the index of the assigned worker thread. Worker threads are storead at self.workers
228 # Look for a thread in idle status
232 for i
in range(len(self
.workers
))
233 if self
.workers
[i
] and self
.workers
[i
].idle
238 if worker_id
is not None:
239 # unset idle status to avoid race conditions
240 self
.workers
[worker_id
].idle
= False
242 worker_id
= len(self
.workers
)
244 if worker_id
< self
.config
["global"]["server.ns_threads"]:
245 # create a new worker
247 NsWorker(worker_id
, self
.config
, self
.plugins
, self
.db
)
249 self
.workers
[worker_id
].start()
251 # reached maximum number of threads, assign VIM to an existing one
252 worker_id
= self
.next_worker
253 self
.next_worker
= (self
.next_worker
+ 1) % self
.config
["global"][
259 def assign_vim(self
, target_id
):
260 with self
.write_lock
:
261 return self
._assign
_vim
(target_id
)
263 def _assign_vim(self
, target_id
):
264 if target_id
not in self
.vims_assigned
:
265 worker_id
= self
.vims_assigned
[target_id
] = self
._create
_worker
()
266 self
.workers
[worker_id
].insert_task(("load_vim", target_id
))
268 def reload_vim(self
, target_id
):
269 # send reload_vim to the thread working with this VIM and inform all that a VIM has been changed,
270 # this is because database VIM information is cached for threads working with SDN
271 with self
.write_lock
:
272 for worker
in self
.workers
:
273 if worker
and not worker
.idle
:
274 worker
.insert_task(("reload_vim", target_id
))
276 def unload_vim(self
, target_id
):
277 with self
.write_lock
:
278 return self
._unload
_vim
(target_id
)
280 def _unload_vim(self
, target_id
):
281 if target_id
in self
.vims_assigned
:
282 worker_id
= self
.vims_assigned
[target_id
]
283 self
.workers
[worker_id
].insert_task(("unload_vim", target_id
))
284 del self
.vims_assigned
[target_id
]
286 def check_vim(self
, target_id
):
287 with self
.write_lock
:
288 if target_id
in self
.vims_assigned
:
289 worker_id
= self
.vims_assigned
[target_id
]
291 worker_id
= self
._create
_worker
()
293 worker
= self
.workers
[worker_id
]
294 worker
.insert_task(("check_vim", target_id
))
296 def unload_unused_vims(self
):
297 with self
.write_lock
:
300 for target_id
in self
.vims_assigned
:
301 if not self
.db
.get_one(
304 "target_id": target_id
,
305 "tasks.status": ["SCHEDULED", "BUILD", "DONE", "FAILED"],
309 vims_to_unload
.append(target_id
)
311 for target_id
in vims_to_unload
:
312 self
._unload
_vim
(target_id
)
320 """This method reads cloud init from a file.
322 Note: Not used as cloud init content is provided in the http body.
325 db (Type[DbBase]): [description]
326 fs (Type[FsBase]): [description]
327 location (str): can be 'vnfr_id:file:file_name' or 'vnfr_id:vdu:vdu_idex'
330 NsException: [description]
331 NsException: [description]
336 vnfd_id
, _
, other
= location
.partition(":")
337 _type
, _
, name
= other
.partition(":")
338 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
341 base_folder
= vnfd
["_admin"]["storage"]
342 cloud_init_file
= "{}/{}/cloud_init/{}".format(
343 base_folder
["folder"], base_folder
["pkg-dir"], name
348 "Cannot read file '{}'. Filesystem not loaded, change configuration at storage.driver".format(
353 with fs
.file_open(cloud_init_file
, "r") as ci_file
:
354 cloud_init_content
= ci_file
.read()
356 cloud_init_content
= vnfd
["vdu"][int(name
)]["cloud-init"]
358 raise NsException("Mismatch descriptor for cloud init: {}".format(location
))
360 return cloud_init_content
364 cloud_init_content
: str,
365 params
: Dict
[str, Any
],
368 """Function that processes the cloud init to replace Jinja2 encoded parameters.
371 cloud_init_content (str): [description]
372 params (Dict[str, Any]): [description]
373 context (str): [description]
376 NsException: [description]
377 NsException: [description]
383 env
= Environment(undefined
=StrictUndefined
)
384 template
= env
.from_string(cloud_init_content
)
386 return template
.render(params
or {})
387 except UndefinedError
as e
:
389 "Variable '{}' defined at vnfd='{}' must be provided in the instantiation parameters"
390 "inside the 'additionalParamsForVnf' block".format(e
, context
)
392 except (TemplateError
, TemplateNotFound
) as e
:
394 "Error parsing Jinja2 to cloud-init content at vnfd='{}': {}".format(
399 def _create_db_ro_nsrs(self
, nsr_id
, now
):
401 key
= rsa
.generate_private_key(
402 backend
=crypto_default_backend(), public_exponent
=65537, key_size
=2048
404 private_key
= key
.private_bytes(
405 crypto_serialization
.Encoding
.PEM
,
406 crypto_serialization
.PrivateFormat
.PKCS8
,
407 crypto_serialization
.NoEncryption(),
409 public_key
= key
.public_key().public_bytes(
410 crypto_serialization
.Encoding
.OpenSSH
,
411 crypto_serialization
.PublicFormat
.OpenSSH
,
413 private_key
= private_key
.decode("utf8")
414 # Change first line because Paramiko needs a explicit start with 'BEGIN RSA PRIVATE KEY'
415 i
= private_key
.find("\n")
416 private_key
= "-----BEGIN RSA PRIVATE KEY-----" + private_key
[i
:]
417 public_key
= public_key
.decode("utf8")
418 except Exception as e
:
419 raise NsException("Cannot create ssh-keys: {}".format(e
))
421 schema_version
= "1.1"
422 private_key_encrypted
= self
.db
.encrypt(
423 private_key
, schema_version
=schema_version
, salt
=nsr_id
430 "schema_version": schema_version
,
432 "public_key": public_key
,
433 "private_key": private_key_encrypted
,
436 self
.db
.create("ro_nsrs", db_content
)
442 deployment_info
: Dict
[str, Any
],
447 target_record_id
: str,
448 extra_dict
: Dict
[str, Any
] = None,
450 """Function to create task dict from deployment information.
453 deployment_info (Dict[str, Any]): [description]
454 target_id (str): [description]
455 item (str): [description]
456 action (str): [description]
457 target_record (str): [description]
458 target_record_id (str): [description]
459 extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
462 Dict[str, Any]: [description]
465 "target_id": target_id
, # it will be removed before pushing at database
466 "action_id": deployment_info
.get("action_id"),
467 "nsr_id": deployment_info
.get("nsr_id"),
468 "task_id": f
"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
469 "status": "SCHEDULED",
472 "target_record": target_record
,
473 "target_record_id": target_record_id
,
477 task
.update(extra_dict
) # params, find_params, depends_on
479 deployment_info
["task_index"] = deployment_info
.get("task_index", 0) + 1
486 task
: Dict
[str, Any
],
488 """Function to create an RO task from task information.
491 target_id (str): [description]
492 task (Dict[str, Any]): [description]
495 Dict[str, Any]: [description]
499 _id
= task
.get("task_id")
504 "target_id": target_id
,
507 "created_items": None,
524 def _process_image_params(
525 target_image
: Dict
[str, Any
],
526 indata
: Dict
[str, Any
],
527 vim_info
: Dict
[str, Any
],
528 target_record_id
: str,
529 **kwargs
: Dict
[str, Any
],
531 """Function to process VDU image parameters.
534 target_image (Dict[str, Any]): [description]
535 indata (Dict[str, Any]): [description]
536 vim_info (Dict[str, Any]): [description]
537 target_record_id (str): [description]
540 Dict[str, Any]: [description]
544 if target_image
.get("image"):
545 find_params
["filter_dict"] = {"name": target_image
.get("image")}
547 if target_image
.get("vim_image_id"):
548 find_params
["filter_dict"] = {"id": target_image
.get("vim_image_id")}
550 if target_image
.get("image_checksum"):
551 find_params
["filter_dict"] = {
552 "checksum": target_image
.get("image_checksum")
555 return {"find_params": find_params
}
558 def _get_resource_allocation_params(
559 quota_descriptor
: Dict
[str, Any
],
561 """Read the quota_descriptor from vnfd and fetch the resource allocation properties from the
565 quota_descriptor (Dict[str, Any]): cpu/mem/vif/disk-io quota descriptor
568 Dict[str, Any]: quota params for limit, reserve, shares from the descriptor object
572 if quota_descriptor
.get("limit"):
573 quota
["limit"] = int(quota_descriptor
["limit"])
575 if quota_descriptor
.get("reserve"):
576 quota
["reserve"] = int(quota_descriptor
["reserve"])
578 if quota_descriptor
.get("shares"):
579 quota
["shares"] = int(quota_descriptor
["shares"])
584 def _process_guest_epa_quota_params(
585 guest_epa_quota
: Dict
[str, Any
],
588 """Function to extract the guest epa quota parameters.
591 guest_epa_quota (Dict[str, Any]): [description]
592 epa_vcpu_set (bool): [description]
595 Dict[str, Any]: [description]
599 if guest_epa_quota
.get("cpu-quota") and not epa_vcpu_set
:
600 cpuquota
= Ns
._get
_resource
_allocation
_params
(
601 guest_epa_quota
.get("cpu-quota")
605 result
["cpu-quota"] = cpuquota
607 if guest_epa_quota
.get("mem-quota"):
608 vduquota
= Ns
._get
_resource
_allocation
_params
(
609 guest_epa_quota
.get("mem-quota")
613 result
["mem-quota"] = vduquota
615 if guest_epa_quota
.get("disk-io-quota"):
616 diskioquota
= Ns
._get
_resource
_allocation
_params
(
617 guest_epa_quota
.get("disk-io-quota")
621 result
["disk-io-quota"] = diskioquota
623 if guest_epa_quota
.get("vif-quota"):
624 vifquota
= Ns
._get
_resource
_allocation
_params
(
625 guest_epa_quota
.get("vif-quota")
629 result
["vif-quota"] = vifquota
634 def _process_guest_epa_numa_params(
635 guest_epa_quota
: Dict
[str, Any
],
636 ) -> Tuple
[Dict
[str, Any
], bool]:
640 guest_epa_quota (Dict[str, Any]): [description]
643 Tuple[Dict[str, Any], bool]: [description]
648 if guest_epa_quota
.get("numa-node-policy"):
649 numa_node_policy
= guest_epa_quota
.get("numa-node-policy")
651 if numa_node_policy
.get("node"):
652 numa_node
= numa_node_policy
["node"][0]
654 if numa_node
.get("num-cores"):
655 numa
["cores"] = numa_node
["num-cores"]
658 paired_threads
= numa_node
.get("paired-threads", {})
659 if paired_threads
.get("num-paired-threads"):
660 numa
["paired-threads"] = int(
661 numa_node
["paired-threads"]["num-paired-threads"]
665 if paired_threads
.get("paired-thread-ids"):
666 numa
["paired-threads-id"] = []
668 for pair
in paired_threads
["paired-thread-ids"]:
669 numa
["paired-threads-id"].append(
671 str(pair
["thread-a"]),
672 str(pair
["thread-b"]),
676 if numa_node
.get("num-threads"):
677 numa
["threads"] = int(numa_node
["num-threads"])
680 if numa_node
.get("memory-mb"):
681 numa
["memory"] = max(int(int(numa_node
["memory-mb"]) / 1024), 1)
683 return numa
, epa_vcpu_set
686 def _process_guest_epa_cpu_pinning_params(
687 guest_epa_quota
: Dict
[str, Any
],
690 ) -> Tuple
[Dict
[str, Any
], bool]:
694 guest_epa_quota (Dict[str, Any]): [description]
695 vcpu_count (int): [description]
696 epa_vcpu_set (bool): [description]
699 Tuple[Dict[str, Any], bool]: [description]
702 local_epa_vcpu_set
= epa_vcpu_set
705 guest_epa_quota
.get("cpu-pinning-policy") == "DEDICATED"
710 if guest_epa_quota
.get("cpu-thread-pinning-policy") != "PREFER"
712 ] = max(vcpu_count
, 1)
713 local_epa_vcpu_set
= True
715 return numa
, local_epa_vcpu_set
718 def _process_epa_params(
719 target_flavor
: Dict
[str, Any
],
724 target_flavor (Dict[str, Any]): [description]
727 Dict[str, Any]: [description]
732 if target_flavor
.get("guest-epa"):
733 guest_epa
= target_flavor
["guest-epa"]
735 numa
, epa_vcpu_set
= Ns
._process
_guest
_epa
_numa
_params
(
736 guest_epa_quota
=guest_epa
739 if guest_epa
.get("mempage-size"):
740 extended
["mempage-size"] = guest_epa
.get("mempage-size")
742 tmp_numa
, epa_vcpu_set
= Ns
._process
_guest
_epa
_cpu
_pinning
_params
(
743 guest_epa_quota
=guest_epa
,
744 vcpu_count
=int(target_flavor
.get("vcpu-count", 1)),
745 epa_vcpu_set
=epa_vcpu_set
,
747 numa
.update(tmp_numa
)
750 Ns
._process
_guest
_epa
_quota
_params
(
751 guest_epa_quota
=guest_epa
,
752 epa_vcpu_set
=epa_vcpu_set
,
757 extended
["numas"] = [numa
]
762 def _process_flavor_params(
763 target_flavor
: Dict
[str, Any
],
764 indata
: Dict
[str, Any
],
765 vim_info
: Dict
[str, Any
],
766 target_record_id
: str,
767 **kwargs
: Dict
[str, Any
],
772 target_flavor (Dict[str, Any]): [description]
773 indata (Dict[str, Any]): [description]
774 vim_info (Dict[str, Any]): [description]
775 target_record_id (str): [description]
778 Dict[str, Any]: [description]
780 db
= kwargs
.get("db")
784 "disk": int(target_flavor
["storage-gb"]),
785 "ram": int(target_flavor
["memory-mb"]),
786 "vcpus": int(target_flavor
["vcpu-count"]),
789 for vnf
in indata
.get("vnf", []):
790 for vdur
in vnf
.get("vdur", []):
791 if vdur
.get("ns-flavor-id") == target_flavor
.get("id"):
794 if db
and isinstance(indata
.get("vnf"), list):
795 vnfd_id
= indata
.get("vnf")[0].get("vnfd-id")
796 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
797 # check if there is persistent root disk
798 for vdu
in vnfd
.get("vdu", ()):
799 if vdu
["name"] == target_vdur
.get("vdu-name"):
800 for vsd
in vnfd
.get("virtual-storage-desc", ()):
801 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
804 root_disk
.get("type-of-storage")
805 == "persistent-storage:persistent-storage"
807 flavor_data
["disk"] = 0
809 for storage
in target_vdur
.get("virtual-storages", []):
811 storage
.get("type-of-storage")
812 == "etsi-nfv-descriptors:ephemeral-storage"
814 flavor_data
["ephemeral"] = int(storage
.get("size-of-storage", 0))
815 elif storage
.get("type-of-storage") == "etsi-nfv-descriptors:swap-storage":
816 flavor_data
["swap"] = int(storage
.get("size-of-storage", 0))
818 extended
= Ns
._process
_epa
_params
(target_flavor
)
820 flavor_data
["extended"] = extended
822 extra_dict
= {"find_params": {"flavor_data": flavor_data
}}
823 flavor_data_name
= flavor_data
.copy()
824 flavor_data_name
["name"] = target_flavor
["name"]
825 extra_dict
["params"] = {"flavor_data": flavor_data_name
}
830 def _ip_profile_to_ro(
831 ip_profile
: Dict
[str, Any
],
836 ip_profile (Dict[str, Any]): [description]
839 Dict[str, Any]: [description]
846 if "v4" in ip_profile
.get("ip-version", "ipv4")
848 "subnet_address": ip_profile
.get("subnet-address"),
849 "gateway_address": ip_profile
.get("gateway-address"),
850 "dhcp_enabled": ip_profile
.get("dhcp-params", {}).get("enabled", False),
851 "dhcp_start_address": ip_profile
.get("dhcp-params", {}).get(
852 "start-address", None
854 "dhcp_count": ip_profile
.get("dhcp-params", {}).get("count", None),
857 if ip_profile
.get("dns-server"):
858 ro_ip_profile
["dns_address"] = ";".join(
859 [v
["address"] for v
in ip_profile
["dns-server"] if v
.get("address")]
862 if ip_profile
.get("security-group"):
863 ro_ip_profile
["security_group"] = ip_profile
["security-group"]
868 def _process_net_params(
869 target_vld
: Dict
[str, Any
],
870 indata
: Dict
[str, Any
],
871 vim_info
: Dict
[str, Any
],
872 target_record_id
: str,
873 **kwargs
: Dict
[str, Any
],
875 """Function to process network parameters.
878 target_vld (Dict[str, Any]): [description]
879 indata (Dict[str, Any]): [description]
880 vim_info (Dict[str, Any]): [description]
881 target_record_id (str): [description]
884 Dict[str, Any]: [description]
888 if vim_info
.get("sdn"):
889 # vnf_preffix = "vnfrs:{}".format(vnfr_id)
890 # ns_preffix = "nsrs:{}".format(nsr_id)
891 # remove the ending ".sdn
892 vld_target_record_id
, _
, _
= target_record_id
.rpartition(".")
893 extra_dict
["params"] = {
895 for k
in ("sdn-ports", "target_vim", "vlds", "type")
899 # TODO needed to add target_id in the dependency.
900 if vim_info
.get("target_vim"):
901 extra_dict
["depends_on"] = [
902 f
"{vim_info.get('target_vim')} {vld_target_record_id}"
907 if vim_info
.get("vim_network_name"):
908 extra_dict
["find_params"] = {
910 "name": vim_info
.get("vim_network_name"),
913 elif vim_info
.get("vim_network_id"):
914 extra_dict
["find_params"] = {
916 "id": vim_info
.get("vim_network_id"),
919 elif target_vld
.get("mgmt-network"):
920 extra_dict
["find_params"] = {
922 "name": target_vld
["id"],
926 extra_dict
["params"] = {
928 f
"{indata.get('name')[:16]}-{target_vld.get('name', target_vld.get('id'))[:16]}"
930 "ip_profile": Ns
._ip
_profile
_to
_ro
(vim_info
.get("ip_profile")),
931 "provider_network_profile": vim_info
.get("provider_network"),
934 if not target_vld
.get("underlay"):
935 extra_dict
["params"]["net_type"] = "bridge"
937 extra_dict
["params"]["net_type"] = (
938 "ptp" if target_vld
.get("type") == "ELINE" else "data"
944 def find_persistent_root_volumes(
947 vdu_instantiation_volumes_list
: list,
950 """Find the persistent root volumes and add them to the disk_list
951 by parsing the instantiation parameters
955 target_vdu: processed VDU
956 vdu_instantiation_volumes_list: instantiation parameters for the each VDU as a list
957 disk_list: to be filled up
960 disk_list: filled VDU list which is used for VDU creation
963 persistent_root_disk
= {}
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
:
980 vdu_volume
["vim-volume-id"]
981 and root_disk
["id"] == vdu_volume
["name"]
984 persistent_root_disk
[vsd
["id"]] = {
985 "vim_volume_id": vdu_volume
["vim-volume-id"],
986 "image_id": vdu
.get("sw-image-desc"),
989 disk_list
.append(persistent_root_disk
[vsd
["id"]])
991 # There can be only one root disk, when we find it, it will return the result
992 return disk_list
, persistent_root_disk
996 if root_disk
.get("size-of-storage"):
997 persistent_root_disk
[vsd
["id"]] = {
998 "image_id": vdu
.get("sw-image-desc"),
999 "size": root_disk
.get("size-of-storage"),
1002 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1003 return disk_list
, persistent_root_disk
1005 return disk_list
, persistent_root_disk
1008 def find_persistent_volumes(
1009 persistent_root_disk
: dict,
1011 vdu_instantiation_volumes_list
: list,
1014 """Find the ordinary persistent volumes and add them to the disk_list
1015 by parsing the instantiation parameters
1018 persistent_root_disk: persistent root disk dictionary
1019 target_vdu: processed VDU
1020 vdu_instantiation_volumes_list: instantiation parameters for the each VDU as a list
1021 disk_list: to be filled up
1024 disk_list: filled VDU list which is used for VDU creation
1027 # Find the ordinary volumes which are not added to the persistent_root_disk
1028 persistent_disk
= {}
1029 for disk
in target_vdu
.get("virtual-storages", {}):
1031 disk
.get("type-of-storage") == "persistent-storage:persistent-storage"
1032 and disk
["id"] not in persistent_root_disk
.keys()
1034 for vdu_volume
in vdu_instantiation_volumes_list
:
1036 if vdu_volume
["vim-volume-id"] and disk
["id"] == vdu_volume
["name"]:
1038 persistent_disk
[disk
["id"]] = {
1039 "vim_volume_id": vdu_volume
["vim-volume-id"],
1041 disk_list
.append(persistent_disk
[disk
["id"]])
1044 if disk
["id"] not in persistent_disk
.keys():
1045 persistent_disk
[disk
["id"]] = {
1046 "size": disk
.get("size-of-storage"),
1048 disk_list
.append(persistent_disk
[disk
["id"]])
1053 def _process_vdu_params(
1054 target_vdu
: Dict
[str, Any
],
1055 indata
: Dict
[str, Any
],
1056 vim_info
: Dict
[str, Any
],
1057 target_record_id
: str,
1058 **kwargs
: Dict
[str, Any
],
1059 ) -> Dict
[str, Any
]:
1060 """Function to process VDU parameters.
1063 target_vdu (Dict[str, Any]): [description]
1064 indata (Dict[str, Any]): [description]
1065 vim_info (Dict[str, Any]): [description]
1066 target_record_id (str): [description]
1069 Dict[str, Any]: [description]
1071 vnfr_id
= kwargs
.get("vnfr_id")
1072 nsr_id
= kwargs
.get("nsr_id")
1073 vnfr
= kwargs
.get("vnfr")
1074 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1075 tasks_by_target_record_id
= kwargs
.get("tasks_by_target_record_id")
1076 logger
= kwargs
.get("logger")
1077 db
= kwargs
.get("db")
1078 fs
= kwargs
.get("fs")
1079 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1081 vnf_preffix
= "vnfrs:{}".format(vnfr_id
)
1082 ns_preffix
= "nsrs:{}".format(nsr_id
)
1083 image_text
= ns_preffix
+ ":image." + target_vdu
["ns-image-id"]
1084 flavor_text
= ns_preffix
+ ":flavor." + target_vdu
["ns-flavor-id"]
1085 extra_dict
= {"depends_on": [image_text
, flavor_text
]}
1088 # If the position info is provided for all the interfaces, it will be sorted
1089 # according to position number ascendingly.
1090 if all(i
.get("position") for i
in target_vdu
["interfaces"]):
1091 sorted_interfaces
= sorted(
1092 target_vdu
["interfaces"],
1093 key
=lambda x
: (x
.get("position") is None, x
.get("position")),
1095 target_vdu
["interfaces"] = sorted_interfaces
1097 # If the position info is provided for some interfaces but not all of them, the interfaces
1098 # which has specific position numbers will be placed and others' positions will not be taken care.
1100 if any(i
.get("position") for i
in target_vdu
["interfaces"]):
1101 n
= len(target_vdu
["interfaces"])
1102 sorted_interfaces
= [-1] * n
1105 if target_vdu
["interfaces"][k
].get("position"):
1106 idx
= target_vdu
["interfaces"][k
]["position"]
1107 sorted_interfaces
[idx
- 1] = target_vdu
["interfaces"][k
]
1110 if not target_vdu
["interfaces"][m
].get("position"):
1111 idy
= sorted_interfaces
.index(-1)
1112 sorted_interfaces
[idy
] = target_vdu
["interfaces"][m
]
1115 target_vdu
["interfaces"] = sorted_interfaces
1117 # If the position info is not provided for the interfaces, interfaces will be attached
1118 # according to the order in the VNFD.
1119 for iface_index
, interface
in enumerate(target_vdu
["interfaces"]):
1120 if interface
.get("ns-vld-id"):
1121 net_text
= ns_preffix
+ ":vld." + interface
["ns-vld-id"]
1122 elif interface
.get("vnf-vld-id"):
1123 net_text
= vnf_preffix
+ ":vld." + interface
["vnf-vld-id"]
1126 "Interface {} from vdu {} not connected to any vld".format(
1127 iface_index
, target_vdu
["vdu-name"]
1131 continue # interface not connected to any vld
1133 extra_dict
["depends_on"].append(net_text
)
1135 if "port-security-enabled" in interface
:
1136 interface
["port_security"] = interface
.pop("port-security-enabled")
1138 if "port-security-disable-strategy" in interface
:
1139 interface
["port_security_disable_strategy"] = interface
.pop(
1140 "port-security-disable-strategy"
1145 for x
, v
in interface
.items()
1151 "port_security_disable_strategy",
1155 net_item
["net_id"] = "TASK-" + net_text
1156 net_item
["type"] = "virtual"
1158 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1159 # TODO floating_ip: True/False (or it can be None)
1160 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1161 # mark the net create task as type data
1163 tasks_by_target_record_id
,
1169 tasks_by_target_record_id
[net_text
]["extra_dict"]["params"][
1173 net_item
["use"] = "data"
1174 net_item
["model"] = interface
["type"]
1175 net_item
["type"] = interface
["type"]
1177 interface
.get("type") == "OM-MGMT"
1178 or interface
.get("mgmt-interface")
1179 or interface
.get("mgmt-vnf")
1181 net_item
["use"] = "mgmt"
1183 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1184 net_item
["use"] = "bridge"
1185 net_item
["model"] = interface
.get("type")
1187 if interface
.get("ip-address"):
1188 net_item
["ip_address"] = interface
["ip-address"]
1190 if interface
.get("mac-address"):
1191 net_item
["mac_address"] = interface
["mac-address"]
1193 net_list
.append(net_item
)
1195 if interface
.get("mgmt-vnf"):
1196 extra_dict
["mgmt_vnf_interface"] = iface_index
1197 elif interface
.get("mgmt-interface"):
1198 extra_dict
["mgmt_vdu_interface"] = iface_index
1203 if target_vdu
.get("cloud-init"):
1204 if target_vdu
["cloud-init"] not in vdu2cloud_init
:
1205 vdu2cloud_init
[target_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1208 location
=target_vdu
["cloud-init"],
1211 cloud_content_
= vdu2cloud_init
[target_vdu
["cloud-init"]]
1212 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1213 cloud_init_content
=cloud_content_
,
1214 params
=target_vdu
.get("additionalParams"),
1215 context
=target_vdu
["cloud-init"],
1218 if target_vdu
.get("boot-data-drive"):
1219 cloud_config
["boot-data-drive"] = target_vdu
.get("boot-data-drive")
1223 if target_vdu
.get("ssh-keys"):
1224 ssh_keys
+= target_vdu
.get("ssh-keys")
1226 if target_vdu
.get("ssh-access-required"):
1227 ssh_keys
.append(ro_nsr_public_key
)
1230 cloud_config
["key-pairs"] = ssh_keys
1232 persistent_root_disk
= {}
1233 vdu_instantiation_volumes_list
= []
1235 vnfd_id
= vnfr
["vnfd-id"]
1236 vnfd
= db
.get_one("vnfds", {"_id": vnfd_id
})
1238 if target_vdu
.get("additionalParams"):
1239 vdu_instantiation_volumes_list
= (
1240 target_vdu
.get("additionalParams").get("OSM").get("vdu_volumes")
1243 if vdu_instantiation_volumes_list
:
1245 # Find the root volumes and add to the disk_list
1246 (disk_list
, persistent_root_disk
,) = Ns
.find_persistent_root_volumes(
1247 vnfd
, target_vdu
, vdu_instantiation_volumes_list
, disk_list
1250 # Find the ordinary volumes which are not added to the persistent_root_disk
1251 # and put them to the disk list
1252 disk_list
= Ns
.find_persistent_volumes(
1253 persistent_root_disk
,
1255 vdu_instantiation_volumes_list
,
1261 # vdu_instantiation_volumes_list is empty
1262 for vdu
in vnfd
.get("vdu", ()):
1263 if vdu
["name"] == target_vdu
["vdu-name"]:
1264 for vsd
in vnfd
.get("virtual-storage-desc", ()):
1265 if vsd
.get("id") == vdu
.get("virtual-storage-desc", [[]])[0]:
1269 ) == "persistent-storage:persistent-storage" and root_disk
.get(
1272 persistent_root_disk
[vsd
["id"]] = {
1273 "image_id": vdu
.get("sw-image-desc"),
1274 "size": root_disk
["size-of-storage"],
1276 disk_list
.append(persistent_root_disk
[vsd
["id"]])
1278 if target_vdu
.get("virtual-storages"):
1279 for disk
in target_vdu
["virtual-storages"]:
1281 disk
.get("type-of-storage")
1282 == "persistent-storage:persistent-storage"
1283 and disk
["id"] not in persistent_root_disk
.keys()
1285 disk_list
.append({"size": disk
["size-of-storage"]})
1287 affinity_group_list
= []
1289 if target_vdu
.get("affinity-or-anti-affinity-group-id"):
1291 for affinity_group_id
in target_vdu
["affinity-or-anti-affinity-group-id"]:
1292 affinity_group_text
= (
1293 ns_preffix
+ ":affinity-or-anti-affinity-group." + affinity_group_id
1296 extra_dict
["depends_on"].append(affinity_group_text
)
1297 affinity_group
["affinity_group_id"] = "TASK-" + affinity_group_text
1298 affinity_group_list
.append(affinity_group
)
1300 extra_dict
["params"] = {
1301 "name": "{}-{}-{}-{}".format(
1302 indata
["name"][:16],
1303 vnfr
["member-vnf-index-ref"][:16],
1304 target_vdu
["vdu-name"][:32],
1305 target_vdu
.get("count-index") or 0,
1307 "description": target_vdu
["vdu-name"],
1309 "image_id": "TASK-" + image_text
,
1310 "flavor_id": "TASK-" + flavor_text
,
1311 "affinity_group_list": affinity_group_list
,
1312 "net_list": net_list
,
1313 "cloud_config": cloud_config
or None,
1314 "disk_list": disk_list
,
1315 "availability_zone_index": None, # TODO
1316 "availability_zone_list": None, # TODO
1322 def _process_affinity_group_params(
1323 target_affinity_group
: Dict
[str, Any
],
1324 indata
: Dict
[str, Any
],
1325 vim_info
: Dict
[str, Any
],
1326 target_record_id
: str,
1327 **kwargs
: Dict
[str, Any
],
1328 ) -> Dict
[str, Any
]:
1329 """Get affinity or anti-affinity group parameters.
1332 target_affinity_group (Dict[str, Any]): [description]
1333 indata (Dict[str, Any]): [description]
1334 vim_info (Dict[str, Any]): [description]
1335 target_record_id (str): [description]
1338 Dict[str, Any]: [description]
1342 affinity_group_data
= {
1343 "name": target_affinity_group
["name"],
1344 "type": target_affinity_group
["type"],
1345 "scope": target_affinity_group
["scope"],
1348 if target_affinity_group
.get("vim-affinity-group-id"):
1349 affinity_group_data
["vim-affinity-group-id"] = target_affinity_group
[
1350 "vim-affinity-group-id"
1353 extra_dict
["params"] = {
1354 "affinity_group_data": affinity_group_data
,
1360 def _process_recreate_vdu_params(
1361 existing_vdu
: Dict
[str, Any
],
1362 db_nsr
: Dict
[str, Any
],
1363 vim_info
: Dict
[str, Any
],
1364 target_record_id
: str,
1366 **kwargs
: Dict
[str, Any
],
1367 ) -> Dict
[str, Any
]:
1368 """Function to process VDU parameters to recreate.
1371 existing_vdu (Dict[str, Any]): [description]
1372 db_nsr (Dict[str, Any]): [description]
1373 vim_info (Dict[str, Any]): [description]
1374 target_record_id (str): [description]
1375 target_id (str): [description]
1378 Dict[str, Any]: [description]
1380 vnfr
= kwargs
.get("vnfr")
1381 vdu2cloud_init
= kwargs
.get("vdu2cloud_init")
1382 # logger = kwargs.get("logger")
1383 db
= kwargs
.get("db")
1384 fs
= kwargs
.get("fs")
1385 ro_nsr_public_key
= kwargs
.get("ro_nsr_public_key")
1391 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
1392 if vim_details_text
:
1393 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
1395 for iface_index
, interface
in enumerate(existing_vdu
["interfaces"]):
1397 if "port-security-enabled" in interface
:
1398 interface
["port_security"] = interface
.pop("port-security-enabled")
1400 if "port-security-disable-strategy" in interface
:
1401 interface
["port_security_disable_strategy"] = interface
.pop(
1402 "port-security-disable-strategy"
1407 for x
, v
in interface
.items()
1413 "port_security_disable_strategy",
1417 existing_ifaces
= existing_vdu
["vim_info"][target_id
].get(
1418 "interfaces_backup", []
1423 for i
in existing_ifaces
1424 if i
["ip_address"] == interface
["ip-address"]
1429 net_item
["net_id"] = net_id
1430 net_item
["type"] = "virtual"
1432 # TODO mac_address: used for SR-IOV ifaces #TODO for other types
1433 # TODO floating_ip: True/False (or it can be None)
1434 if interface
.get("type") in ("SR-IOV", "PCI-PASSTHROUGH"):
1435 net_item
["use"] = "data"
1436 net_item
["model"] = interface
["type"]
1437 net_item
["type"] = interface
["type"]
1439 interface
.get("type") == "OM-MGMT"
1440 or interface
.get("mgmt-interface")
1441 or interface
.get("mgmt-vnf")
1443 net_item
["use"] = "mgmt"
1445 # if interface.get("type") in ("VIRTIO", "E1000", "PARAVIRT"):
1446 net_item
["use"] = "bridge"
1447 net_item
["model"] = interface
.get("type")
1449 if interface
.get("ip-address"):
1450 net_item
["ip_address"] = interface
["ip-address"]
1452 if interface
.get("mac-address"):
1453 net_item
["mac_address"] = interface
["mac-address"]
1455 net_list
.append(net_item
)
1457 if interface
.get("mgmt-vnf"):
1458 extra_dict
["mgmt_vnf_interface"] = iface_index
1459 elif interface
.get("mgmt-interface"):
1460 extra_dict
["mgmt_vdu_interface"] = iface_index
1465 if existing_vdu
.get("cloud-init"):
1466 if existing_vdu
["cloud-init"] not in vdu2cloud_init
:
1467 vdu2cloud_init
[existing_vdu
["cloud-init"]] = Ns
._get
_cloud
_init
(
1470 location
=existing_vdu
["cloud-init"],
1473 cloud_content_
= vdu2cloud_init
[existing_vdu
["cloud-init"]]
1474 cloud_config
["user-data"] = Ns
._parse
_jinja
2(
1475 cloud_init_content
=cloud_content_
,
1476 params
=existing_vdu
.get("additionalParams"),
1477 context
=existing_vdu
["cloud-init"],
1480 if existing_vdu
.get("boot-data-drive"):
1481 cloud_config
["boot-data-drive"] = existing_vdu
.get("boot-data-drive")
1485 if existing_vdu
.get("ssh-keys"):
1486 ssh_keys
+= existing_vdu
.get("ssh-keys")
1488 if existing_vdu
.get("ssh-access-required"):
1489 ssh_keys
.append(ro_nsr_public_key
)
1492 cloud_config
["key-pairs"] = ssh_keys
1495 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
1496 disk_list
.append({"vim_id": vol_id
["id"]})
1498 affinity_group_list
= []
1500 if existing_vdu
.get("affinity-or-anti-affinity-group-id"):
1502 for affinity_group_id
in existing_vdu
["affinity-or-anti-affinity-group-id"]:
1503 for group
in db_nsr
.get("affinity-or-anti-affinity-group"):
1505 group
["id"] == affinity_group_id
1506 and group
["vim_info"][target_id
].get("vim_id", None) is not None
1508 affinity_group
["affinity_group_id"] = group
["vim_info"][
1510 ].get("vim_id", None)
1511 affinity_group_list
.append(affinity_group
)
1513 extra_dict
["params"] = {
1514 "name": "{}-{}-{}-{}".format(
1515 db_nsr
["name"][:16],
1516 vnfr
["member-vnf-index-ref"][:16],
1517 existing_vdu
["vdu-name"][:32],
1518 existing_vdu
.get("count-index") or 0,
1520 "description": existing_vdu
["vdu-name"],
1522 "image_id": vim_details
["image"]["id"],
1523 "flavor_id": vim_details
["flavor"]["id"],
1524 "affinity_group_list": affinity_group_list
,
1525 "net_list": net_list
,
1526 "cloud_config": cloud_config
or None,
1527 "disk_list": disk_list
,
1528 "availability_zone_index": None, # TODO
1529 "availability_zone_list": None, # TODO
1534 def calculate_diff_items(
1541 tasks_by_target_record_id
,
1548 """Function that returns the incremental changes (creation, deletion)
1549 related to a specific item `item` to be done. This function should be
1550 called for NS instantiation, NS termination, NS update to add a new VNF
1551 or a new VLD, remove a VNF or VLD, etc.
1552 Item can be `net`, `flavor`, `image` or `vdu`.
1553 It takes a list of target items from indata (which came from the REST API)
1554 and compares with the existing items from db_ro_nsr, identifying the
1555 incremental changes to be done. During the comparison, it calls the method
1556 `process_params` (which was passed as parameter, and is particular for each
1560 indata (Dict[str, Any]): deployment info
1561 db_nsr: NSR record from DB
1562 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1563 db_nsr_update (Dict[str, Any]): NSR info to update in DB
1564 item (str): element to process (net, vdu...)
1565 tasks_by_target_record_id (Dict[str, Any]):
1566 [<target_record_id>, <task>]
1567 action_id (str): action id
1568 nsr_id (str): NSR id
1569 task_index (number): task index to add to task name
1570 vnfr_id (str): VNFR id
1571 vnfr (Dict[str, Any]): VNFR info
1574 List: list with the incremental changes (deletes, creates) for each item
1575 number: current task index
1583 process_params
= None
1584 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
1585 ro_nsr_public_key
= db_ro_nsr
["public_key"]
1587 # According to the type of item, the path, the target_list,
1588 # the existing_list and the method to process params are set
1589 db_path
= self
.db_path_map
[item
]
1590 process_params
= self
.process_params_function_map
[item
]
1591 if item
in ("net", "vdu"):
1592 # This case is specific for the NS VLD (not applied to VDU)
1594 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1595 target_list
= indata
.get("ns", []).get(db_path
, [])
1596 existing_list
= db_nsr
.get(db_path
, [])
1597 # This case is common for VNF VLDs and VNF VDUs
1599 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
1601 (vnf
for vnf
in indata
.get("vnf", ()) if vnf
["_id"] == vnfr_id
),
1604 target_list
= target_vnf
.get(db_path
, []) if target_vnf
else []
1605 existing_list
= vnfr
.get(db_path
, [])
1606 elif item
in ("image", "flavor", "affinity-or-anti-affinity-group"):
1607 db_record
= "nsrs:{}:{}".format(nsr_id
, db_path
)
1608 target_list
= indata
.get(item
, [])
1609 existing_list
= db_nsr
.get(item
, [])
1611 raise NsException("Item not supported: {}", item
)
1613 # ensure all the target_list elements has an "id". If not assign the index as id
1614 if target_list
is None:
1616 for target_index
, tl
in enumerate(target_list
):
1617 if tl
and not tl
.get("id"):
1618 tl
["id"] = str(target_index
)
1620 # step 1 items (networks,vdus,...) to be deleted/updated
1621 for item_index
, existing_item
in enumerate(existing_list
):
1623 (t
for t
in target_list
if t
["id"] == existing_item
["id"]),
1627 for target_vim
, existing_viminfo
in existing_item
.get(
1630 if existing_viminfo
is None:
1634 target_viminfo
= target_item
.get("vim_info", {}).get(target_vim
)
1636 target_viminfo
= None
1638 if target_viminfo
is None:
1640 self
._assign
_vim
(target_vim
)
1641 target_record_id
= "{}.{}".format(db_record
, existing_item
["id"])
1644 if target_vim
.startswith("sdn"):
1645 # item must be sdn-net instead of net if target_vim is a sdn
1647 target_record_id
+= ".sdn"
1650 "action_id": action_id
,
1652 "task_index": task_index
,
1657 "deployment_info": deployment_info
,
1658 "target_id": target_vim
,
1661 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
1662 "target_record_id": target_record_id
,
1667 # step 2 items (networks,vdus,...) to be created
1668 for target_item
in target_list
:
1671 for item_index
, existing_item
in enumerate(existing_list
):
1672 if existing_item
["id"] == target_item
["id"]:
1676 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
1677 existing_list
.append(target_item
)
1678 existing_item
= None
1680 for target_vim
, target_viminfo
in target_item
.get("vim_info", {}).items():
1681 existing_viminfo
= None
1684 existing_viminfo
= existing_item
.get("vim_info", {}).get(target_vim
)
1686 if existing_viminfo
is not None:
1689 target_record_id
= "{}.{}".format(db_record
, target_item
["id"])
1692 if target_vim
.startswith("sdn"):
1693 # item must be sdn-net instead of net if target_vim is a sdn
1695 target_record_id
+= ".sdn"
1698 self
.logger
.warning(
1699 "ns.calculate_diff_items target_item={}".format(target_item
)
1701 if process_params
== Ns
._process
_flavor
_params
:
1707 self
.logger
.warning(
1708 "calculate_diff_items for flavor kwargs={}".format(kwargs
)
1711 if process_params
== Ns
._process
_vdu
_params
:
1712 self
.logger
.warning(
1713 "calculate_diff_items self.fs={}".format(self
.fs
)
1720 "vdu2cloud_init": vdu2cloud_init
,
1721 "tasks_by_target_record_id": tasks_by_target_record_id
,
1722 "logger": self
.logger
,
1725 "ro_nsr_public_key": ro_nsr_public_key
,
1728 self
.logger
.warning("calculate_diff_items kwargs={}".format(kwargs
))
1730 extra_dict
= process_params(
1737 self
._assign
_vim
(target_vim
)
1740 "action_id": action_id
,
1742 "task_index": task_index
,
1746 "deployment_info": deployment_info
,
1747 "target_id": target_vim
,
1750 "target_record": f
"{db_record}.{item_index}.vim_info.{target_vim}",
1751 "target_record_id": target_record_id
,
1752 "extra_dict": extra_dict
,
1753 "common_id": target_item
.get("common_id", None),
1755 diff_items
.append(new_item
)
1756 tasks_by_target_record_id
[target_record_id
] = new_item
1759 db_nsr_update
[db_path
+ ".{}".format(item_index
)] = target_item
1761 return diff_items
, task_index
1763 def calculate_all_differences_to_deploy(
1773 tasks_by_target_record_id
,
1775 """This method calculates the ordered list of items (`changes_list`)
1776 to be created and deleted.
1779 indata (Dict[str, Any]): deployment info
1780 nsr_id (str): NSR id
1781 db_nsr: NSR record from DB
1782 db_vnfrs: VNFRS record from DB
1783 db_ro_nsr (Dict[str, Any]): record from "ro_nsrs"
1784 db_nsr_update (Dict[str, Any]): NSR info to update in DB
1785 db_vnfrs_update (Dict[str, Any]): VNFRS info to update in DB
1786 action_id (str): action id
1787 tasks_by_target_record_id (Dict[str, Any]):
1788 [<target_record_id>, <task>]
1791 List: ordered list of items to be created and deleted.
1795 # set list with diffs:
1798 # NS vld, image and flavor
1799 for item
in ["net", "image", "flavor", "affinity-or-anti-affinity-group"]:
1800 self
.logger
.debug("process NS={} {}".format(nsr_id
, item
))
1801 diff_items
, task_index
= self
.calculate_diff_items(
1804 db_ro_nsr
=db_ro_nsr
,
1805 db_nsr_update
=db_nsr_update
,
1807 tasks_by_target_record_id
=tasks_by_target_record_id
,
1808 action_id
=action_id
,
1810 task_index
=task_index
,
1813 changes_list
+= diff_items
1816 for vnfr_id
, vnfr
in db_vnfrs
.items():
1817 # vnfr_id need to be set as global variable for among others nested method _process_vdu_params
1818 for item
in ["net", "vdu"]:
1819 self
.logger
.debug("process VNF={} {}".format(vnfr_id
, item
))
1820 diff_items
, task_index
= self
.calculate_diff_items(
1823 db_ro_nsr
=db_ro_nsr
,
1824 db_nsr_update
=db_vnfrs_update
[vnfr
["_id"]],
1826 tasks_by_target_record_id
=tasks_by_target_record_id
,
1827 action_id
=action_id
,
1829 task_index
=task_index
,
1833 changes_list
+= diff_items
1837 def define_all_tasks(
1841 tasks_by_target_record_id
,
1843 """Function to create all the task structures obtanied from
1844 the method calculate_all_differences_to_deploy
1847 changes_list (List): ordered list of items to be created or deleted
1848 db_new_tasks (List): tasks list to be created
1849 action_id (str): action id
1850 tasks_by_target_record_id (Dict[str, Any]):
1851 [<target_record_id>, <task>]
1855 for change
in changes_list
:
1856 task
= Ns
._create
_task
(
1857 deployment_info
=change
["deployment_info"],
1858 target_id
=change
["target_id"],
1859 item
=change
["item"],
1860 action
=change
["action"],
1861 target_record
=change
["target_record"],
1862 target_record_id
=change
["target_record_id"],
1863 extra_dict
=change
.get("extra_dict", None),
1866 self
.logger
.warning("ns.define_all_tasks task={}".format(task
))
1867 tasks_by_target_record_id
[change
["target_record_id"]] = task
1868 db_new_tasks
.append(task
)
1870 if change
.get("common_id"):
1871 task
["common_id"] = change
["common_id"]
1873 def upload_all_tasks(
1878 """Function to save all tasks in the common DB
1881 db_new_tasks (List): tasks list to be created
1882 now (time): current time
1886 nb_ro_tasks
= 0 # for logging
1888 for db_task
in db_new_tasks
:
1889 target_id
= db_task
.pop("target_id")
1890 common_id
= db_task
.get("common_id")
1892 # Do not chek tasks with vim_status DELETED
1893 # because in manual heealing there are two tasks for the same vdur:
1894 # one with vim_status deleted and the other one with the actual VM status.
1900 "target_id": target_id
,
1901 "tasks.common_id": common_id
,
1902 "vim_info.vim_status.ne": "DELETED",
1904 update_dict
={"to_check_at": now
, "modified_at": now
},
1905 push
={"tasks": db_task
},
1906 fail_on_empty
=False,
1910 if not self
.db
.set_one(
1913 "target_id": target_id
,
1914 "tasks.target_record": db_task
["target_record"],
1915 "vim_info.vim_status.ne": "DELETED",
1917 update_dict
={"to_check_at": now
, "modified_at": now
},
1918 push
={"tasks": db_task
},
1919 fail_on_empty
=False,
1922 self
.logger
.debug("Updating database, Creating ro_tasks")
1923 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
1925 self
.db
.create("ro_tasks", db_ro_task
)
1928 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
1929 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
1933 def upload_recreate_tasks(
1938 """Function to save recreate tasks in the common DB
1941 db_new_tasks (List): tasks list to be created
1942 now (time): current time
1946 nb_ro_tasks
= 0 # for logging
1948 for db_task
in db_new_tasks
:
1949 target_id
= db_task
.pop("target_id")
1950 self
.logger
.warning("target_id={} db_task={}".format(target_id
, db_task
))
1952 action
= db_task
.get("action", None)
1955 self
.logger
.debug("Updating database, Creating ro_tasks")
1956 db_ro_task
= Ns
._create
_ro
_task
(target_id
, db_task
)
1958 # If DELETE task: the associated created items should be removed
1959 # (except persistent volumes):
1960 if action
== "DELETE":
1961 db_ro_task
["vim_info"]["created"] = True
1962 db_ro_task
["vim_info"]["created_items"] = db_task
.get(
1965 db_ro_task
["vim_info"]["volumes_to_hold"] = db_task
.get(
1966 "volumes_to_hold", []
1968 db_ro_task
["vim_info"]["vim_id"] = db_task
.get("vim_id", None)
1971 self
.logger
.warning("upload_all_tasks db_ro_task={}".format(db_ro_task
))
1972 self
.db
.create("ro_tasks", db_ro_task
)
1975 "Created {} ro_tasks; {} tasks - db_new_tasks={}".format(
1976 nb_ro_tasks
, len(db_new_tasks
), db_new_tasks
1980 def _prepare_created_items_for_healing(
1986 # Get created_items from ro_task
1987 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
1988 for ro_task
in ro_tasks
:
1989 for task
in ro_task
["tasks"]:
1991 task
["target_record"] == target_record
1992 and task
["action"] == "CREATE"
1993 and ro_task
["vim_info"]["created_items"]
1995 created_items
= ro_task
["vim_info"]["created_items"]
1998 return created_items
2000 def _prepare_persistent_volumes_for_healing(
2005 # The associated volumes of the VM shouldn't be removed
2008 vim_details_text
= existing_vdu
["vim_info"][target_id
].get("vim_details", None)
2009 if vim_details_text
:
2010 vim_details
= yaml
.safe_load(f
"{vim_details_text}")
2012 for vol_id
in vim_details
.get("os-extended-volumes:volumes_attached", []):
2013 volumes_list
.append(vol_id
["id"])
2017 def prepare_changes_to_recreate(
2025 tasks_by_target_record_id
,
2027 """This method will obtain an ordered list of items (`changes_list`)
2028 to be created and deleted to meet the recreate request.
2032 "ns.prepare_changes_to_recreate nsr_id={} indata={}".format(nsr_id
, indata
)
2036 # set list with diffs:
2038 db_path
= self
.db_path_map
["vdu"]
2039 target_list
= indata
.get("healVnfData", {})
2040 vdu2cloud_init
= indata
.get("cloud_init_content") or {}
2041 ro_nsr_public_key
= db_ro_nsr
["public_key"]
2043 # Check each VNF of the target
2044 for target_vnf
in target_list
:
2045 # Find this VNF in the list from DB
2046 vnfr_id
= target_vnf
.get("vnfInstanceId", None)
2048 existing_vnf
= db_vnfrs
.get(vnfr_id
)
2049 db_record
= "vnfrs:{}:{}".format(vnfr_id
, db_path
)
2050 # vim_account_id = existing_vnf.get("vim-account-id", "")
2052 # Check each VDU of this VNF
2053 for target_vdu
in target_vnf
["additionalParams"].get("vdu", None):
2054 vdu_name
= target_vdu
.get("vdu-id", None)
2055 # For multi instance VDU count-index is mandatory
2056 # For single session VDU count-indes is 0
2057 count_index
= target_vdu
.get("count-index", 0)
2059 existing_instance
= None
2060 for instance
in existing_vnf
.get("vdur", None):
2062 instance
["vdu-name"] == vdu_name
2063 and instance
["count-index"] == count_index
2065 existing_instance
= instance
2070 target_record_id
= "{}.{}".format(db_record
, existing_instance
["id"])
2072 # The target VIM is the one already existing in DB to recreate
2073 for target_vim
, target_viminfo
in existing_instance
.get(
2076 # step 1 vdu to be deleted
2077 self
._assign
_vim
(target_vim
)
2079 "action_id": action_id
,
2081 "task_index": task_index
,
2084 target_record
= f
"{db_record}.{item_index}.vim_info.{target_vim}"
2085 created_items
= self
._prepare
_created
_items
_for
_healing
(
2086 nsr_id
, target_record
2089 volumes_to_hold
= self
._prepare
_persistent
_volumes
_for
_healing
(
2090 target_vim
, existing_instance
2093 # Specific extra params for recreate tasks:
2095 "created_items": created_items
,
2096 "vim_id": existing_instance
["vim-id"],
2097 "volumes_to_hold": volumes_to_hold
,
2100 changes_list
.append(
2102 "deployment_info": deployment_info
,
2103 "target_id": target_vim
,
2106 "target_record": target_record
,
2107 "target_record_id": target_record_id
,
2108 "extra_dict": extra_dict
,
2111 delete_task_id
= f
"{action_id}:{task_index}"
2114 # step 2 vdu to be created
2120 "vnfr": existing_vnf
,
2121 "vdu2cloud_init": vdu2cloud_init
,
2122 "tasks_by_target_record_id": tasks_by_target_record_id
,
2123 "logger": self
.logger
,
2126 "ro_nsr_public_key": ro_nsr_public_key
,
2130 extra_dict
= self
._process
_recreate
_vdu
_params
(
2139 # The CREATE task depens on the DELETE task
2140 extra_dict
["depends_on"] = [delete_task_id
]
2142 # Add volumes created from created_items if any
2143 # Ports should be deleted with delete task and automatically created with create task
2145 for k
, v
in created_items
.items():
2147 k_item
, _
, k_id
= k
.partition(":")
2148 if k_item
== "volume":
2150 except Exception as e
:
2152 "Error evaluating created item {}: {}".format(k
, e
)
2154 extra_dict
["previous_created_volumes"] = volumes
2157 "action_id": action_id
,
2159 "task_index": task_index
,
2161 self
._assign
_vim
(target_vim
)
2164 "deployment_info": deployment_info
,
2165 "target_id": target_vim
,
2168 "target_record": target_record
,
2169 "target_record_id": target_record_id
,
2170 "extra_dict": extra_dict
,
2172 changes_list
.append(new_item
)
2173 tasks_by_target_record_id
[target_record_id
] = new_item
2178 def recreate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2179 self
.logger
.debug("ns.recreate nsr_id={} indata={}".format(nsr_id
, indata
))
2180 # TODO: validate_input(indata, recreate_schema)
2181 action_id
= indata
.get("action_id", str(uuid4()))
2182 # get current deployment
2183 db_vnfrs
= {} # vnf's info indexed by _id
2185 logging_text
= "Recreate nsr_id={} action_id={} indata={}".format(
2186 nsr_id
, action_id
, indata
2188 self
.logger
.debug(logging_text
+ "Enter")
2191 step
= "Getting ns and vnfr record from db"
2192 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2194 tasks_by_target_record_id
= {}
2195 # read from db: vnf's of this ns
2196 step
= "Getting vnfrs from db"
2197 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2198 self
.logger
.debug("ns.recreate: db_vnfrs_list={}".format(db_vnfrs_list
))
2200 if not db_vnfrs_list
:
2201 raise NsException("Cannot obtain associated VNF for ns")
2203 for vnfr
in db_vnfrs_list
:
2204 db_vnfrs
[vnfr
["_id"]] = vnfr
2207 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2208 self
.logger
.debug("ns.recreate: db_ro_nsr={}".format(db_ro_nsr
))
2211 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2213 with self
.write_lock
:
2215 step
= "process NS elements"
2216 changes_list
= self
.prepare_changes_to_recreate(
2221 db_ro_nsr
=db_ro_nsr
,
2222 action_id
=action_id
,
2223 tasks_by_target_record_id
=tasks_by_target_record_id
,
2226 self
.define_all_tasks(
2227 changes_list
=changes_list
,
2228 db_new_tasks
=db_new_tasks
,
2229 tasks_by_target_record_id
=tasks_by_target_record_id
,
2232 # Delete all ro_tasks registered for the targets vdurs (target_record)
2233 # If task of type CREATE exist then vim will try to get info form deleted VMs.
2234 # So remove all task related to target record.
2235 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2236 for change
in changes_list
:
2237 for ro_task
in ro_tasks
:
2238 for task
in ro_task
["tasks"]:
2239 if task
["target_record"] == change
["target_record"]:
2243 "_id": ro_task
["_id"],
2244 "modified_at": ro_task
["modified_at"],
2246 fail_on_empty
=False,
2249 step
= "Updating database, Appending tasks to ro_tasks"
2250 self
.upload_recreate_tasks(
2251 db_new_tasks
=db_new_tasks
,
2256 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2260 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2264 except Exception as e
:
2265 if isinstance(e
, (DbException
, NsException
)):
2267 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2270 e
= traceback_format_exc()
2271 self
.logger
.critical(
2272 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2276 raise NsException(e
)
2278 def deploy(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2279 self
.logger
.debug("ns.deploy nsr_id={} indata={}".format(nsr_id
, indata
))
2280 validate_input(indata
, deploy_schema
)
2281 action_id
= indata
.get("action_id", str(uuid4()))
2283 # get current deployment
2284 db_nsr_update
= {} # update operation on nsrs
2285 db_vnfrs_update
= {}
2286 db_vnfrs
= {} # vnf's info indexed by _id
2288 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2289 self
.logger
.debug(logging_text
+ "Enter")
2292 step
= "Getting ns and vnfr record from db"
2293 db_nsr
= self
.db
.get_one("nsrs", {"_id": nsr_id
})
2294 self
.logger
.debug("ns.deploy: db_nsr={}".format(db_nsr
))
2296 tasks_by_target_record_id
= {}
2297 # read from db: vnf's of this ns
2298 step
= "Getting vnfrs from db"
2299 db_vnfrs_list
= self
.db
.get_list("vnfrs", {"nsr-id-ref": nsr_id
})
2301 if not db_vnfrs_list
:
2302 raise NsException("Cannot obtain associated VNF for ns")
2304 for vnfr
in db_vnfrs_list
:
2305 db_vnfrs
[vnfr
["_id"]] = vnfr
2306 db_vnfrs_update
[vnfr
["_id"]] = {}
2307 self
.logger
.debug("ns.deploy db_vnfrs={}".format(db_vnfrs
))
2310 db_ro_nsr
= self
.db
.get_one("ro_nsrs", {"_id": nsr_id
}, fail_on_empty
=False)
2313 db_ro_nsr
= self
._create
_db
_ro
_nsrs
(nsr_id
, now
)
2315 # check that action_id is not in the list of actions. Suffixed with :index
2316 if action_id
in db_ro_nsr
["actions"]:
2320 new_action_id
= "{}:{}".format(action_id
, index
)
2322 if new_action_id
not in db_ro_nsr
["actions"]:
2323 action_id
= new_action_id
2326 + "Changing action_id in use to {}".format(action_id
)
2332 def _process_action(indata
):
2333 nonlocal db_new_tasks
2340 if indata
["action"]["action"] == "inject_ssh_key":
2341 key
= indata
["action"].get("key")
2342 user
= indata
["action"].get("user")
2343 password
= indata
["action"].get("password")
2345 for vnf
in indata
.get("vnf", ()):
2346 if vnf
["_id"] not in db_vnfrs
:
2347 raise NsException("Invalid vnf={}".format(vnf
["_id"]))
2349 db_vnfr
= db_vnfrs
[vnf
["_id"]]
2351 for target_vdu
in vnf
.get("vdur", ()):
2352 vdu_index
, vdur
= next(
2355 for i_v
in enumerate(db_vnfr
["vdur"])
2356 if i_v
[1]["id"] == target_vdu
["id"]
2363 "Invalid vdu vnf={}.{}".format(
2364 vnf
["_id"], target_vdu
["id"]
2368 target_vim
, vim_info
= next(
2369 k_v
for k_v
in vdur
["vim_info"].items()
2371 self
._assign
_vim
(target_vim
)
2372 target_record
= "vnfrs:{}:vdur.{}.ssh_keys".format(
2373 vnf
["_id"], vdu_index
2377 "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdur
["id"])
2380 "ip_address": vdur
.get("ip-address"),
2383 "password": password
,
2384 "private_key": db_ro_nsr
["private_key"],
2385 "salt": db_ro_nsr
["_id"],
2386 "schema_version": db_ro_nsr
["_admin"][
2393 "action_id": action_id
,
2395 "task_index": task_index
,
2398 task
= Ns
._create
_task
(
2399 deployment_info
=deployment_info
,
2400 target_id
=target_vim
,
2403 target_record
=target_record
,
2404 target_record_id
=None,
2405 extra_dict
=extra_dict
,
2408 task_index
= deployment_info
.get("task_index")
2410 db_new_tasks
.append(task
)
2412 with self
.write_lock
:
2413 if indata
.get("action"):
2414 _process_action(indata
)
2416 # compute network differences
2418 step
= "process NS elements"
2419 changes_list
= self
.calculate_all_differences_to_deploy(
2424 db_ro_nsr
=db_ro_nsr
,
2425 db_nsr_update
=db_nsr_update
,
2426 db_vnfrs_update
=db_vnfrs_update
,
2427 action_id
=action_id
,
2428 tasks_by_target_record_id
=tasks_by_target_record_id
,
2430 self
.define_all_tasks(
2431 changes_list
=changes_list
,
2432 db_new_tasks
=db_new_tasks
,
2433 tasks_by_target_record_id
=tasks_by_target_record_id
,
2436 step
= "Updating database, Appending tasks to ro_tasks"
2437 self
.upload_all_tasks(
2438 db_new_tasks
=db_new_tasks
,
2442 step
= "Updating database, nsrs"
2444 self
.db
.set_one("nsrs", {"_id": nsr_id
}, db_nsr_update
)
2446 for vnfr_id
, db_vnfr_update
in db_vnfrs_update
.items():
2448 step
= "Updating database, vnfrs={}".format(vnfr_id
)
2449 self
.db
.set_one("vnfrs", {"_id": vnfr_id
}, db_vnfr_update
)
2452 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2456 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2460 except Exception as e
:
2461 if isinstance(e
, (DbException
, NsException
)):
2463 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2466 e
= traceback_format_exc()
2467 self
.logger
.critical(
2468 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2472 raise NsException(e
)
2474 def delete(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2475 self
.logger
.debug("ns.delete version={} nsr_id={}".format(version
, nsr_id
))
2476 # self.db.del_list({"_id": ro_task["_id"], "tasks.nsr_id.ne": nsr_id})
2478 with self
.write_lock
:
2480 NsWorker
.delete_db_tasks(self
.db
, nsr_id
, None)
2481 except NsWorkerException
as e
:
2482 raise NsException(e
)
2484 return None, None, True
2486 def status(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2488 "ns.status version={} nsr_id={}, action_id={} indata={}".format(
2489 version
, nsr_id
, action_id
, indata
2495 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.action_id": action_id
})
2496 global_status
= "DONE"
2499 for ro_task
in ro_tasks
:
2500 for task
in ro_task
["tasks"]:
2501 if task
and task
["action_id"] == action_id
:
2502 task_list
.append(task
)
2505 if task
["status"] == "FAILED":
2506 global_status
= "FAILED"
2507 error_text
= "Error at {} {}: {}".format(
2508 task
["action"].lower(),
2510 ro_task
["vim_info"].get("vim_message") or "unknown",
2512 details
.append(error_text
)
2513 elif task
["status"] in ("SCHEDULED", "BUILD"):
2514 if global_status
!= "FAILED":
2515 global_status
= "BUILD"
2520 "status": global_status
,
2521 "details": ". ".join(details
)
2523 else "progress {}/{}".format(done
, total
),
2525 "action_id": action_id
,
2529 return return_data
, None, True
2531 def recreate_status(
2532 self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
2534 return self
.status(session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
)
2536 def cancel(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2538 "ns.cancel session={} indata={} version={} nsr_id={}, action_id={}".format(
2539 session
, indata
, version
, nsr_id
, action_id
2543 return None, None, True
2545 def rebuild_start_stop_task(
2556 self
._assign
_vim
(target_vim
)
2557 target_record
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_index
)
2558 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf_id
, vdu_id
)
2560 "action_id": action_id
,
2562 "task_index": task_index
,
2565 task
= Ns
._create
_task
(
2566 deployment_info
=deployment_info
,
2567 target_id
=target_vim
,
2570 target_record
=target_record
,
2571 target_record_id
=target_record_id
,
2572 extra_dict
=extra_dict
,
2576 def rebuild_start_stop(
2577 self
, session
, action_dict
, version
, nsr_id
, *args
, **kwargs
2582 action_id
= action_dict
.get("action_id", str(uuid4()))
2584 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2585 self
.logger
.debug(logging_text
+ "Enter")
2587 action
= list(action_dict
.keys())[0]
2588 task_dict
= action_dict
.get(action
)
2589 vim_vm_id
= action_dict
.get(action
).get("vim_vm_id")
2591 if action_dict
.get("stop"):
2595 step
= "lock the operation & do task creation"
2596 with self
.write_lock
:
2597 extra_dict
["params"] = {
2598 "vim_vm_id": vim_vm_id
,
2601 task
= self
.rebuild_start_stop_task(
2602 task_dict
["vdu_id"],
2603 task_dict
["vnf_id"],
2604 task_dict
["vdu_index"],
2608 task_dict
["target_vim"],
2611 db_new_tasks
.append(task
)
2612 step
= "upload Task to db"
2613 self
.upload_all_tasks(
2614 db_new_tasks
=db_new_tasks
,
2618 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2621 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2625 except Exception as e
:
2626 if isinstance(e
, (DbException
, NsException
)):
2628 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2631 e
= traceback_format_exc()
2632 self
.logger
.critical(
2633 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2636 raise NsException(e
)
2638 def get_deploy(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2639 nsrs
= self
.db
.get_list("nsrs", {})
2643 return_data
.append({"_id": ns
["_id"], "name": ns
["name"]})
2645 return return_data
, None, True
2647 def get_actions(self
, session
, indata
, version
, nsr_id
, action_id
, *args
, **kwargs
):
2648 ro_tasks
= self
.db
.get_list("ro_tasks", {"tasks.nsr_id": nsr_id
})
2651 for ro_task
in ro_tasks
:
2652 for task
in ro_task
["tasks"]:
2653 if task
["action_id"] not in return_data
:
2654 return_data
.append(task
["action_id"])
2656 return return_data
, None, True
2659 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
2661 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
2662 self
._assign
_vim
(target_vim
)
2663 target_record
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu_index
)
2664 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
2666 "action_id": action_id
,
2668 "task_index": task_index
,
2671 task
= Ns
._create
_task
(
2672 deployment_info
=deployment_info
,
2673 target_id
=target_vim
,
2676 target_record
=target_record
,
2677 target_record_id
=target_record_id
,
2678 extra_dict
=extra_dict
,
2683 def migrate(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2687 action_id
= indata
.get("action_id", str(uuid4()))
2689 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2690 self
.logger
.debug(logging_text
+ "Enter")
2692 vnf_instance_id
= indata
["vnfInstanceId"]
2693 step
= "Getting vnfrs from db"
2694 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
2695 vdu
= indata
.get("vdu")
2696 migrateToHost
= indata
.get("migrateToHost")
2699 with self
.write_lock
:
2701 vdu_id
= indata
["vdu"]["vduId"]
2702 vdu_count_index
= indata
["vdu"].get("vduCountIndex", 0)
2703 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
2705 vdu
["vdu-id-ref"] == vdu_id
2706 and vdu
["count-index"] == vdu_count_index
2708 extra_dict
["params"] = {
2709 "vim_vm_id": vdu
["vim-id"],
2710 "migrate_host": migrateToHost
,
2711 "vdu_vim_info": vdu
["vim_info"],
2713 step
= "Creating migration task for vdu:{}".format(vdu
)
2714 task
= self
.migrate_task(
2723 db_new_tasks
.append(task
)
2728 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
2729 extra_dict
["params"] = {
2730 "vim_vm_id": vdu
["vim-id"],
2731 "migrate_host": migrateToHost
,
2732 "vdu_vim_info": vdu
["vim_info"],
2734 step
= "Creating migration task for vdu:{}".format(vdu
)
2735 task
= self
.migrate_task(
2744 db_new_tasks
.append(task
)
2747 self
.upload_all_tasks(
2748 db_new_tasks
=db_new_tasks
,
2753 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2756 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2760 except Exception as e
:
2761 if isinstance(e
, (DbException
, NsException
)):
2763 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2766 e
= traceback_format_exc()
2767 self
.logger
.critical(
2768 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2771 raise NsException(e
)
2773 def verticalscale_task(
2774 self
, vdu
, vnf
, vdu_index
, action_id
, nsr_id
, task_index
, extra_dict
2776 target_vim
, vim_info
= next(k_v
for k_v
in vdu
["vim_info"].items())
2777 self
._assign
_vim
(target_vim
)
2778 target_record
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu_index
)
2779 target_record_id
= "vnfrs:{}:vdur.{}".format(vnf
["_id"], vdu
["id"])
2781 "action_id": action_id
,
2783 "task_index": task_index
,
2786 task
= Ns
._create
_task
(
2787 deployment_info
=deployment_info
,
2788 target_id
=target_vim
,
2789 item
="verticalscale",
2791 target_record
=target_record
,
2792 target_record_id
=target_record_id
,
2793 extra_dict
=extra_dict
,
2797 def verticalscale(self
, session
, indata
, version
, nsr_id
, *args
, **kwargs
):
2801 action_id
= indata
.get("action_id", str(uuid4()))
2803 logging_text
= "Task deploy nsr_id={} action_id={} ".format(nsr_id
, action_id
)
2804 self
.logger
.debug(logging_text
+ "Enter")
2806 VnfFlavorData
= indata
.get("changeVnfFlavorData")
2807 vnf_instance_id
= VnfFlavorData
["vnfInstanceId"]
2808 step
= "Getting vnfrs from db"
2809 db_vnfr
= self
.db
.get_one("vnfrs", {"_id": vnf_instance_id
})
2810 vduid
= VnfFlavorData
["additionalParams"]["vduid"]
2811 vduCountIndex
= VnfFlavorData
["additionalParams"]["vduCountIndex"]
2812 virtualMemory
= VnfFlavorData
["additionalParams"]["virtualMemory"]
2813 numVirtualCpu
= VnfFlavorData
["additionalParams"]["numVirtualCpu"]
2814 sizeOfStorage
= VnfFlavorData
["additionalParams"]["sizeOfStorage"]
2816 "name": vduid
+ "-flv",
2817 "ram": virtualMemory
,
2818 "vcpus": numVirtualCpu
,
2819 "disk": sizeOfStorage
,
2822 step
= "Creating Tasks for vertical scaling"
2823 with self
.write_lock
:
2824 for vdu_index
, vdu
in enumerate(db_vnfr
["vdur"]):
2826 vdu
["vdu-id-ref"] == vduid
2827 and vdu
["count-index"] == vduCountIndex
2829 extra_dict
["params"] = {
2830 "vim_vm_id": vdu
["vim-id"],
2831 "flavor_dict": flavor_dict
,
2833 task
= self
.verticalscale_task(
2842 db_new_tasks
.append(task
)
2845 self
.upload_all_tasks(
2846 db_new_tasks
=db_new_tasks
,
2850 logging_text
+ "Exit. Created {} tasks".format(len(db_new_tasks
))
2853 {"status": "ok", "nsr_id": nsr_id
, "action_id": action_id
},
2857 except Exception as e
:
2858 if isinstance(e
, (DbException
, NsException
)):
2860 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
)
2863 e
= traceback_format_exc()
2864 self
.logger
.critical(
2865 logging_text
+ "Exit Exception while '{}': {}".format(step
, e
),
2868 raise NsException(e
)